Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

xbus

package module

Go to main page

Versions in this module

v3

v3.3.3

Sep 22, 2023

v3.3.2

Sep 20, 2023

v3.3.1

Dec 16, 2022

v3.3.0

Feb 1, 2022

v3.2.1

Jan 6, 2021

v3.2.0

Dec 15, 2020

Changes in this version

+ func GetZeroLogger(logger Logger) *zerolog.Logger

type LoggerOptions

+ AppName string

+ OutputMiddleware func(io.Writer) io.Writer

+ type ZeroLogger interface

+ Zero func() *zerolog.Logger

+ func LoggerWithStr(logger Logger, name, value string) ZeroLogger

+ func WrapZeroLogger(log zerolog.Logger) ZeroLogger

v3.1.0

May 4, 2020

Changes in this version

+ const Version

+ var ClientName string

+ var ClientVersion string

+ var DefaultMaxChunksByFragment = 1000

+ var EnvelopeSenderDefaultDelay = 10 * time.Second

+ var ErrAlreadyExists = errors.New("Key or Certificate already exists")

+ var ErrCouldNotGetLock = errors.New("Could not get lock on the file")

+ var ErrEmissionCanceled = errors.New("Emission canceled")

+ var ErrEventListFrozen = errors.New("Event list is frozen")

+ var ErrEventTypesNotKnown = errors.New("Event types are not known yet")

+ var ErrIncompleteEnvelope = errors.New("Incomplete envelope")

+ var ErrInvalidGraph = errors.New("Invalid graph")

+ var ErrInvalidKey = errors.New("Store keys cannot contain ':'")

+ var ErrNoSuchInput = errors.New("No such input")

+ var ErrNotFound = errors.New("Not found")

+ var ErrReceptionError = errors.New("Reception error")

+ var ErrTimeout = errors.New("Timeout")

+ var ErrWrongStatus = errors.New("Receiver is done receiving")

+ func ApcTestSetCloseOutput(apc *ActorProcessingContext, ...)

+ func ApcTestSetOpenOutput(apc *ActorProcessingContext, ...)

+ func ApcTestSetProcessingEnd(apc *ActorProcessingContext, ...)

+ func ApcTestSetReceiveEnvelope(apc *ActorProcessingContext, ...)

+ func ApcTestSetSend(apc *ActorProcessingContext, send func(api.ActorProcessingState))

+ func ApcTestSetSendFragment(apc *ActorProcessingContext, ...)

+ func DefineAndBindPFlags(v *viper.Viper, flags *pflag.FlagSet) error

+ func ForgeEnvelopeFromEvent(event api.Event) api.Envelope

+ func ForgeEnvelopeFromItems(eventType string, items []api.Item) api.Envelope

+ func RegisterActorService(typename string, init NewActorServiceFunc)

+ func SendEnvelope(ctx context.Context, env envelope.Envelope, maxSize int, ...) error

+ func UnregisterActorService(typename string)

+ type APCLogger interface

+ EnvErrorf func(*api.Envelope, string, ...interface{})

+ EnvNoticef func(*api.Envelope, string, ...interface{})

+ EnvWarningf func(*api.Envelope, string, ...interface{})

+ Errorf func(string, ...interface{})

+ Noticef func(string, ...interface{})

+ Warningf func(string, ...interface{})

+ type Actor struct

+ Client *Client

+ ID api.UUID

+ Kind api.Actor_Kind

+ Log *ActorLogger

+ Name string

+ Options ActorOptions

+ func (a *Actor) CloseOutput(context *api.ProcessingContext, output string) (api.EnvelopeAck, error)

+ func (a *Actor) CustomEmission(ctx context.Context, envelope envelope.Envelope, request bool) (*Emission, error)

+ func (a *Actor) Emit(ctx context.Context, envelope envelope.Envelope) (*Emission, error)

+ func (a *Actor) GetBoolSettingD(name string, defaultValue bool) bool

+ func (a *Actor) GetClient() *Client

+ func (a *Actor) GetIntSettingD(name string, defaultValue int) int

