kafka

Documentation

Index

• Variables
• func CreateKafkaTransporter(options KafkaOptions) transit.Transport
• type KafkaOptions
• type KafkaTransporter
  • func (t *KafkaTransporter) Connect(registry moleculer.Registry) chan error
  • func (t *KafkaTransporter) Disconnect() chan error
  • func (t *KafkaTransporter) Publish(command, nodeID string, message moleculer.Payload)
  • func (t *KafkaTransporter) SetNodeID(nodeID string)
  • func (t *KafkaTransporter) SetPrefix(prefix string)
  • func (t *KafkaTransporter) SetSerializer(serializer serializer.Serializer)
  • func (t *KafkaTransporter) Subscribe(command, nodeID string, handler transit.TransportHandler)

Variables

var DefaultConfig = KafkaOptions{
	// contains filtered or unexported fields
}

Functions

func CreateKafkaTransporter

func CreateKafkaTransporter(options KafkaOptions) transit.Transport

Types

type KafkaOptions

type KafkaOptions struct {
	Url        string
	Addr       string
	Name       string
	Logger     *log.Entry
	Serializer serializer.Serializer
	// contains filtered or unexported fields
}

type KafkaTransporter

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

func (*KafkaTransporter) Connect

func (t *KafkaTransporter) Connect(registry moleculer.Registry) chan error

func (*KafkaTransporter) Disconnect

func (t *KafkaTransporter) Disconnect() chan error

func (*KafkaTransporter) Publish

func (t *KafkaTransporter) Publish(command, nodeID string, message moleculer.Payload)

func (*KafkaTransporter) SetNodeID

func (t *KafkaTransporter) SetNodeID(nodeID string)

func (*KafkaTransporter) SetPrefix

func (t *KafkaTransporter) SetPrefix(prefix string)

func (*KafkaTransporter) SetSerializer

func (t *KafkaTransporter) SetSerializer(serializer serializer.Serializer)

func (*KafkaTransporter) Subscribe

func (t *KafkaTransporter) Subscribe(command, nodeID string, handler transit.TransportHandler)