consumer

Documentation

Index

• type Consumer
• type KafkaComsumer
  • func NewKafkaConsumer(log logr.Logger, environmentManager *helper.ConfigManager, ...) (*KafkaComsumer, error)
  • func (c *KafkaComsumer) GetGenericBundleChan() chan *bundle.GenericBundle
  • func (c *KafkaComsumer) Register(msgID string, customBundleRegistration *bundle.CustomBundleRegistration)
  • func (c *KafkaComsumer) Start()
  • func (c *KafkaComsumer) Stop()
  • func (c *KafkaComsumer) SyncCustomBundle(customBundleRegistration *bundle.CustomBundleRegistration, payload []byte) error
• type SyncService
  • func NewSyncService(log logr.Logger, environmentManager *helper.ConfigManager, ...) (*SyncService, error)
  • func (s *SyncService) GetGenericBundleChan() chan *bundle.GenericBundle
  • func (s *SyncService) Register(msgID string, customBundleRegistration *bundle.CustomBundleRegistration)
  • func (s *SyncService) Start()
  • func (s *SyncService) Stop()
  • func (c *SyncService) SyncCustomBundle(customBundleRegistration *bundle.CustomBundleRegistration, payload []byte) error

Types

type Consumer

type Consumer interface {
	// Start starts the transport.
	Start()
	// Stop stops the transport.
	Stop()
	// Register registers a bundle ID to a CustomBundleRegistration. None-registered bundles are assumed to be
	// of type GenericBundle, and are handled by the GenericBundleSyncer.
	Register(msgID string, customBundleRegistration *bundle.CustomBundleRegistration)

	// provide the generic bundle for message producer
	GetGenericBundleChan() chan *bundle.GenericBundle
}

Transport is an interface for transport layer.

type KafkaComsumer

type KafkaComsumer struct {
	// contains filtered or unexported fields
}

Consumer abstracts hub-of-hubs-kafka-transport kafka-consumer's generic usage.

func NewKafkaConsumer

func NewKafkaConsumer(log logr.Logger, environmentManager *helper.ConfigManager, genericBundlesChan chan *bundle.GenericBundle) (*KafkaComsumer, error)

NewConsumer creates a new instance of Consumer.

func (*KafkaComsumer) GetGenericBundleChan

func (c *KafkaComsumer) GetGenericBundleChan() chan *bundle.GenericBundle

func (*KafkaComsumer) Register

func (c *KafkaComsumer) Register(msgID string, customBundleRegistration *bundle.CustomBundleRegistration)

Register function registers a bundle ID to a CustomBundleRegistration.

func (*KafkaComsumer) Start

func (c *KafkaComsumer) Start()

Start function starts the consumer.

func (*KafkaComsumer) Stop

func (c *KafkaComsumer) Stop()

Stop stops the consumer.

func (*KafkaComsumer) SyncCustomBundle

func (c *KafkaComsumer) SyncCustomBundle(customBundleRegistration *bundle.CustomBundleRegistration, payload []byte) error

SyncCustomBundle writes a custom bundle to its respective syncer channel.

type SyncService

type SyncService struct {
	// contains filtered or unexported fields
}

SyncService abstracts Sync Service client.

func NewSyncService

func NewSyncService(log logr.Logger, environmentManager *helper.ConfigManager, genericBundlesUpdatesChan chan *bundle.GenericBundle) (*SyncService, error)

NewSyncService creates a new instance of SyncService.

func (*SyncService) GetGenericBundleChan

func (s *SyncService) GetGenericBundleChan() chan *bundle.GenericBundle

func (*SyncService) Register

func (s *SyncService) Register(msgID string, customBundleRegistration *bundle.CustomBundleRegistration)

Register function registers a bundle ID to a CustomBundleRegistration.

func (*SyncService) Start

func (s *SyncService) Start()

Start function starts sync service.

func (*SyncService) Stop

func (s *SyncService) Stop()

Stop function stops sync service.

func (*SyncService) SyncCustomBundle

func (c *SyncService) SyncCustomBundle(customBundleRegistration *bundle.CustomBundleRegistration, payload []byte) error

SyncCustomBundle writes a custom bundle to its respective syncer channel.