<commit_before>// Package mq provides an ability to integrate with message broker via AMQP in a declarative way.
package mq

import (
	"fmt"
	"net"
	"sync/atomic"
	"time"

	"github.com/NeowayLabs/wabbit"
	"github.com/NeowayLabs/wabbit/amqp"
	"github.com/NeowayLabs/wabbit/amqptest"
	"github.com/NeowayLabs/wabbit/utils"
	amqpDriver "github.com/streadway/amqp"
)

const (
	// Describes states during reconnect.
	statusReadyForReconnect int32 = 0
	statusReconnecting            = 1
)

// Used for creating connection to the fake AMQP server for tests.
var brokerIsMocked bool

type conn interface {
	Channel() (wabbit.Channel, error)
	Close() error
	NotifyClose(chan wabbit.Error) chan wabbit.Error
}

// MQ describes methods provided by message broker adapter.
type MQ interface {
	// Consumer returns consumer object by its name.
	Consumer(name string) (Consumer, error)
	// SetConsumerHandler allows you to set handler callback without getting consumer.
	SetConsumerHandler(name string, handler ConsumerHandler) error
	// AsyncProducer returns async producer. Should be used in most cases.
	AsyncProducer(name string) (AsyncProducer, error)
	// SyncProducer returns sync producer.
	SyncProducer(name string) (SyncProducer, error)
	// Error returns channel with all occurred errors.
	// Errors from sync producer won't be accessible.
	Error() <-chan error
	// Close stop all consumers and producers and close connection to broker.
	Close()
}

type mq struct {
	channel              wabbit.Channel
	config               Config
	connection           conn
	errorChannel         chan error
	internalErrorChannel chan error
	consumers            *consumersRegistry
	producers            *producersRegistry
	reconnectStatus      int32 // Defines whether client is trying to reconnect or not.
}

// New initializes AMQP connection to the message broker
// and returns adapter that provides an ability
// to get configured consumers and producers, read occurred errors and shutdown all workers.
func New(config Config) (MQ, error) {
	config.normalize()

	mq := &mq{
		config:               config,
		errorChannel:         make(chan error),
		internalErrorChannel: make(chan error),
		consumers:            newConsumersRegistry(len(config.Consumers)),
		producers:            newProducersRegistry(len(config.Producers)),
	}

	if err := mq.connect(); err != nil {
		return nil, err
	}

	go mq.errorHandler()

	return mq, mq.initialSetup()
}

// Set handler for consumer by its name. Returns false if consumer wasn't found.
// Can be called once for each consumer.
func (mq *mq) SetConsumerHandler(name string, handler ConsumerHandler) error {
	consumer, err := mq.Consumer(name)
	if err != nil {
		return err
	}

	consumer.Consume(handler)

	return nil
}

// Consumer returns a consumer by its name or error if consumer wasn't found.
func (mq *mq) Consumer(name string) (consumer Consumer, err error) {
	consumer, ok := mq.consumers.Get(name)
	if !ok {
		err = fmt.Errorf("consumer '%s' is not registered. Check your configuration", name)
	}

	return
}

// AsyncProducer returns an async producer by its name or error if producer wasn't found.
func (mq *mq) AsyncProducer(name string) (AsyncProducer, error) {
	item, exists := mq.producers.Get(name)
	producer, asserted := item.(*asyncProducer)

	if !exists || !asserted {
		return nil, fmt.Errorf("producer '%s' is not registered. Check your configuration", name)
	}

	return producer, nil
}

// SyncProducer returns a sync producer by its name or error if producer wasn't found.
func (mq *mq) SyncProducer(name string) (SyncProducer, error) {
	item, exists := mq.producers.Get(name)
	producer, asserted := item.(*syncProducer)

	if !exists || !asserted {
		return nil, fmt.Errorf("producer '%s' is not registered. Check your configuration", name)
	}

	return producer, nil
}

// Error provides an ability to access occurring errors.
func (mq *mq) Error() <-chan error {
	return mq.errorChannel
}

// Shutdown all workers and close connection to the message broker.
func (mq *mq) Close() {
	mq.stopProducersAndConsumers()

	if mq.channel != nil {
		mq.channel.Close()
	}

	if mq.connection != nil {
		mq.connection.Close()
	}
}

func (mq *mq) connect() error {
	connection, err := mq.createConnection()
	if err != nil {
		return err
	}

	channel, err := connection.Channel()
	if err != nil {
		connection.Close()

		return err
	}

	mq.connection = connection
	mq.channel = channel

	go mq.handleCloseEvent()

	return nil
}