+ func (a *Actor) GetStringSetting(name string) string

+ func (a *Actor) GetStringSliceSetting(name string) []string

+ func (a *Actor) MustGetStringSetting(name string) string

+ func (a *Actor) OpenOutput(context *api.ProcessingContext, output string, eventTypes ...string) *EnvelopeSender

+ func (a *Actor) ProcessingEnd(context *api.ProcessingContext, status api.ActorProcessingState_Status, ...) error

+ func (a *Actor) ProcessingError(context *api.ProcessingContext, format string, fmtArgs ...interface{}) error

+ func (a *Actor) ProcessingSuccess(context *api.ProcessingContext) error

+ func (a *Actor) Request(ctx context.Context, envelope envelope.Envelope) (envelope.Envelope, error)

+ func (a *Actor) SendEnvelope(ctx *api.ProcessingContext, output string, env api.Envelope) (status api.EnvelopeAck_ReceptionStatus, err error)

+ func (a *Actor) SendFragment(context *api.ProcessingContext, output string, fragment *api.Envelope) (api.EnvelopeAck_ReceptionStatus, error)

+ func (a *Actor) SendItems(ctx *api.ProcessingContext, output, eventType string, items []api.Item) (status api.EnvelopeAck_ReceptionStatus, err error)

+ func (a *Actor) SendLogs(context *api.ProcessingContext, envelope *api.Envelope, ...) error

+ func (a *Actor) Subscribe(handle ActorProcessHandler) (*Subscription, error)

+ type ActorAPCLogger struct

+ func (l *ActorAPCLogger) EnvErrorf(env *api.Envelope, format string, args ...interface{})

+ func (l *ActorAPCLogger) EnvNoticef(env *api.Envelope, format string, args ...interface{})

+ func (l *ActorAPCLogger) EnvWarningf(env *api.Envelope, format string, args ...interface{})

+ func (l *ActorAPCLogger) Errorf(format string, args ...interface{})

+ func (l *ActorAPCLogger) Noticef(format string, args ...interface{})

+ func (l *ActorAPCLogger) Warningf(format string, args ...interface{})

+ type ActorLogger struct

+ func (al ActorLogger) EnvErrorf(context *api.ProcessingContext, env *api.Envelope, format string, ...)

+ func (al ActorLogger) EnvNoticef(context *api.ProcessingContext, env *api.Envelope, format string, ...)

+ func (al ActorLogger) EnvWarningf(context *api.ProcessingContext, env *api.Envelope, format string, ...)

+ func (al ActorLogger) Errorf(format string, fmtArgs ...interface{})

+ func (al ActorLogger) Noticef(format string, fmtArgs ...interface{})

+ func (al ActorLogger) Warningf(format string, fmtArgs ...interface{})

+ type ActorOptions struct

+ Service ServiceOptions

+ func ReadActor(v *viper.Viper, key string) (ActorOptions, error)

+ type ActorProcessHandler func(*ActorProcessingContext) error

+ type ActorProcessService interface

+ Process func(*ActorProcessingContext) error

+ type ActorProcessingContext struct

+ Actor *Actor

+ Context context.Context

+ LocalLog Logger

+ Log APCLogger

+ Request *api.ActorProcessRequest

+ func (apc *ActorProcessingContext) ActorLeaving()

+ func (apc *ActorProcessingContext) CloseOutput(outputName string) error

+ func (apc *ActorProcessingContext) CompleteEnvelope(inputName string, timeout time.Duration) (*api.Envelope, error)

+ func (apc *ActorProcessingContext) Detach()

+ func (apc *ActorProcessingContext) Done()

+ func (apc *ActorProcessingContext) End(status api.ActorProcessingState_Status, msgs []api.LogMessage)

+ func (apc *ActorProcessingContext) Error(err error)

+ func (apc *ActorProcessingContext) ForwardEnvelope(outputName string, envelopeID api.UUID) error

+ func (apc *ActorProcessingContext) ForwardInput(inputName, outputName string) error

+ func (apc *ActorProcessingContext) GetEnvelope(inputName string) (envelope.Envelope, error)

