transport

Documentation

Index

• func OptKafkaSendOnly() func(k *Kafka) error
• func OptRabbitMQSendOnly() func(mq *RabbitMQ) error
• type Kafka
  • func NewKafka(brokerAddrs []string, options ...func(k *Kafka) error) (*Kafka, error)
  • func (k *Kafka) Close()
  • func (k *Kafka) Publish(ctx context.Context, event *events.CloudEvent, topic string, tenant string) error
  • func (k *Kafka) Subscribe(ctx context.Context, topic string, handler events.Handler) (events.Subscription, error)
• type Noop
  • func NewNoop(out io.Writer) *Noop
  • func (t *Noop) Close()
  • func (t *Noop) Publish(ctx context.Context, event *events.CloudEvent, topic string, tenant string) error
  • func (t *Noop) Subscribe(ctx context.Context, topic string, handler events.Handler) (events.Subscription, error)
• type RabbitMQ
  • func NewRabbitMQ(url string, defaultExchangeName string, options ...func(mq *RabbitMQ) error) (mq *RabbitMQ, err error)
  • func (mq *RabbitMQ) Close()
  • func (mq *RabbitMQ) Publish(ctx context.Context, event *events.CloudEvent, topic string, tenant string) error
  • func (mq *RabbitMQ) Subscribe(ctx context.Context, topic string, handler events.Handler) (events.Subscription, error)

Functions

func OptKafkaSendOnly

func OptKafkaSendOnly() func(k *Kafka) error

OptKafkaSendOnly creates producer only. Subscribe operation will panic

func OptRabbitMQSendOnly

func OptRabbitMQSendOnly() func(mq *RabbitMQ) error

OptRabbitMQSendOnly creates only sending connection. Subscribe operation will panic

Types

type Kafka

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

Kafka Implements transport interface using Kafka broker.

func NewKafka

func NewKafka(brokerAddrs []string, options ...func(k *Kafka) error) (*Kafka, error)

NewKafka creates an instance of transport based on Kafka broker.

func (*Kafka) Close

func (k *Kafka) Close()

Close closes the transport

func (*Kafka) Publish

func (k *Kafka) Publish(ctx context.Context, event *events.CloudEvent, topic string, tenant string) error

Publish publishes an event

func (*Kafka) Subscribe

func (k *Kafka) Subscribe(ctx context.Context, topic string, handler events.Handler) (events.Subscription, error)

Subscribe subscribes to an event

type Noop

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

Noop implements dummy transport which does nothing (optionally prints to the output specified

func NewNoop

func NewNoop(out io.Writer) *Noop

NewNoop creates an instance of transport. out can be set to configure optional output.

func (*Noop) Close

func (t *Noop) Close()

Close closes transport.

func (*Noop) Publish

func (t *Noop) Publish(ctx context.Context, event *events.CloudEvent, topic string, tenant string) error

Publish publishes event.

func (*Noop) Subscribe

func (t *Noop) Subscribe(ctx context.Context, topic string, handler events.Handler) (events.Subscription, error)

Subscribe subscribes to event.

type RabbitMQ

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

RabbitMQ implements transport over AMQP protocol and RabbitMQ messaging service

func NewRabbitMQ

func NewRabbitMQ(url string, defaultExchangeName string, options ...func(mq *RabbitMQ) error) (mq *RabbitMQ, err error)

NewRabbitMQ creates new instance of RabbitMQ MessageQueue driver. Accepts variadic list of function options.

func (*RabbitMQ) Close

func (mq *RabbitMQ) Close()

Close closes AMQP connections and stops all subscriptions

func (*RabbitMQ) Publish

func (mq *RabbitMQ) Publish(ctx context.Context, event *events.CloudEvent, topic string, tenant string) error

Publish sends an event to RabbitMQ. tenant specifies the tenant

func (*RabbitMQ) Subscribe

func (mq *RabbitMQ) Subscribe(ctx context.Context, topic string, handler events.Handler) (events.Subscription, error)

Subscribe creates an active subscription on specified topic, and invokes handler function for every event received on given topic.