natsevents

Documentation

Index

• func EventFillDefaults(e *event.Event)
• func NewEvent(eventPrefix, eventType string, obj interface{}) *event.Event
• func SNI(serverName string) nats.Option
• func WithTransport(ctx context.Context, transport *NatsTransport) context.Context
• type CloudEventHandler
• type CloudEventReceiver
• type CloudEventSender
• type NatsTransport
  • func FromContext(ctx context.Context) *NatsTransport
  • func NewNatsTransport(server string) (*NatsTransport, error)
  • func (t *NatsTransport) Close() error
  • func (t *NatsTransport) Connected() bool
  • func (t *NatsTransport) Flush(timeout time.Duration) error
  • func (t *NatsTransport) RegisterHandler(eventHandler CloudEventHandler, topic string) error
  • func (t *NatsTransport) Request(ctx context.Context, event *event.Event, topic string, timeout time.Duration) (*event.Event, error)
  • func (t *NatsTransport) Send(ctx context.Context, event *event.Event, topic string) error
• type SNIDialer
  • func (d *SNIDialer) Dial(network, address string) (net.Conn, error)
• type Transport
• type Transporter

Functions

func EventFillDefaults

func EventFillDefaults(e *event.Event)

EventFillDefaults fills the required field with default values if not already set

func NewEvent

func NewEvent(eventPrefix, eventType string, obj interface{}) *event.Event

NewEvent creates a cloud event given minimal parameters

func SNI

func SNI(serverName string) nats.Option

func WithTransport

func WithTransport(ctx context.Context, transport *NatsTransport) context.Context

WithTransport adds the transport as context value

Types

type CloudEventHandler

type CloudEventHandler func(topic string, event *cloudevents.Event, payload []byte, err error) (*cloudevents.Event, error)

CloudEventHandler : callback to handle CloudEvents must return as fast as possible. event can be nil, in this case the payload will be set if event is not nil (properly decoded) the payload is nil

type CloudEventReceiver

type CloudEventReceiver interface {
	RegisterHandler(eventHandler CloudEventHandler, topic string) error
	Respond(*event.Event, error)
}

CloudEventReceiver allows to receive cloud events

type CloudEventSender

type CloudEventSender interface {
	Flush(timeout time.Duration) error
	Send(ctx context.Context, event *event.Event, topic string) error
	Request(ctx context.Context, event *event.Event, topic string, timeout time.Duration) (*event.Event, error)
}

CloudEventSender Sends cloud events

type NatsTransport

type NatsTransport struct {
	CloudEventSender
	CloudEventReceiver
	// contains filtered or unexported fields
}

NatsTransport allows to send and receive CloudEvents on nats

func FromContext

func FromContext(ctx context.Context) *NatsTransport

FromContext returns the transport set with WithTransport or nil

func NewNatsTransport

func NewNatsTransport(server string) (*NatsTransport, error)

NewNatsTransport connects to nats and creates a NatsTransport

func (*NatsTransport) Close

func (t *NatsTransport) Close() error

Close closes the underlying nats connection

func (*NatsTransport) Connected

func (t *NatsTransport) Connected() bool

Connected returns true if nats is connected

func (*NatsTransport) Flush

func (t *NatsTransport) Flush(timeout time.Duration) error

Flush flushes nats pending writes

func (*NatsTransport) RegisterHandler

func (t *NatsTransport) RegisterHandler(eventHandler CloudEventHandler, topic string) error

func (*NatsTransport) Request

func (t *NatsTransport) Request(ctx context.Context, event *event.Event, topic string, timeout time.Duration) (*event.Event, error)

Request sends a request and waits the response for timeout

func (*NatsTransport) Send

func (t *NatsTransport) Send(ctx context.Context, event *event.Event, topic string) error

Send sends the json representation of the event on nats topic

type SNIDialer

type SNIDialer struct {
	ServerName string
}

func (*SNIDialer) Dial

func (d *SNIDialer) Dial(network, address string) (net.Conn, error)

type Transport

type Transport interface {
	CloudEventSender
	CloudEventReceiver
	io.Closer
}

type Transporter

type Transporter interface {
	Transport() Transport
	SetTransport(transport Transport)
}