egress

Documentation

Index

• Constants
• type Batch
• type EmbeddedNotifier
  • func (n EmbeddedNotifier) NotifyAgent(batchID string) error
• type Forwarder
  • func NewForwarder(cfg ForwarderConfig) Forwarder
  • func (f Forwarder) Forward(batchID string) error
  • func (f Forwarder) ForwardBatch(batch Batch) error
  • func (f Forwarder) Shutdown()
  • func (f Forwarder) Start() error
• type ForwarderConfig
  • func NewForwarderDefaultConfig() ForwarderConfig
• type NetworkNotifier
  • func (n NetworkNotifier) NotifyAgent(batchID string) error
• type NoopStorage
  • func (n NoopStorage) Commit(_ context.Context, _ string) error
  • func (n NoopStorage) GetBatch(_ context.Context, _ string) (Batch, error)
• type Notifier
• type Storage
• type StorageConfig
• type StorageOption
  • func WithEgressTable(table string) StorageOption

Constants

const (
	DefaultEgressTableName = "streams_egress" // default egress table name.

)

Types

type Batch

type Batch struct {
	BatchID           string
	TransportBatchRaw []byte
	InsertTime        time.Time
}

A Batch is an aggregate of messages written by a system ready to be published to message brokers or similar infrastructure.

type EmbeddedNotifier

type EmbeddedNotifier struct {
	Forwarder Forwarder
}

EmbeddedNotifier an agent-less Notifier implementation used by a system to call a Forwarder instance directly.

func (EmbeddedNotifier) NotifyAgent

func (n EmbeddedNotifier) NotifyAgent(batchID string) error

type Forwarder

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

A Forwarder is an internal component used by an egress proxy agent to redirect queued-traffic (i.e. stream messages) to a message broker or similar infrastructure.

func NewForwarder

func NewForwarder(cfg ForwarderConfig) Forwarder

func (Forwarder) Forward

func (f Forwarder) Forward(batchID string) error

Forward triggers a new forward job for the specified batch.

func (Forwarder) ForwardBatch

func (f Forwarder) ForwardBatch(batch Batch) error

ForwardBatch triggers a new forward job for the specified batch.

func (Forwarder) Shutdown

func (f Forwarder) Shutdown()

Shutdown gracefully shuts down the Forwarder instance.

func (Forwarder) Start

func (f Forwarder) Start() error

Start initializes the Forwarder instance, blocking the I/O. The instance contains internal job scheduling mechanisms for asynchronous job processing.

type ForwarderConfig

type ForwarderConfig struct {
	Storage                   Storage        // A storage a Forwarder instance uses to fetch message batches (queued-traffic).
	Writer                    streams.Writer // A writer implementation a Forwarder instance uses to publish message batches to.
	Codec                     codec.Codec    // Codec used by the egress writer to store traffic messages.
	Logger                    *log.Logger    // Logger to write information to.
	TableName                 string         // Name of the egress table.
	ForwardJobTimeout         time.Duration  // Maximum time duration to wait a forward job to finish.
	ForwardJobTotalRetries    int            // Maximum count a forward job will be retried.
	ForwardJobRetryBackoff    time.Duration  // Initial time duration between each retry process.
	ForwardJobRetryBackoffMax time.Duration  // Maximum time duration between each retry process.
}

A ForwarderConfig is the configuration used by a Forwarder.

func NewForwarderDefaultConfig

func NewForwarderDefaultConfig() ForwarderConfig

type NetworkNotifier

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

func (NetworkNotifier) NotifyAgent

func (n NetworkNotifier) NotifyAgent(batchID string) error

type NoopStorage

type NoopStorage struct {
	WantGetBatch    Batch
	WantGetBatchErr error
	WantCommitErr   error
}

func (NoopStorage) Commit

func (n NoopStorage) Commit(_ context.Context, _ string) error

func (NoopStorage) GetBatch

func (n NoopStorage) GetBatch(_ context.Context, _ string) (Batch, error)

type Notifier

type Notifier interface {
	// NotifyAgent triggers egress proxy agent to forward traffic buffer (batch) into a stream.
	NotifyAgent(batchID string) error
}

A Notifier is an egress proxy component used by a system to notify the egress proxy agent to forward traffic buffer (batch) into a stream.

type Storage

type Storage interface {
	// GetBatch retrieves specified batch.
	GetBatch(ctx context.Context, batchID string) (Batch, error)
	// Commit Evicts specified batch.
	Commit(ctx context.Context, batchID string) error
}

A Storage is a special kind of storage where traffic is ingested (queued) so an egress proxy agent (or similar artifacts) may forward message batches to another set of infrastructure (e.g. message broker).

type StorageConfig

type StorageConfig struct {
	TableName string
}

A StorageConfig is the main configuration of a Storage.

type StorageOption

type StorageOption interface {
	Apply(*StorageConfig)
}

func WithEgressTable

func WithEgressTable(table string) StorageOption