streams

Documentation

Index

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

Types

type Streams

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

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

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 = choria_streams
plugin.choria.adapter.discovery.queue_len = 1000 # default

Configure the stream output:

plugin.choria.adapter.discovery.stream.servers = js1:4222,js2:4222 # uses normal middleware server resolution when unset
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 inter.Framework) (adapter *Streams, err error)

func (*Streams) Init

func (sa *Streams) Init(ctx context.Context, cm inter.ConnectionManager) (err error)

func (*Streams) Process

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