+ func (apc *ActorProcessingContext) GetEnvelopeBuilder(inputName string) (*envelope.BuilderFromFragment, error)

+ func (apc *ActorProcessingContext) GetInput(name string) *api.ActorProcessRequest_Input

+ func (apc *ActorProcessingContext) IsDetached() bool

+ func (apc *ActorProcessingContext) IsDone() bool

+ func (apc *ActorProcessingContext) OpenOutput(outputName string, eventTypes ...string) *EnvelopeSender

+ func (apc *ActorProcessingContext) ReadEnvelope(inputName string) (*EnvelopeReceiver, error)

+ func (apc *ActorProcessingContext) Send(outputName string, e envelope.Envelope) error

+ type ActorService interface

+ func MakeActorService(typename string, actor *Actor) ActorService

+ type COptions struct

+ ActorList []ActorOptions

+ ConnOptions nats.Options

+ ForceStart bool

+ ID api.UUID

+ Name string

+ NatsTimeout time.Duration

+ NoReconnect bool

+ PersistentStore PersistentStore

+ SentryDSN string

+ TLS TLSOptions

+ Type api.Account_Type

+ func COptionsFromFile(configPath string, accountType api.Account_Type, logger Logger) (COptions, error)

+ func COptionsFromViper(v *viper.Viper, accountType api.Account_Type, varDir string, logger Logger) COptions

+ type Client struct

+ API *api.Client

+ Control *api_control.Client

+ Log Logger

+ func ClientFromFile(path string, accountType api.Account_Type, logger Logger) (*Client, error)

+ func NewClient(options COptions, logger Logger) *Client

+ func (c *Client) AccountAccept(id api.UUID, expire time.Time) (api.Account, error)

+ func (c *Client) AccountDelete(id api.UUID) error

+ func (c *Client) AccountFind(identifier string) (api.Account, error)

+ func (c *Client) AccountList() ([]api.Account, error)

+ func (c *Client) AccountReject(id api.UUID) (api.Account, error)

+ func (c *Client) AccountRenewAPIKey(id api.UUID) (api.Account, error)

+ func (c *Client) ActorAccept(id api.UUID) (api.Actor, error)

+ func (c *Client) ActorCreate(actors []api.Actor) ([]api.Actor, error)

+ func (c *Client) ActorDelete(ids []api.UUID) error

+ func (c *Client) ActorFind(identifier string) (api.Actor, error)

+ func (c *Client) ActorList(actors []api.Actor) ([]api.Actor, error)

+ func (c *Client) ActorOptions() []ActorOptions

+ func (c *Client) ActorReject(id api.UUID) (api.Actor, error)

+ func (c *Client) AddActor(actor ActorOptions)

+ func (c *Client) ClientGetActors() ([]api.Actor, error)

+ func (c *Client) Close()

+ func (c *Client) Connect() error

+ func (c *Client) CreateCSR(host, org, cn string, overwrite bool) (err error)

+ func (c *Client) CreateKey(force bool) error

+ func (c *Client) Done() <-chan struct{}

+ func (c *Client) EnvelopePurge(ctx context.Context, cb func(count uint32)) error

+ func (c *Client) GetActor(id api.UUID) *Actor

+ func (c *Client) GetActorByName(name string) *Actor

+ func (c *Client) GetActorOptions(id api.UUID) (ActorOptions, bool)

+ func (c *Client) GetActorOptionsByName(name string) (ActorOptions, bool)

+ func (c *Client) GetConn() *nats.Conn

+ func (c *Client) GetOptions() COptions

+ func (c *Client) GetPersistentStore() PersistentStore

+ func (c *Client) ID() api.UUID

+ func (c *Client) LogsPurge(ctx context.Context, before time.Time) (count int64, err error)

+ func (c *Client) Name() string

+ func (c *Client) PMProcessQuery(ctx context.Context, level api.LogLevel, includeClosed bool, ...) ([]api.PMProcess, error)

+ func (c *Client) PMProcessSetStatus(processID api.UUID, status api.PMProcess_Status, comment string) error