func (mq *mq) createConnection() (conn conn, err error) {<commit_msg>Applied code style rules. (#31)<commit_after>// Package mq provides an ability to integrate with message broker via AMQP in a declarative way.
package mq

import (
	"fmt"
	"net"
	"sync/atomic"
	"time"

	"github.com/NeowayLabs/wabbit"
	"github.com/NeowayLabs/wabbit/amqp"
	"github.com/NeowayLabs/wabbit/amqptest"
	"github.com/NeowayLabs/wabbit/utils"
	amqpDriver "github.com/streadway/amqp"
)

const (
	// Describes states during reconnect.
	statusReadyForReconnect int32 = 0
	statusReconnecting      int32 = 1
)

// Used for creating connection to the fake AMQP server for tests.
var brokerIsMocked bool

type conn interface {
	Channel() (wabbit.Channel, error)
	Close() error
	NotifyClose(chan wabbit.Error) chan wabbit.Error
}

// MQ describes methods provided by message broker adapter.
type MQ interface {
	// Consumer returns consumer object by its name.
	Consumer(name string) (Consumer, error)
	// SetConsumerHandler allows you to set handler callback without getting consumer.
	SetConsumerHandler(name string, handler ConsumerHandler) error
	// AsyncProducer returns async producer. Should be used in most cases.
	AsyncProducer(name string) (AsyncProducer, error)
	// SyncProducer returns sync producer.
	SyncProducer(name string) (SyncProducer, error)
	// Error returns channel with all occurred errors.
	// Errors from sync producer won't be accessible.
	Error() <-chan error
	// Close stop all consumers and producers and close connection to broker.
	Close()
}

type mq struct {
	channel              wabbit.Channel
	config               Config
	connection           conn
	errorChannel         chan error
	internalErrorChannel chan error
	consumers            *consumersRegistry
	producers            *producersRegistry
	reconnectStatus      int32 // Defines whether client is trying to reconnect or not.
}

// New initializes AMQP connection to the message broker
// and returns adapter that provides an ability
// to get configured consumers and producers, read occurred errors and shutdown all workers.
func New(config Config) (MQ, error) {
	config.normalize()

	mq := &mq{
		config:               config,
		errorChannel:         make(chan error),
		internalErrorChannel: make(chan error),
		consumers:            newConsumersRegistry(len(config.Consumers)),
		producers:            newProducersRegistry(len(config.Producers)),
	}

	if err := mq.connect(); err != nil {
		return nil, err
	}

	go mq.errorHandler()

	return mq, mq.initialSetup()
}

// Set handler for consumer by its name. Returns false if consumer wasn't found.
// Can be called once for each consumer.
func (mq *mq) SetConsumerHandler(name string, handler ConsumerHandler) error {
	consumer, err := mq.Consumer(name)
	if err != nil {
		return err
	}

	consumer.Consume(handler)

	return nil
}

// Consumer returns a consumer by its name or error if consumer wasn't found.
func (mq *mq) Consumer(name string) (Consumer, error) {
	consumer, ok := mq.consumers.Get(name)
	if !ok {
		err := fmt.Errorf("consumer '%s' is not registered. Check your configuration", name)

		return nil, err
	}

	return consumer, nil
}

// AsyncProducer returns an async producer by its name or error if producer wasn't found.
func (mq *mq) AsyncProducer(name string) (AsyncProducer, error) {
	item, exists := mq.producers.Get(name)
	producer, asserted := item.(*asyncProducer)

	if !exists || !asserted {
		return nil, fmt.Errorf("producer '%s' is not registered. Check your configuration", name)
	}

	return producer, nil
}

// SyncProducer returns a sync producer by its name or error if producer wasn't found.
func (mq *mq) SyncProducer(name string) (SyncProducer, error) {
	item, exists := mq.producers.Get(name)
	producer, asserted := item.(*syncProducer)

	if !exists || !asserted {
		return nil, fmt.Errorf("producer '%s' is not registered. Check your configuration", name)
	}

	return producer, nil
}

// Error provides an ability to access occurring errors.
func (mq *mq) Error() <-chan error {
	return mq.errorChannel
}

// Shutdown all workers and close connection to the message broker.
func (mq *mq) Close() {
	mq.stopProducersAndConsumers()

	if mq.channel != nil {
		_ = mq.channel.Close()
	}

	if mq.connection != nil {
		_ = mq.connection.Close()
	}
}

func (mq *mq) connect() error {
	connection, err := mq.createConnection()
	if err != nil {
		return err
	}

	channel, err := connection.Channel()
	if err != nil {
		_ = connection.Close()

		return err
	}

	mq.connection = connection
	mq.channel = channel

	go mq.handleCloseEvent()

	return nil
}

func (mq *mq) createConnection() (conn, error) {
{'fraction_non_alphanumeric': 0.09769484083424808, 'fraction_numerical': 0.001756311745334797, 'mean_word_length': 3.6603580562659848, 'pattern_counts': {'":': 0, '<': 7, '<?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': '837452', 'n_tokens_mistral': 2887, 'n_tokens_neox': 2717, 'n_words': 980}