From 04b88cc8645ed081b0b4eadfc3154cbfd5074553 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Mon, 5 Apr 2021 15:12:46 +0300 Subject: [PATCH] Add documentation for event stream --- events/event_stream.go | 17 ++++++++++++++ events/init.go | 51 ++++++++++++++++++++++++++++++++++++++++++ events/noop.go | 2 ++ mtglib/init.go | 15 +++++++++++++ 4 files changed, 85 insertions(+) diff --git a/events/event_stream.go b/events/event_stream.go index 9995eb0..cc971a6 100644 --- a/events/event_stream.go +++ b/events/event_stream.go @@ -9,12 +9,23 @@ import ( "github.com/OneOfOne/xxhash" ) +// EventStream is a default implementation of the mtglib.EventStream +// interface. +// +// EventStream manages a set of goroutines, observers. Main +// responsibility of the event stream is to route an event to relevant +// observer based on some hash so each observer will have all events +// which belong to some stream id. +// +// Thus, EventStream can spawn many observers. type EventStream struct { ctx context.Context ctxCancel context.CancelFunc chans []chan mtglib.Event } +// Send starts delivering of the message to observer with respect to a +// given context If context is closed, message could be not delivered. func (e EventStream) Send(ctx context.Context, evt mtglib.Event) { var chanNo uint32 @@ -31,10 +42,16 @@ func (e EventStream) Send(ctx context.Context, evt mtglib.Event) { } } +// Shutdown stops an event stream pipeline. func (e EventStream) Shutdown() { e.ctxCancel() } +// NewEventStream builds a new default event stream. +// +// If you give an empty array of observers, then NoopObserver is going +// to be used. If you give many observers, then they will process a +// message concurrently. func NewEventStream(observerFactories []ObserverFactory) EventStream { if len(observerFactories) == 0 { observerFactories = append(observerFactories, NewNoopObserver) diff --git a/events/init.go b/events/init.go index 1204652..f754ede 100644 --- a/events/init.go +++ b/events/init.go @@ -1,18 +1,69 @@ +// Events has a default implementations of EventStream for mtglib. +// +// Please see documentation for mtglib.EventStream interface to get an +// idea of such an abstraction. This package has implementations for the +// default event stream. +// +// Default event stream has a list of its own concepts. First, all it +// does is a routing of messages to known observers. It takes an event, +// defines its type and pass this message to a method of the observer. +// +// There might be many observers, but default event stream has a +// guarantee though. It uses StreamID as a sharding key and guarantees +// that a message with the same StreamID will be devlivered to the same +// observer instance. So, each producer is guarateed to get all relevant +// messages related to the same session. It is not possible that it will +// get EventFinish if it has not seen EventStart for that session yet. package events import "github.com/9seconds/mtg/v2/mtglib" +// Observer is an instance that listens for the incoming events. +// +// As it is said in the package description, the default event stream +// guarantees that all events with the same StreamID are going to be +// routed to the same instance of the observer. So, there is no need +// to synchronize information about streams between many observers +// instances, they can have their local storage. type Observer interface { + // EventStart reacts on incoming mtglib.EventStart event. EventStart(mtglib.EventStart) + + // EventFinish reacts on incoming mtglib.EventFinish event. EventFinish(mtglib.EventFinish) + + // EventConnectedToDC reacts on incoming mtglib.EventConnectedToDC + // event. EventConnectedToDC(mtglib.EventConnectedToDC) + + // EventDomainFronting reacts on incoming mtglib.EventDomainFronting + // event. EventDomainFronting(mtglib.EventDomainFronting) + + // EventTraffic reacts on incoming mtglib.EventTraffic event. EventTraffic(mtglib.EventTraffic) + + // EventConcurrencyLimited reacts on incoming + // mtglib.EventConcurrencyLimited event. EventConcurrencyLimited(mtglib.EventConcurrencyLimited) + + // EventIPBlocklisted reacts on incoming mtglib.EventIPBlocklisted event. EventIPBlocklisted(mtglib.EventIPBlocklisted) + + // EventReplayAttack reacts on incoming mtglib.EventReplayAttack event. EventReplayAttack(mtglib.EventReplayAttack) + // Shutdown stop observer. Default event stream guarantees: + // 1. If shutdown is executed, it is executed only once + // 2. Observer won't receieve any new message after this + // function call. Shutdown() } +// ObserverFactory creates a new instance of the observer. +// +// Default event stream creates a small set of goroutines to manage +// incoming messages. Each message is routed to an appropriate observer +// based on a sharding key, stream id. So, it is possible that an +// instance of mtg will have many observer instances, not a single one. type ObserverFactory func() Observer diff --git a/events/noop.go b/events/noop.go index 2fdb238..0582d9f 100644 --- a/events/noop.go +++ b/events/noop.go @@ -10,6 +10,7 @@ type noop struct{} func (n noop) Send(ctx context.Context, evt mtglib.Event) {} +// NewNoopStream creates a stream which discards each message. func NewNoopStream() mtglib.EventStream { return noop{} } @@ -26,6 +27,7 @@ func (n noopObserver) EventIPBlocklisted(_ mtglib.EventIPBlocklisted) func (n noopObserver) EventReplayAttack(_ mtglib.EventReplayAttack) {} func (n noopObserver) Shutdown() {} +// NewNoopObserver creates an observer which discards each message. func NewNoopObserver() Observer { return noopObserver{} } diff --git a/mtglib/init.go b/mtglib/init.go index 321da03..860e334 100644 --- a/mtglib/init.go +++ b/mtglib/init.go @@ -78,7 +78,22 @@ type Event interface { Timestamp() time.Time } +// EventStream is an abstraction that accepts a set of events produced +// by mtg. Its main goal is to inject your logging or monitoring system. +// +// The idea is simple. When mtg works, it emits a set of events during +// a lifecycle of the requestor: EventStart, EventFinish etc. mtg is a +// producer which puts these events into a stream. Responsibility of +// the stream is to deliver this event to consumers/observers. There +// might be many different observers (for example, you want to have both +// statsd and prometheus), mtg should know nothing about them. type EventStream interface { + // Send delivers an event to observers. Given context has to be + // respected. If the context is closed, all blocking operations should + // be released ASAP. + // + // It is possible that context is closed but the message is delivered. + // EventStream implementations should solve this issue somehow. Send(context.Context, Event) }