<commit_before>package rabbit

import (
	"github.com/streadway/amqp"
)

var publishingConnection *amqp.Connection
var publishingChannel *amqp.Channel

// InitPublisher initializes the RabbitMQ Connection and Channel for Publishing messages.
func InitPublisher() {
	if publishingConnection == nil {
		publishingConnection = connect()
	}

	if publishingChannel == nil {
		publishingChannel = createChannel(publishingConnection)
	}
}

// InitPublisher reinitializes the RabbitMQ Connection and Channel for Publishing messages.
func ReInitPublisher() {
	publishingConnection = connect()
	publishingChannel = createChannel(publishingConnection)
}

// ConfirmPublish enables reliable mode for the publisher.
func ConfirmPublish(wait bool) error {
	InitPublisher()
	return publishingChannel.Confirm(wait)
}

// NotifyPublish registers a listener for reliable publishing.
func NotifyPublish(c chan amqp.Confirmation) chan amqp.Confirmation {
	InitPublisher()
	return publishingChannel.NotifyPublish(c)
}

// Publish pushes items on to a RabbitMQ Queue.
func Publish(message string, subscriber *Subscriber) {
	PublishBytes([]byte(message), subscriber)
}

// PublishBytes is the same as Publish but accepts a []byte instead of a string
func PublishBytes(message []byte, subscriber *Subscriber) {
	InitPublisher()

	publishingChannel.Publish(
		subscriber.Exchange,   // exchange
		subscriber.RoutingKey, // routing key
		false, // mandatory
		false, // immediate
		amqp.Publishing{
			ContentType:  "application/json",
			Body:         message,<commit_msg>return error from Publisher's functions<commit_after>package rabbit

import (
	"github.com/streadway/amqp"
)

var publishingConnection *amqp.Connection
var publishingChannel *amqp.Channel

// InitPublisher initializes the RabbitMQ Connection and Channel for Publishing messages.
func InitPublisher() {
	if publishingConnection == nil {
		publishingConnection = connect()
	}

	if publishingChannel == nil {
		publishingChannel = createChannel(publishingConnection)
	}
}

// InitPublisher reinitializes the RabbitMQ Connection and Channel for Publishing messages.
func ReInitPublisher() {
	publishingConnection = connect()
	publishingChannel = createChannel(publishingConnection)
}

// ConfirmPublish enables reliable mode for the publisher.
func ConfirmPublish(wait bool) error {
	InitPublisher()
	return publishingChannel.Confirm(wait)
}

// NotifyPublish registers a listener for reliable publishing.
func NotifyPublish(c chan amqp.Confirmation) chan amqp.Confirmation {
	InitPublisher()
	return publishingChannel.NotifyPublish(c)
}

// Publish pushes items on to a RabbitMQ Queue.
func Publish(message string, subscriber *Subscriber) error {
	return PublishBytes([]byte(message), subscriber)
}

// PublishBytes is the same as Publish but accepts a []byte instead of a string
func PublishBytes(message []byte, subscriber *Subscriber) error {
	InitPublisher()

	return publishingChannel.Publish(
		subscriber.Exchange,   // exchange
		subscriber.RoutingKey, // routing key
		false, // mandatory
		false, // immediate
		amqp.Publishing{
			ContentType:  "application/json",
			Body:         message,
{'fraction_non_alphanumeric': 0.08174300254452926, 'fraction_numerical': 0.0, 'mean_word_length': 5.083172147001934, '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': '1817689', 'n_tokens_mistral': 918, 'n_tokens_neox': 843, 'n_words': 263}