Add documentation for event stream

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)

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

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{}
 }

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)
 }