+ func (c *Client) PipelineGet(info api.PipelineInfo) (string, error)

+ func (c *Client) PipelineQuery(name string) ([]api.PipelineInfo, error)

+ func (c *Client) PipelineSave(info api.PipelineInfo, graph string) (api.PipelineInfo, string, []string, error)

+ func (c *Client) PipelineSetStatus(info api.PipelineInfo) (api.PipelineInfo, error)

+ func (c *Client) ProcessControl(processID api.UUID, command control.ProcessControlRequest_Command) error

+ func (c *Client) ProcessExport(processIDs []api.UUID) ([]string, error)

+ func (c *Client) ProcessPurge(ctx context.Context, processIDs []api.UUID, ...) error

+ func (c *Client) ProcessQuery(ctx context.Context, filter api.ProcessFilter, cb func([]api.Process)) error

+ func (c *Client) RawProcessingEnd(actorID api.UUID, context api.ProcessingContext, ...) error

+ func (c *Client) ReceiveEnvelope(ctx context.Context, clientKind, clientID string, envelopeID api.UUID, ...) (*envelope.BuilderFromFragment, error)

+ func (c *Client) Register() (api.Registration_Status, error)

+ func (c *Client) RetrieveEnvelope(ctx context.Context, clientKind, clientID string, envelopeID api.UUID) *EnvelopeReceiver

+ func (c *Client) RetrieveEnvelopeFrom(ctx context.Context, clientKind, clientID string, envelope api.Envelope, ...) *EnvelopeReceiver

+ func (c *Client) SendEnvelopeFragment(actorID api.UUID, ctx *api.ProcessingContext, output string, env api.Envelope) (api.EnvelopeAck_ReceptionStatus, error)

+ func (c *Client) Shutdown()

+ func (c *Client) StartServices() error

+ func (c *Client) Startup(startServices bool) error

+ func (c *Client) StopServices() error

+ func (c *Client) UpdateActors() error

+ func (c *Client) WhoAmI(apiKey string) (api.Account, error)

+ type Emission struct

+ func NewEmission(ctx context.Context, a *Actor, envelope envelope.Envelope, request bool) (*Emission, error)

+ func (e *Emission) Close()

+ func (e *Emission) GetResponse() (envelope.Envelope, error)

+ func (e *Emission) Wait() error

+ func (e *Emission) WaitCompletion() error

+ type EnvelopeReceiver struct

+ ID api.UUID

+ Input string

+ func NewEnvelopeReceiver(input string, e api.Envelope) (*EnvelopeReceiver, error)

+ func NewEnvelopeReceiverEmpty() *EnvelopeReceiver

+ func NewEnvelopeReceiverFromBuilder(input string, builder *envelope.BuilderFromFragment) (*EnvelopeReceiver, error)

+ func NewEnvelopeReceiverSimple(id api.UUID) *EnvelopeReceiver

+ func (r *EnvelopeReceiver) AddFragment(e api.Envelope) error

+ func (r *EnvelopeReceiver) Cancel()

+ func (r *EnvelopeReceiver) Complete(timeout time.Duration) (*api.Envelope, error)

+ func (r *EnvelopeReceiver) Done() <-chan struct{}

+ func (r *EnvelopeReceiver) EventTypes() []string

+ func (r *EnvelopeReceiver) Events() []api.Event

+ func (r *EnvelopeReceiver) Finalize() error

+ func (r *EnvelopeReceiver) GetItemChan(eventType string) <-chan api.Item

+ func (r *EnvelopeReceiver) LastError() error

+ func (r *EnvelopeReceiver) NextEventItems(eventID api.UUID, max int) (items []api.Item)

+ func (r *EnvelopeReceiver) NextItems(eventType string, max int) (items []api.Item)

+ func (r *EnvelopeReceiver) PartialEvents() []api.Event

+ func (r *EnvelopeReceiver) SetError(err error)

+ func (r *EnvelopeReceiver) Status() api.EnvelopeAck_ReceptionStatus

