natsstream

Documentation

Index

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

Types

type ChoriaFramework

type ChoriaFramework interface {
	Configuration() *config.Config
	MiddlewareServers() (servers srvcache.Servers, err error)
	NewRequestFromTransportJSON(payload []byte, skipvalidate bool) (msg protocol.Request, err error)
	NewReplyFromTransportJSON(payload []byte, skipvalidate bool) (msg protocol.Reply, err error)
}

type NatStream

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

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

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

Configure the stream:

plugin.choria.adapter.discovery.stream.servers = stan1:4222,stan2:4222
plugin.choria.adapter.discovery.stream.clusterid = prod
plugin.choria.adapter.discovery.stream.topic = discovery # default
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 ChoriaFramework) (adapter *NatStream, err error)

func (*NatStream) Init

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

func (*NatStream) Process

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