kafkax

Documentation

Index

• func Disposable() error
• func Publish(e ...event.IntegrationEvent)
• func PublishToDelay(times time.Duration, e ...event.IntegrationEvent)
• func Subscribe[T event.IntegrationEvent, TH event.IntegrationEventHandler[T]]()
• func SubscribeDynamic[TH event.IDynamicIntegrationEventHandler](e string)
• func UnSubscribe[T event.IntegrationEvent]()
• func UnSubscribeDynamic(e string)
• type BusKafka
  • func NewEventBus(connection IDefaultKafkaConnection) BusKafka
  • func NewEventBusWithConfig(connection IDefaultKafkaConnection, config cqrs.EventBusConfig) BusKafka
• type Config
  • func NewConfig(adders []string, clientId string, username string, password string) Config
• type DefaultKafkaConnection
  • func (d *DefaultKafkaConnection) Close()
  • func (d *DefaultKafkaConnection) GetClient() sarama.Client
  • func (d *DefaultKafkaConnection) GetConfig() *sarama.Config
  • func (d *DefaultKafkaConnection) GetIsConnected() bool
  • func (d *DefaultKafkaConnection) TryConnect() bool
• type IDefaultKafkaConnection
  • func NewConnectionFactory(conf Config) IDefaultKafkaConnection

Functions

func Disposable

func Disposable() error

func Publish

func Publish(e ...event.IntegrationEvent)

func PublishToDelay

func PublishToDelay(times time.Duration, e ...event.IntegrationEvent)

func Subscribe

func Subscribe[T event.IntegrationEvent, TH event.IntegrationEventHandler[T]]()

func SubscribeDynamic

func SubscribeDynamic[TH event.IDynamicIntegrationEventHandler](e string)

func UnSubscribe

func UnSubscribe[T event.IntegrationEvent]()

func UnSubscribeDynamic

func UnSubscribeDynamic(e string)

Types

type BusKafka

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

func NewEventBus

func NewEventBus(connection IDefaultKafkaConnection) BusKafka

func NewEventBusWithConfig

func NewEventBusWithConfig(connection IDefaultKafkaConnection, config cqrs.EventBusConfig) BusKafka

type Config

type Config struct {
	Adders   []string
	Username string
	Password string
	ClientId string
}

func NewConfig

func NewConfig(adders []string, clientId string, username string, password string) Config

type DefaultKafkaConnection

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

func (*DefaultKafkaConnection) Close

func (d *DefaultKafkaConnection) Close()

func (*DefaultKafkaConnection) GetClient

func (d *DefaultKafkaConnection) GetClient() sarama.Client

func (*DefaultKafkaConnection) GetConfig

func (d *DefaultKafkaConnection) GetConfig() *sarama.Config

func (*DefaultKafkaConnection) GetIsConnected

func (d *DefaultKafkaConnection) GetIsConnected() bool

func (*DefaultKafkaConnection) TryConnect

func (d *DefaultKafkaConnection) TryConnect() bool

type IDefaultKafkaConnection

type IDefaultKafkaConnection interface {
	GetClient() sarama.Client
	GetConfig() *sarama.Config
	GetIsConnected() bool
	TryConnect() bool
	Close()
}

func NewConnectionFactory

func NewConnectionFactory(conf Config) IDefaultKafkaConnection