+ func (r *EnvelopeReceiver) WaitCompletion(timeout time.Duration) bool

+ func (r *EnvelopeReceiver) WaitEventTypes(timeout time.Duration) []string

+ type EnvelopeSender struct

+ func NewEnvelopeSender(eventTypes []string, maxSize int, sendEnvelope SendEnvelopeFunc) *EnvelopeSender

+ func (s *EnvelopeSender) AddEvent(eventType string) (api.UUID, error)

+ func (s *EnvelopeSender) AddEventItems(id api.UUID, items ...api.Item) error

+ func (s *EnvelopeSender) AddEventWithID(id api.UUID, eventType string) error

+ func (s *EnvelopeSender) AddItems(eventType string, items ...api.Item) error

+ func (s *EnvelopeSender) Cancel()

+ func (s *EnvelopeSender) Close() error

+ func (s *EnvelopeSender) CloseEvent(id api.UUID, checksum api.Checksum) error

+ func (s *EnvelopeSender) Flush()

+ func (s *EnvelopeSender) FreezeEventList()

+ func (s *EnvelopeSender) GetID() api.UUID

+ func (s *EnvelopeSender) SetMaxDelay(delay time.Duration)

+ type ErrProcessFailed struct

+ func NewErrProcessFailed(errors []api.LogMessage) ErrProcessFailed

+ func (e ErrProcessFailed) Error() string

+ type EventValidatorMap map[api.UUID]tools.EventValidator

+ type FilePersistentStore struct

+ func NewFilePersistentStore(path string) (*FilePersistentStore, error)

+ func (s *FilePersistentStore) Del(key string) error

+ func (s *FilePersistentStore) Get(key string) (string, error)

+ func (s *FilePersistentStore) Put(key, value string) error

+ type ItemQueue []api.Item

+ func (q *ItemQueue) Add(items ...api.Item)

+ func (q *ItemQueue) Remove() api.Item

+ func (q *ItemQueue) RemoveN(n int) []api.Item

+ func (q *ItemQueue) RemoveNS(n, maxBytes int) ([]api.Item, int)

+ func (q *ItemQueue) RemoveS(maxBytes int) ([]api.Item, int)

+ type LegacyEnvelopeSender struct

+ func NewLegacyEnvelopeSender(eventTypes []string, maxSize int, sendEnvelope SendEnvelopeFunc) *LegacyEnvelopeSender

+ func NewLegacyEnvelopeSenderWithIDs(envelopeID api.UUID, eventTypes []string, eventIDs []api.UUID, maxSize int, ...) *LegacyEnvelopeSender

+ func (s *LegacyEnvelopeSender) AddEvent(eventType string) (api.UUID, error)

+ func (s *LegacyEnvelopeSender) AddEventItems(id api.UUID, items ...api.Item) error

+ func (s *LegacyEnvelopeSender) AddEventWithID(id api.UUID, eventType string) error

+ func (s *LegacyEnvelopeSender) AddItems(eventType string, items ...api.Item) error

+ func (s *LegacyEnvelopeSender) Cancel()

+ func (s *LegacyEnvelopeSender) Close() error

+ func (s *LegacyEnvelopeSender) CloseEvent(id api.UUID, checksum api.Checksum) error

+ func (s *LegacyEnvelopeSender) Flush()

+ func (s *LegacyEnvelopeSender) FreezeEventList()

+ func (s *LegacyEnvelopeSender) GetID() api.UUID

+ func (s *LegacyEnvelopeSender) SetMaxDelay(delay time.Duration)

+ type Logger interface

+ Debugf func(format string, v ...interface{})

+ Errorf func(format string, v ...interface{})

+ Fatalf func(format string, v ...interface{})

+ Noticef func(format string, v ...interface{})

+ Tracef func(format string, v ...interface{})

+ func GetLogger(opts *LoggerOptions) Logger

+ type LoggerOptions struct

+ Debug bool

+ LogFile string

+ Logtime bool

+ RemoteSyslog string

+ Syslog bool

+ Trace bool

+ type NATSLogger struct

+ func NewNATSLogger(logger Logger) NATSLogger

