<commit_before>package mux

import (
	"fmt"
	"github.com/ligato/cn-infra/db/keyval"
	lg "github.com/ligato/cn-infra/logging/logrus"
	"github.com/ligato/cn-infra/messaging/kafka/client"
	"sync"
	"time"
)

// Multiplexer encapsulates clients to kafka cluster (syncProducer, asyncProducer, consumer).
// It allows to create multiple Connections that use multiplexer's clients for communication
// with kafka cluster. The aim of Multiplexer is to decrease the number of connections needed.
// The set of topics to be consumed by Connections needs to be selected before the underlying
// consumer in Multiplexer is started. Once the Multiplexer's consumer has been
// started new topics can not be added.
type Multiplexer struct {
	// consumer used by the Multiplexer
	consumer *client.Consumer
	// syncProducer used by the Multiplexer
	syncProducer *client.SyncProducer
	// asyncProducer used by the Multiplexer
	asyncProducer *client.AsyncProducer

	// name is used for identification of stored last consumed offset in kafka. This allows
	// to follow up messages after restart.
	name string

	// guards access to mapping and started flag
	rwlock sync.RWMutex

	// started denotes whether the multiplexer is dispatching the messages or accepting subscriptions to
	// consume a topic. Once the multiplexer is started, new subscription can not be added.
	started bool

	// Mapping provides the mapping of subscribed consumers organized by topics(key of the first map)
	// name of the consumer(key of the second map)
	mapping map[string]*map[string]chan *client.ConsumerMessage

	// factory that crates consumer used in the Multiplexer
	consumerFactory func(topics []string, groupId string) (*client.Consumer, error)
	closeCh         chan struct{}
}

// asyncMeta is auxiliary structure used by Multiplexer to distribute consumer messages
type asyncMeta struct {
	successChan chan *client.ProducerMessage
	errorChan   chan *client.ProducerError
	usersMeta   interface{}
}

// NewMultiplexer creates new instance of Kafka Multiplexer
func NewMultiplexer(consumerFactory ConsumerFactory, syncP *client.SyncProducer, asyncP *client.AsyncProducer, name string) *Multiplexer {
	cl := &Multiplexer{consumerFactory: consumerFactory,
		syncProducer:  syncP,
		asyncProducer: asyncP,
		name:          name,
		mapping:       map[string]*map[string]chan *client.ConsumerMessage{},
		closeCh:       make(chan struct{}),
	}

	go cl.watchAsyncProducerChannels()
	return cl
}

func (mux *Multiplexer) watchAsyncProducerChannels() {
	for {
		select {
		case err := <-mux.asyncProducer.Config.ErrorChan:
			log.Println("Failed to produce message", err.Err)
			errMsg := err.Msg

			if errMeta, ok := errMsg.Metadata.(*asyncMeta); ok && errMeta.errorChan != nil {
				err.Msg.Metadata = errMeta.usersMeta
				select {
				case errMeta.errorChan <- err:
				default:
					//case <-time.NewTimer(time.Second).C:
					log.Warn("Unable to send error notification")
				}
			}
		case success := <-mux.asyncProducer.Config.SuccessChan:

			if succMeta, ok := success.Metadata.(*asyncMeta); ok && succMeta.successChan != nil {
				success.Metadata = succMeta.usersMeta
				select {
				case succMeta.successChan <- success:
				default:
					//case <-time.NewTimer(time.Second).C:
					log.Warn("Unable to send success notification")
				}
			}
		case <-mux.asyncProducer.GetCloseChannel():
			log.Debug("Closing watch loop for async producer")
		}
	}
}

// Start should be called once all the Connections have been subscribed
// for topic consumption. An attempt to start consuming a topic after the multiplexer is started
// returns an error.
func (mux *Multiplexer) Start() error {
	mux.rwlock.Lock()
	defer mux.rwlock.Unlock()
	var err error

	if mux.started {
		return fmt.Errorf("Multiplexer has been started already")
	}

	// block further consumer consumers
	mux.started = true

	var topics []string

	for topic := range mux.mapping {
		topics = append(topics, topic)
	}

	log.WithFields(lg.Fields{"topics": topics}).Debug("Consuming started")

	mux.consumer, err = mux.consumerFactory(topics, mux.name)
	if err != nil {
		log.Error(err)
		return err
	}

	go mux.genericConsumer()

	return nil
}

// Close cleans up the resources used by the Multiplexer
func (mux *Multiplexer) Close() {
	close(mux.closeCh)
	mux.consumer.Close()
	mux.syncProducer.Close()
	mux.asyncProducer.Close()
}

// NewConnection creates instance of the Connection that will be provide access to shared Multiplexer's clients.
func (mux *Multiplexer) NewConnection(name string) *Connection {
	return &Connection{multiplexer: mux, name: name}
}

// NewProtoConnection creates instance of the ProtoConnection that will be provide access to shared Multiplexer's clients.
func (mux *Multiplexer) NewProtoConnection(name string, serializer keyval.Serializer) *ProtoConnection {
	return &ProtoConnection{multiplexer: mux, serializer: serializer, name: name}
}

