messaging

README

Messaging

The package contains the implementation of a front-end for messaging systems. Currently, only kafka is supported.

Documentation

Overview

Package messaging is the parent package for implementations of various messaging clients, e.g. Kafka.

Index

• Constants
• func ParseOpts(opts ...interface{}) (time.Duration, logging.Logger)
• func ToProtoMsgChan(ch chan ProtoMessage, opts ...interface{}) func(ProtoMessage)
• func ToProtoMsgErrChan(ch chan ProtoMessageErr, opts ...interface{}) func(ProtoMessageErr)
• type Mux
• type ProtoMessage
• type ProtoMessageErr
• type ProtoPublisher
• type ProtoWatcher
• type WithLoggerOpt
  • func WithLogger(logger logging.Logger) *WithLoggerOpt
• type WithTimeoutOpt
  • func WithTimeout(timeout time.Duration) *WithTimeoutOpt

Constants

const DefaultMsgTimeout = 2 * time.Second

DefaultMsgTimeout for delivery of notification

Functions

func ParseOpts

func ParseOpts(opts ...interface{}) (time.Duration, logging.Logger)

ParseOpts returns timeout and logger to be used based on the given options.

func ToProtoMsgChan

func ToProtoMsgChan(ch chan ProtoMessage, opts ...interface{}) func(ProtoMessage)

ToProtoMsgChan allows to receive messages through channel instead of callback.

func ToProtoMsgErrChan

func ToProtoMsgErrChan(ch chan ProtoMessageErr, opts ...interface{}) func(ProtoMessageErr)

ToProtoMsgErrChan allows to receive error messages through channel instead of callback.

Types

type Mux

type Mux interface {
	// Creates new Kafka synchronous publisher sending messages to given topic. Partitioner has to be set to 'hash' (default)
	// or 'random' scheme, otherwise an error is thrown
	NewSyncPublisher(topic string) (ProtoPublisher, error)

	// Creates new Kafka synchronous publisher sending messages to given topic and partition. Partitioner has to be
	// set to 'manual' scheme, otherwise an error is thrown
	NewSyncPublisherToPartition(topic string, partition int32) (ProtoPublisher, error)

	// Creates new Kafka asynchronous publisher sending messages to given topic. Partitioner has to be set to 'hash' (default)
	// or 'random' scheme, otherwise an error is thrown
	NewAsyncPublisher(topic string, successClb func(ProtoMessage), errorClb func(err ProtoMessageErr)) (ProtoPublisher, error)

	// Creates new Kafka asynchronous publisher sending messages to given topic and partition. Partitioner has to be
	// set to 'manual' scheme, otherwise an error is thrown
	NewAsyncPublisherToPartition(topic string, partition int32,
		successClb func(ProtoMessage), errorClb func(err ProtoMessageErr)) (ProtoPublisher, error)

	// Initializes new watcher which can start/stop watching on topic, eventually partition and offset
	NewWatcher(subscriberName string) ProtoWatcher
}

Mux defines API for the plugins that use access to kafka brokers.

type ProtoMessage

type ProtoMessage interface {
	keyval.ProtoKvPair
	GetTopic() string
	GetPartition() int32
	GetOffset() int64
}

ProtoMessage defines functions for inspection of a message receive from messaging system.

type ProtoMessageErr

type ProtoMessageErr interface {
	ProtoMessage
	Error() error
}

ProtoMessageErr represents a message that was not published successfully to a messaging system.

type ProtoPublisher

type ProtoPublisher interface {
	datasync.KeyProtoValWriter
}

ProtoPublisher allows to publish a message of type proto.Message into messaging system.

type ProtoWatcher

type ProtoWatcher interface {
	// Watch given topic. Returns error if 'manual' partitioner scheme is chosen
	Watch(msgCallback func(ProtoMessage), topics ...string) error

	// Stop watching on topic. Return error if topic is not subscribed
	StopWatch(topic string) error

	// Watch given topic, partition and offset. Offset is the oldest message index consumed, all previously written
	// messages are ignored. Manual partitioner must be set, otherwise error is thrown
	WatchPartition(msgCallback func(ProtoMessage), topic string, partition int32, offset int64) error

	// Stop watching on topic/partition/offset. Return error if such a combination is not subscribed
	StopWatchPartition(topic string, partition int32, offset int64) error
}

ProtoWatcher allows to subscribe for receiving of messages published to given topics.

type WithLoggerOpt

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

WithLoggerOpt defines a logger that logs if delivery of notification is unsuccessful.

func WithLogger

func WithLogger(logger logging.Logger) *WithLoggerOpt

WithLogger creates an option for ToChan function that specifies a logger to be used.

type WithTimeoutOpt

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

WithTimeoutOpt defines the maximum time that is attempted to deliver notification.

func WithTimeout

func WithTimeout(timeout time.Duration) *WithTimeoutOpt

WithTimeout creates an option for ToChan function that defines a timeout for notification delivery.