jetstream

Documentation

Index

• type Framework
• type JetStream
  • func Create(name string, choria Framework) (adapter *JetStream, err error)
  • func (sa *JetStream) Init(ctx context.Context, cm choria.ConnectionManager) (err error)
  • func (sa *JetStream) Process(ctx context.Context, wg *sync.WaitGroup)

Types

type Framework

type Framework interface {
	Configuration() *config.Config
	MiddlewareServers() (servers srvcache.Servers, err error)
	NewConnector(ctx context.Context, servers func() (srvcache.Servers, error), name string, logger *log.Entry) (conn choria.Connector, err error)
	NewRequestFromTransportJSON(payload []byte, skipvalidate bool) (msg protocol.Request, err error)
	NewReplyFromTransportJSON(payload []byte, skipvalidate bool) (msg protocol.Reply, err error)
}

type JetStream

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

JetStream is an adapter that connects a NATS topic with messages sent from Choria in its usual transport protocol to a NATS JetStream Message Set.

On the stream the messages will be JSON format with keys body, sender and time. Body is a base64 encoded string

Configure the adapters:

# required
plugin.choria.adapters = discovery
plugin.choria.adapter.discovery.type = jetstream
plugin.choria.adapter.discovery.queue_len = 1000 # default

Configure the stream output:

plugin.choria.adapter.discovery.stream.servers = js1:4222,js2:4222
plugin.choria.adapter.discovery.stream.topic = discovery # default, %s gets replaced with sender id
plugin.choria.adapter.discovery.stream.workers = 10 # default

Configure the NATS ingest:

plugin.choria.adapter.discovery.ingest.topic = mcollective.broadcast.agent.discovery
plugin.choria.adapter.discovery.ingest.protocol = request # or reply
plugin.choria.adapter.discovery.ingest.workers = 10 # default

func Create

func Create(name string, choria Framework) (adapter *JetStream, err error)

func (*JetStream) Init

func (sa *JetStream) Init(ctx context.Context, cm choria.ConnectionManager) (err error)

func (*JetStream) Process

func (sa *JetStream) Process(ctx context.Context, wg *sync.WaitGroup)