nats

Documentation

Index

• type Event
• type Ingress
  • func New(conf *config.NATS, monitor monitor.Monitor) (*Ingress, error)
  • func (i *Ingress) Close()
  • func (i *Ingress) SubsribeHandler(handler func(b []map[string]interface{}, table string)) error
  • func (i *Ingress) SubsribeHandlerWithPool(ctx context.Context, handler func(b []map[string]interface{}, split string)) error

Types

type Event

type Event map[string]interface{}

type Ingress

type Ingress struct {
	JSClient jetstream.Client
	// contains filtered or unexported fields
}

func New

func New(conf *config.NATS, monitor monitor.Monitor) (*Ingress, error)

New create new ingestion from nats jetstream to sinks.

func (*Ingress) Close

func (i *Ingress) Close()

Close ingress

func (*Ingress) SubsribeHandler

func (i *Ingress) SubsribeHandler(handler func(b []map[string]interface{}, table string)) error

SubsribeHandler subscribes to specific subject and unmarshal the message into talaria's event type. The event message then will be used as the input of the handler function defined.

func (*Ingress) SubsribeHandlerWithPool

func (i *Ingress) SubsribeHandlerWithPool(ctx context.Context, handler func(b []map[string]interface{}, split string)) error

SubscribeHandlerWithPool process the message concurrently using goroutine pool. The message will be asynchornously executed to reduce the message process time to avoid being slow consumer.