broker

README

Event bus

The broker is the entry point for core events. Core engines only ever use a single method of the broker interface. Subscribers don't interact with the broker directly, but are called by the broker (which pushes events to them).

For the core

Where engines previously depended on any number of buffers to push data to that plugins, API's, and stores needed to handle, they will now push events onto the bus via the broker. In the events package, you'll find a number of events (trade event, order event, market data, etc...). These events all have a constructor for convenience. Most events can even be created through the generic constructor. Say, for example, we receive a new order, or the state of an order changed:

func (e *Engine) Foo(ctx, order *types.Order) {
    // some code, order changes:
    e.broker.Send(events.NewOrderEvent(ctx, *order))
    // or, using the generic constructor
    if evt, err := events.New(ctx, *order); err == nil {
        e.broker.Send(evt)
    }
}

The buffers needed to be flushed at the end of a block. Now, the end of a block is considered an event (TimeUpdate). If some events/data needs to be batched, and only processed at the end of the block, the subscriber should subscribe to the time event, and use that as a trigger/signal to do what is expected.

Some events (e.g. position related events) are created a ton of times. Sending them individually creates a metric ton of routines sending out all the data, which is bad for performance. To get around this issue, the broker includes a SendBatch(events []Events) function. The events in the slice can be of different types, but we derive the subscriber types from the first event in the slice, and assume that subscribers who listen for this event can either safely ignore any other events in the slice, or can handle all of them (currently, all subscribers can handle unknown events - they simply ignore them). It might be a good idea to have some placeholder event added for things like MarketEvent (which basically is a catch-all event type for things we want to log).

For non-core (subscribers etc...)

The core spits out events, without caring where the data ends up. The subscribers need to be registered with the broker to receive the data and process it. There are 2 main categories of subscribers: required (think of it as ack) and non-required subscribers. If a subscriber is registered as required, the broker will make sure that the subscriber receives all events, in the correct order. The non-required subscribers have a channel onto which the broker pushes events, unless the channel buffer is full. The subscriber is not required, so rather than blocking the broker, the broker is free to skip these subscribers and simply carry on. As a result the latter type of subscriber is not guaranteed to receive every single event, but unlike their required counterparts, they are unable to have a meaningful impact on the performance of the broker.

Subscribers implement a fairly simple interface:

type Subscriber interface {
    Ack() bool
	Push(val ...events.Event)
	Skip() <-chan struct{}
	Closed() <-chan struct{}
	C() chan<- events.Event
	Types() []events.Type
	SetID(id int)
	ID() int
}