+ func (l NATSLogger) Warnf(format string, v ...interface{})

+ type NewActorServiceFunc func(*Actor) ActorService

+ type NoopLogger struct

+ func NewNoopLogger() NoopLogger

+ func (NoopLogger) Debugf(format string, v ...interface{})

+ func (NoopLogger) Errorf(format string, v ...interface{})

+ func (NoopLogger) Fatalf(format string, v ...interface{})

+ func (NoopLogger) Noticef(format string, v ...interface{})

+ func (NoopLogger) Tracef(format string, v ...interface{})

+ type PersistentStore interface

+ Del func(key string) error

+ Get func(key string) (string, error)

+ Put func(key string, value string) error

+ type PrefixedLogger struct

+ func NewPrefixedLogger(prefix string, logger Logger) *PrefixedLogger

+ func (l *PrefixedLogger) Debugf(format string, v ...interface{})

+ func (l *PrefixedLogger) Errorf(format string, v ...interface{})

+ func (l *PrefixedLogger) Fatalf(format string, v ...interface{})

+ func (l *PrefixedLogger) Noticef(format string, v ...interface{})

+ func (l *PrefixedLogger) Tracef(format string, v ...interface{})

+ type RavenLogger struct

+ func NewRavenLogger(client *raven.Client, nextLogger Logger) *RavenLogger

+ func (r *RavenLogger) Debugf(format string, v ...interface{})

+ func (r *RavenLogger) Errorf(format string, v ...interface{})

+ func (r *RavenLogger) Fatalf(format string, v ...interface{})

+ func (r *RavenLogger) Noticef(format string, v ...interface{})

+ func (r *RavenLogger) Tracef(format string, v ...interface{})

+ type RunningService struct

+ func StartupService(actor *Actor, svc ActorService, log Logger) (*RunningService, error)

+ func (rs *RunningService) Shutdown() error

+ type SendEnvelopeFunc func(*api.Envelope) (api.EnvelopeAck_ReceptionStatus, error)

+ type ServiceOptions struct

+ Settings map[string]interface{}

+ Type string

+ type StartupShutdown interface

+ Shutdown func() error

+ Startup func() error

+ type Subscription struct

+ func (s Subscription) Unsubscribe() error

+ type TLSKeyValueSaver struct

+ func NewTLSKeyValueSaver(set func(key, value string) error) TLSKeyValueSaver

+ func (s TLSKeyValueSaver) SetCSR(csr []byte) error

+ func (s TLSKeyValueSaver) SetCertificate(cert []byte) error

+ func (s TLSKeyValueSaver) SetPrivKey(key *ecdsa.PrivateKey) error

+ func (s TLSKeyValueSaver) SetRootCAs(cas [][]byte) error

+ type TLSOptions struct

+ CSR []byte

+ Certificate []byte

+ PrivKey *ecdsa.PrivateKey

+ PubKey crypto.PublicKey

+ RootCAs [][]byte

+ TLSSaver TLSOptionsSaver

+ func (o *TLSOptions) SetCSR(csr []byte) error

+ func (o *TLSOptions) SetCertificate(cert []byte) error

+ func (o *TLSOptions) SetPrivKey(key *ecdsa.PrivateKey) error

+ func (o *TLSOptions) SetRootCAs(cas [][]byte) error

+ type TLSOptionsSaver interface

+ SetCSR func([]byte) error

+ SetCertificate func([]byte) error

+ SetPrivKey func(*ecdsa.PrivateKey) error

+ SetRootCAs func([][]byte) error

+ func LoadTLSOptionFiles(v *viper.Viper, configPath string, logger Logger) (TLSOptionsSaver, error)

+ func TLSMapSaver(initialOptions TLSOptions, save func(map[string]string) error) (TLSOptionsSaver, error)

+ func TLSOptionsFileSaver(certFile string, keySources map[string]string) (TLSOptionsSaver, error)

+ type ValidationMessagesList []string

+ func (l ValidationMessagesList) Error() string

Other modules containing this package

xbus.io/go-xbus/v4