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 thePush
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. IfSendBatch()
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 subscriberClosed
: 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 removedSkip
: 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
andSetID
: 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 callSetID
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
- func ReadRawEvent(eventFile *os.File, offset int64) (event []byte, seqNum uint64, totalBytesRead uint32, err error)
- type BlockStore
- type Broker
- type BufferedEventSourceConfig
- type ChainInfoI
- type Config
- type Deserializer
- type EventReceiver
- type FileBufferedEventSource
- type FileEventSourceConfig
- type OrderEventWithVegaTime
- type ProtocolUpgradeHandler
- type RawEventReceiver
- type SQLBrokerSubscriber
- type SQLStoreBroker
- type SQLStoreEventBroker
- type SocketConfig
- type TestInterface
- type TransactionManager
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewBufferFilesEventSource ¶ added in v0.69.0
func NewEventReceiverSender ¶ added in v0.65.0
func NewFanOutEventSource ¶
func NewFanOutEventSource(source EventReceiver, sendChannelBufferSize int, expectedNumSubscribers int) *fanOutEventSource
Types ¶
type BlockStore ¶
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) 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 ¶
Unsubscribe removes subscriber from broker this does not change the state of the subscriber.
type BufferedEventSourceConfig ¶ added in v0.61.0
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 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 ¶ added in v0.69.0
type Deserializer struct {
// contains filtered or unexported fields
}
func NewDeserializer ¶ added in v0.69.0
func NewDeserializer(source RawEventReceiver) *Deserializer
func (*Deserializer) Listen ¶ added in v0.69.0
func (e *Deserializer) Listen() error
type EventReceiver ¶ added in v0.65.0
type FileBufferedEventSource ¶ added in v0.61.0
type FileBufferedEventSource struct {
// contains filtered or unexported fields
}
func NewBufferedEventSource ¶ added in v0.61.0
func NewBufferedEventSource(ctx context.Context, log *logging.Logger, config BufferedEventSourceConfig, source RawEventReceiver, bufferFilesDir string, archiveFilesDir string, ) (*FileBufferedEventSource, error)
func (*FileBufferedEventSource) Listen ¶ added in v0.61.0
func (m *FileBufferedEventSource) Listen() 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 ¶
func (*OrderEventWithVegaTime) GetOrder ¶
func (oe *OrderEventWithVegaTime) GetOrder() *vega.Order
func (*OrderEventWithVegaTime) VegaTime ¶
func (oe *OrderEventWithVegaTime) VegaTime() time.Time
type ProtocolUpgradeHandler ¶ added in v0.63.0
type RawEventReceiver ¶ added in v0.69.0
type SQLBrokerSubscriber ¶ added in v0.55.0
type SQLStoreBroker ¶ added in v0.57.0
type SQLStoreBroker struct {
// contains filtered or unexported fields
}
SQLStoreBroker : push events to each subscriber with a single go routine across all types.
func NewSQLStoreBroker ¶ added in v0.55.0
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
type SQLStoreEventBroker ¶ added in v0.55.0
type SocketConfig ¶
type TestInterface ¶ added in v0.55.0
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.