• Ack(): Indicates whether or not this subscriber "Ack's" the events it receives. In this case: the event has to be passed to the Push function.
• Types: The broker uses this call to determine what events this subscriber wants to receive.
• Push: A required subscriber will receive all its events from the broker through a normal function call. This ensures the event is indeed received. This function accepts one or more events. If SendBatch() was called on the broker, ack'ing subscribers will receive the entire batch of events in a single call.
• C: This is used for non-required subscribers. This returns a write channel where the prober attempts to push an event onto. If this fails (because the buffer is full), the event is dropped for that subscriber
• Closed: A subscriber can be halted (if it's no longer needed). This function will return a closed channel indicating that this subscriber is redundant, and should be removed
• Skip: If a subscriber only periodically needs to get data, we kan keep it registered, but put it in a "paused" state. A paused subscriber will return a closed channel for as long as we're not interested in receiving data.
• ID and SetID: Subscribers have a broker ID (int). This is unique for all subscribers, and allows us to remove a specific subscriber manually, should we need to. The broker will also call SetID when the subscriber is registered.

Documentation

Index

• func NewBufferFilesEventSource(bufferFilesDir string, timeBetweenBlocks time.Duration, ...) (*bufferFileEventSource, error)
• func NewEventReceiverSender(config Config, log *logging.Logger, chainID string) (rawEventReceiverSender, error)
• func NewFanOutEventSource(source EventReceiver, sendChannelBufferSize int, expectedNumSubscribers int) *fanOutEventSource
• type BlockStore
• type Broker
  • func New(ctx context.Context, log *logging.Logger, config Config, chainID string, ...) (*Broker, error)
  • func (b *Broker) Receive(ctx context.Context) error
  • func (b *Broker) Send(event events.Event)
  • func (b *Broker) Subscribe(s broker.Subscriber) int
  • func (b *Broker) SubscribeBatch(subs ...broker.Subscriber)
  • func (b *Broker) Unsubscribe(k int)
• type BufferedEventSourceConfig
• type ChainInfoI
• type Config
  • func NewDefaultConfig() Config
• type Deserializer
  • func NewDeserializer(source RawEventReceiver) *Deserializer
  • func (e *Deserializer) Listen() error
  • func (e *Deserializer) Receive(ctx context.Context) (<-chan events.Event, <-chan error)
• type EventReceiver
• type FileBufferedEventSource
  • func NewBufferedEventSource(ctx context.Context, log *logging.Logger, config BufferedEventSourceConfig, ...) (*FileBufferedEventSource, error)
  • func (m *FileBufferedEventSource) Listen() error
  • func (m *FileBufferedEventSource) Receive(ctx context.Context) (<-chan []byte, <-chan error)
• type FileEventSourceConfig
• type OrderEventWithVegaTime
  • func (oe *OrderEventWithVegaTime) GetOrder() *vega.Order
  • func (oe *OrderEventWithVegaTime) VegaTime() time.Time
• type ProtocolUpgradeHandler
• type RawEventReceiver
• type SQLBrokerSubscriber
• type SQLStoreBroker
  • func NewSQLStoreBroker(log *logging.Logger, config Config, chainID string, eventsource EventReceiver, ...) *SQLStoreBroker
  • func (b *SQLStoreBroker) Receive(ctx context.Context) error
• type SQLStoreEventBroker
• type SocketConfig
• type TestInterface
• type TransactionManager

Functions

func NewBufferFilesEventSource

func NewBufferFilesEventSource(bufferFilesDir string, timeBetweenBlocks time.Duration,
	sendChannelBufferSize int, chainID string) (*bufferFileEventSource, error,
)

func NewEventReceiverSender

func NewEventReceiverSender(config Config, log *logging.Logger, chainID string) (rawEventReceiverSender, error)

func NewFanOutEventSource

func NewFanOutEventSource(source EventReceiver, sendChannelBufferSize int, expectedNumSubscribers int) *fanOutEventSource

Types

type BlockStore

type BlockStore interface {
	Add(ctx context.Context, b entities.Block) error
	GetLastBlock(ctx context.Context) (entities.Block, error)
}

type Broker

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

Broker - the base broker type perhaps we can extend this to embed into type-specific brokers.

func New

func New(ctx context.Context, log *logging.Logger, config Config, chainID string,
	eventsource EventReceiver,
) (*Broker, error)

New creates a new base broker.

func (*Broker) Receive

func (b *Broker) Receive(ctx context.Context) error

func (*Broker) Send

func (b *Broker) Send(event events.Event)

Send sends an event to all subscribers.

func (*Broker) Subscribe

func (b *Broker) Subscribe(s broker.Subscriber) int

Subscribe registers a new subscriber, returning the key.

func (*Broker) SubscribeBatch

func (b *Broker) SubscribeBatch(subs ...broker.Subscriber)

func (*Broker) Unsubscribe

func (b *Broker) Unsubscribe(k int)

Unsubscribe removes subscriber from broker this does not change the state of the subscriber.

type BufferedEventSourceConfig

type BufferedEventSourceConfig struct {
	EventsPerFile           int   `description:"the number of events to store in a file buffer, set to 0 to disable the buffer" long:"events-per-file"`
	SendChannelBufferSize   int   `description:"sink event channel buffer size"                                                 long:"send-buffer-size"`
	Archive                 bool  `description:"archives event buffer files after they have been read, default false"           long:"archive"`
	ArchiveMaximumSizeBytes int64 `description:"the maximum size of the archive directory"                                      long:"archive-maximum-size"`
}

type ChainInfoI

type ChainInfoI interface {
	SetChainID(string) error
	GetChainID() (string, error)
}

type Config

type Config struct {
	Level                          encoding.LogLevel         `long:"log-level"`
	SocketConfig                   SocketConfig              `group:"Socket"                                                                           namespace:"socket"`
	SocketServerInboundBufferSize  int                       `long:"socket-server-inbound-buffer-size"`
	SocketServerOutboundBufferSize int                       `long:"socket-server-outbound-buffer-size"`
	FileEventSourceConfig          FileEventSourceConfig     `group:"FileEventSourceConfig"                                                            namespace:"fileeventsource"`
	UseEventFile                   encoding.Bool             `description:"set to true to source events from a file"                                   long:"use-event-file"`
	PanicOnError                   encoding.Bool             `description:"if an error occurs on event push the broker will panic, else log the error" long:"panic-on-error"`
	UseBufferedEventSource         encoding.Bool             `description:"if true datanode will buffer events"                                        long:"use-buffered-event-source"`
	BufferedEventSourceConfig      BufferedEventSourceConfig `group:"BufferedEventSource"                                                              namespace:"bufferedeventsource"`
	EventBusClientBufferSize       int                       `long:"event-bus-client-buffer-size"`
}

Config represents the configuration of the broker.

func NewDefaultConfig

func NewDefaultConfig() Config

NewDefaultConfig creates an instance of config with default values.

type Deserializer

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

func NewDeserializer

func NewDeserializer(source RawEventReceiver) *Deserializer

func (*Deserializer) Listen

func (e *Deserializer) Listen() error

func (*Deserializer) Receive

func (e *Deserializer) Receive(ctx context.Context) (<-chan events.Event, <-chan error)

type EventReceiver

type EventReceiver interface {
	Listen() error
	Receive(ctx context.Context) (<-chan events.Event, <-chan error)
}

type FileBufferedEventSource

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

func NewBufferedEventSource

func NewBufferedEventSource(ctx context.Context, log *logging.Logger, config BufferedEventSourceConfig,
	source RawEventReceiver, bufferFilesDir string,
	archiveFilesDir string,
) (*FileBufferedEventSource, error)

func (*FileBufferedEventSource) Listen

func (m *FileBufferedEventSource) Listen() error

func (*FileBufferedEventSource) Receive

func (m *FileBufferedEventSource) Receive(ctx context.Context) (<-chan []byte, <-chan error)

type FileEventSourceConfig

type FileEventSourceConfig struct {
	Directory             string            `description:"the directory container the event files"               long:"directory"`
	TimeBetweenBlocks     encoding.Duration `description:"the time between sending blocks"                       string:"time-between-blocks"`
	SendChannelBufferSize int               `description:"size of channel buffer used to send events to broker " long:"send-buffer-size"`
}

type OrderEventWithVegaTime

type OrderEventWithVegaTime struct {
	events.Order
	// contains filtered or unexported fields
}

func (*OrderEventWithVegaTime) GetOrder

func (oe *OrderEventWithVegaTime) GetOrder() *vega.Order

func (*OrderEventWithVegaTime) VegaTime

func (oe *OrderEventWithVegaTime) VegaTime() time.Time

type ProtocolUpgradeHandler

type ProtocolUpgradeHandler interface {
	OnProtocolUpgradeEvent(ctx context.Context, chainID string, lastCommittedBlockHeight int64) error
	GetProtocolUpgradeStarted() bool
}

type RawEventReceiver

type RawEventReceiver interface {
	Listen() error
	Receive(ctx context.Context) (<-chan []byte, <-chan error)
}

type SQLBrokerSubscriber

type SQLBrokerSubscriber interface {
	SetVegaTime(vegaTime time.Time)
	Flush(ctx context.Context) error
	Push(ctx context.Context, val events.Event) error
	Types() []events.Type
}

type SQLStoreBroker

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

SQLStoreBroker : push events to each subscriber with a single go routine across all types.

func NewSQLStoreBroker

func NewSQLStoreBroker(
	log *logging.Logger,
	config Config,
	chainID string,
	eventsource EventReceiver,
	transactionManager TransactionManager,
	blockStore BlockStore,
	onBlockCommitted func(ctx context.Context, chainId string, lastCommittedBlockHeight int64, snapshotTaken bool),
	protocolUpdateHandler ProtocolUpgradeHandler,
	subs []SQLBrokerSubscriber,
) *SQLStoreBroker

func (*SQLStoreBroker) Receive

func (b *SQLStoreBroker) Receive(ctx context.Context) error

type SQLStoreEventBroker

type SQLStoreEventBroker interface {
	Receive(ctx context.Context) error
}

type SocketConfig

type SocketConfig struct {
	IP                 string `description:" "             long:"ip"`
	Port               int    `description:" "             long:"port"`
	MaxReceiveTimeouts int    `long:"max-receive-timeouts"`
	TransportType      string `long:"transport-type"`
}

type TestInterface

type TestInterface interface {
	Send(event events.Event)
	Subscribe(s broker.Subscriber) int
	SubscribeBatch(subs ...broker.Subscriber)
	Unsubscribe(k int)
	Receive(ctx context.Context) error
}

TestInterface interface (horribly named) is declared here to provide a drop-in replacement for broker mocks used throughout in addition to providing the classical mockgen functionality, this mock can be used to check the actual events that will be generated so we don't have to rely on test-only helper functions.

type TransactionManager

type TransactionManager interface {
	WithConnection(ctx context.Context) (context.Context, error)
	WithTransaction(ctx context.Context) (context.Context, error)
	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}