eventhub

Documentation

Overview

The package implements a minimal set of Azure Event Hubs functionality.

We could use https://github.com/Azure/azure-event-hubs-go but it's too huge and also it has a number of dependencies that aren't desired in the project.

Index

• func CheckMessageResponse(msg *amqp.Message) error
• type Client
  • func Dial(host, name string, opts ...Option) (*Client, error)
  • func DialConnectionString(cs string, opts ...Option) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Subscribe(ctx context.Context, fn func(event *Event) error, opts ...SubscribeOption) error
• type Credentials
  • func ParseConnectionString(cs string) (*Credentials, error)
• type Event
• type Option
  • func WithConnOption(opt amqp.ConnOption) Option
  • func WithSASLPlain(username, password string) Option
  • func WithTLSConfig(tc *tls.Config) Option
• type SubscribeOption
  • func WithSubscribeConsumerGroup(name string) SubscribeOption
  • func WithSubscribeLinkOption(opt amqp.LinkOption) SubscribeOption
  • func WithSubscribeSince(t time.Time) SubscribeOption

Functions

func CheckMessageResponse

func CheckMessageResponse(msg *amqp.Message) error

CheckMessageResponse checks for 200 response code otherwise returns an error.

Types

type Client

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

Client is an EventHub client.

func Dial

func Dial(host, name string, opts ...Option) (*Client, error)

Dial connects to the named EventHub and returns a client instance.

func DialConnectionString

func DialConnectionString(cs string, opts ...Option) (*Client, error)

DialConnectionString dials an EventHub instance using the given connection string.

func (*Client) Close

func (c *Client) Close() error

Close closes underlying AMQP connection.

func (*Client) Subscribe

func (c *Client) Subscribe(
	ctx context.Context,
	fn func(event *Event) error,
	opts ...SubscribeOption,
) error

Subscribe subscribes to all hub's partitions and registers the given handler and blocks until it encounters an error or the context is cancelled.

Non-nil errors returned from fn reject AMQP messages.

type Credentials

type Credentials struct {
	Endpoint            string
	SharedAccessKeyName string
	SharedAccessKey     string
	EntityPath          string
}

Credentials is an evenhub connection string representation.

func ParseConnectionString

func ParseConnectionString(cs string) (*Credentials, error)

ParseConnectionString parses the given connection string into Credentials structure.

type Event

type Event struct {
	*amqp.Message
	// contains filtered or unexported fields
}

Event is an Event Hub event, simply wraps an AMQP message.

type Option

type Option func(c *Client)

Option is a client configuration option.

func WithConnOption

func WithConnOption(opt amqp.ConnOption) Option

WithConnOption sets a low-level connection option.

func WithSASLPlain

func WithSASLPlain(username, password string) Option

WithSASLPlain configures connection username and password.

func WithTLSConfig

func WithTLSConfig(tc *tls.Config) Option

WithTLSConfig sets connection TLS configuration.

type SubscribeOption

type SubscribeOption func(r *sub)

SubscribeOption is a Subscribe option.

func WithSubscribeConsumerGroup

func WithSubscribeConsumerGroup(name string) SubscribeOption

WithSubscribeConsumerGroup overrides default consumer group, default is `$Default`.

func WithSubscribeLinkOption

func WithSubscribeLinkOption(opt amqp.LinkOption) SubscribeOption

WithSubscribeLinkOption is a low-level subscription configuration option.

func WithSubscribeSince

func WithSubscribeSince(t time.Time) SubscribeOption

WithSubscribeSince requests events that occurred after the given time.