federation

Documentation

Index

• Constants
• func NewChoriaNatsEgest(workers int, mode int, capacity int, broker *FederationBroker, ...) (*pooledWorker, error)
• func NewChoriaNatsIngest(workers int, mode int, capacity int, broker *FederationBroker, ...) (*pooledWorker, error)
• func NewChoriaReplyTransformer(workers int, capacity int, broker *FederationBroker, logger *log.Entry) (*pooledWorker, error)
• func NewChoriaRequestTransformer(workers int, capacity int, broker *FederationBroker, logger *log.Entry) (*pooledWorker, error)
• func PooledWorkerFactory(name string, workers int, mode int, capacity int, broker *FederationBroker, ...) (*pooledWorker, error)
• type ChoriaFramework
• type FederationBroker
  • func NewFederationBroker(clusterName string, choria ChoriaFramework) (broker *FederationBroker, err error)
  • func (fb *FederationBroker) Start(ctx context.Context, wg *sync.WaitGroup)

Constants

const (
	Unconnected = iota
	Federation
	Collective
)

Functions

func NewChoriaNatsEgest

func NewChoriaNatsEgest(workers int, mode int, capacity int, broker *FederationBroker, logger *log.Entry) (*pooledWorker, error)

func NewChoriaNatsIngest

func NewChoriaNatsIngest(workers int, mode int, capacity int, broker *FederationBroker, logger *log.Entry) (*pooledWorker, error)

func NewChoriaReplyTransformer

func NewChoriaReplyTransformer(workers int, capacity int, broker *FederationBroker, logger *log.Entry) (*pooledWorker, error)

func NewChoriaRequestTransformer

func NewChoriaRequestTransformer(workers int, capacity int, broker *FederationBroker, logger *log.Entry) (*pooledWorker, error)

func PooledWorkerFactory

func PooledWorkerFactory(name string, workers int, mode int, capacity int, broker *FederationBroker, logger *log.Entry, worker func(context.Context, *pooledWorker, int, *log.Entry)) (*pooledWorker, error)

Types

type ChoriaFramework

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

type FederationBroker

type FederationBroker struct {
	Name string
	// contains filtered or unexported fields
}

func NewFederationBroker

func NewFederationBroker(clusterName string, choria ChoriaFramework) (broker *FederationBroker, err error)

func (*FederationBroker) Start

func (fb *FederationBroker) Start(ctx context.Context, wg *sync.WaitGroup)