func (mux *Multiplexer) propagateMessage(msg *client.ConsumerMessage) {
	mux.rwlock.RLock()
	defer mux.rwlock.RUnlock()

	if msg == nil {
		return
	}
	cons, found := mux.mapping[msg.Topic]

	// notify consumers
	if found {
		for _, ch := range *cons {<commit_msg>messaging/kafka: handle no topics to consume<commit_after>package mux

import (
	"fmt"
	"github.com/ligato/cn-infra/db/keyval"
	lg "github.com/ligato/cn-infra/logging/logrus"
	"github.com/ligato/cn-infra/messaging/kafka/client"
	"github.com/ligato/cn-infra/utils/safeclose"
	"sync"
	"time"
)

// Multiplexer encapsulates clients to kafka cluster (syncProducer, asyncProducer, consumer).
// It allows to create multiple Connections that use multiplexer's clients for communication
// with kafka cluster. The aim of Multiplexer is to decrease the number of connections needed.
// The set of topics to be consumed by Connections needs to be selected before the underlying
// consumer in Multiplexer is started. Once the Multiplexer's consumer has been
// started new topics can not be added.
type Multiplexer struct {
	// consumer used by the Multiplexer
	consumer *client.Consumer
	// syncProducer used by the Multiplexer
	syncProducer *client.SyncProducer
	// asyncProducer used by the Multiplexer
	asyncProducer *client.AsyncProducer

	// name is used for identification of stored last consumed offset in kafka. This allows
	// to follow up messages after restart.
	name string

	// guards access to mapping and started flag
	rwlock sync.RWMutex

	// started denotes whether the multiplexer is dispatching the messages or accepting subscriptions to
	// consume a topic. Once the multiplexer is started, new subscription can not be added.
	started bool

	// Mapping provides the mapping of subscribed consumers organized by topics(key of the first map)
	// name of the consumer(key of the second map)
	mapping map[string]*map[string]chan *client.ConsumerMessage

	// factory that crates consumer used in the Multiplexer
	consumerFactory func(topics []string, groupId string) (*client.Consumer, error)
	closeCh         chan struct{}
}

// asyncMeta is auxiliary structure used by Multiplexer to distribute consumer messages
type asyncMeta struct {
	successChan chan *client.ProducerMessage
	errorChan   chan *client.ProducerError
	usersMeta   interface{}
}

// NewMultiplexer creates new instance of Kafka Multiplexer
func NewMultiplexer(consumerFactory ConsumerFactory, syncP *client.SyncProducer, asyncP *client.AsyncProducer, name string) *Multiplexer {
	cl := &Multiplexer{consumerFactory: consumerFactory,
		syncProducer:  syncP,
		asyncProducer: asyncP,
		name:          name,
		mapping:       map[string]*map[string]chan *client.ConsumerMessage{},
		closeCh:       make(chan struct{}),
	}

	go cl.watchAsyncProducerChannels()
	return cl
}

func (mux *Multiplexer) watchAsyncProducerChannels() {
	for {
		select {
		case err := <-mux.asyncProducer.Config.ErrorChan:
			log.Println("Failed to produce message", err.Err)
			errMsg := err.Msg

			if errMeta, ok := errMsg.Metadata.(*asyncMeta); ok && errMeta.errorChan != nil {
				err.Msg.Metadata = errMeta.usersMeta
				select {
				case errMeta.errorChan <- err:
				default:
					//case <-time.NewTimer(time.Second).C:
					log.Warn("Unable to send error notification")
				}
			}
		case success := <-mux.asyncProducer.Config.SuccessChan:

			if succMeta, ok := success.Metadata.(*asyncMeta); ok && succMeta.successChan != nil {
				success.Metadata = succMeta.usersMeta
				select {
				case succMeta.successChan <- success:
				default:
					//case <-time.NewTimer(time.Second).C:
					log.Warn("Unable to send success notification")
				}
			}
		case <-mux.asyncProducer.GetCloseChannel():
			log.Debug("Closing watch loop for async producer")
		}
	}
}

// Start should be called once all the Connections have been subscribed
// for topic consumption. An attempt to start consuming a topic after the multiplexer is started
// returns an error.
func (mux *Multiplexer) Start() error {
	mux.rwlock.Lock()
	defer mux.rwlock.Unlock()
	var err error

	if mux.started {
		return fmt.Errorf("Multiplexer has been started already")
	}

	// block further consumer consumers
	mux.started = true

	var topics []string

	for topic := range mux.mapping {
		topics = append(topics, topic)
	}

	if len(topics) == 0 {
		log.Debug("No topics to be consumed")
		return nil
	}

	log.WithFields(lg.Fields{"topics": topics}).Debug("Consuming started")

	mux.consumer, err = mux.consumerFactory(topics, mux.name)
	if err != nil {
		log.Error(err)
		return err
	}

	go mux.genericConsumer()

	return nil
}

// Close cleans up the resources used by the Multiplexer
func (mux *Multiplexer) Close() {
	close(mux.closeCh)
	safeclose.Close(mux.consumer)
	safeclose.Close(mux.syncProducer)
	safeclose.Close(mux.asyncProducer)
}

// NewConnection creates instance of the Connection that will be provide access to shared Multiplexer's clients.
func (mux *Multiplexer) NewConnection(name string) *Connection {
	return &Connection{multiplexer: mux, name: name}
}

// NewProtoConnection creates instance of the ProtoConnection that will be provide access to shared Multiplexer's clients.
func (mux *Multiplexer) NewProtoConnection(name string, serializer keyval.Serializer) *ProtoConnection {
	return &ProtoConnection{multiplexer: mux, serializer: serializer, name: name}
}

func (mux *Multiplexer) propagateMessage(msg *client.ConsumerMessage) {
	mux.rwlock.RLock()
	defer mux.rwlock.RUnlock()

	if msg == nil {
		return
	}
	cons, found := mux.mapping[msg.Topic]

	// notify consumers
	if found {
		for _, ch := range *cons {
{'fraction_non_alphanumeric': 0.0940292831336756, 'fraction_numerical': 9.507510933637573e-05, 'mean_word_length': 4.096414728682171, 'pattern_counts': {'":': 2, '<': 17, '<?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': '3811619', 'n_tokens_mistral': 3407, 'n_tokens_neox': 3078, 'n_words': 1038}