<commit_before>package topicmodifier

import (
	"encoding/json"
	"github.com/streadway/amqp"
	"koding/messaging/rabbitmq"
)

var (
	Done      chan error
	Consumer  *rabbitmq.Consumer
	Publisher *rabbitmq.Producer
)

type Status string

const (
	DELETE            Status = "delete"
	MERGE             Status = "merge"
	TRY_COUNT_LIMIT   int    = 3
	EXCHANGE_NAME            = "topicModifierExchange"
	WORKER_QUEUE_NAME        = "topicModifierWorkerQueue"
)

type TagModifierData struct {
	TagId    string `json:"tagId"`
	Status   Status `json:"status"`
	TryCount int    `json:"tryCount"`
	Error    string `json:"error"`
}

func createConnections() {
	initConsumer()
	initPublisher()
	initGraphPublisher()
}

func ConsumeMessage() {
	log.Info("Topic Modifier: Checking for message")
	createConnections()
	if err := Consumer.ConsumeMessage(messageConsumer); err != nil {
		log.Error(err.Error())
	}
	Shutdown()
}

var messageConsumer = func(delivery amqp.Delivery) {
	Completed = false
	modifierData := &TagModifierData{}
	if err := json.Unmarshal([]byte(delivery.Body), modifierData); err != nil {
		log.Error("Wrong Post Format", err, delivery)
		delivery.Ack(false)
		return<commit_msg>Moderation: ConsumeMessage is replaced by Get method<commit_after>package topicmodifier

import (
	"encoding/json"
	"github.com/streadway/amqp"
	"koding/messaging/rabbitmq"
)

var (
	Done      chan error
	Consumer  *rabbitmq.Consumer
	Publisher *rabbitmq.Producer
)

type Status string

const (
	DELETE            Status = "delete"
	MERGE             Status = "merge"
	TRY_COUNT_LIMIT   int    = 3
	EXCHANGE_NAME            = "topicModifierExchange"
	WORKER_QUEUE_NAME        = "topicModifierWorkerQueue"
)

type TagModifierData struct {
	TagId    string `json:"tagId"`
	Status   Status `json:"status"`
	TryCount int    `json:"tryCount"`
	Error    string `json:"error"`
}

func createConnections() {
	initConsumer()
	initPublisher()
	initGraphPublisher()
}

func ConsumeMessage() {
	log.Info("Topic Modifier: Checking for message")
	createConnections()
	if err := Consumer.Get(messageConsumer); err != nil {
		log.Error(err.Error())
	}
	Shutdown()
}

var messageConsumer = func(delivery amqp.Delivery) {
	Completed = false
	modifierData := &TagModifierData{}
	if err := json.Unmarshal([]byte(delivery.Body), modifierData); err != nil {
		log.Error("Wrong Post Format", err, delivery)
		delivery.Ack(false)
		return
{'fraction_non_alphanumeric': 0.11375, 'fraction_numerical': 0.0008333333333333334, 'mean_word_length': 4.012526096033403, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2270465', 'n_tokens_mistral': 820, 'n_tokens_neox': 788, 'n_words': 179}