Further scaffolding

cmd/datad/main.go

@@ -0,0 +1,3 @@
+package main
+
+func main() {}

core/data/bus.go

@@ -0,0 +1,91 @@
+package data
+
+import (
+	"sync"
+
+	"github.com/google/uuid"
+)
+
+type Bus interface {
+	Publish(id uuid.UUID, env Envelope) error
+	Subscribe(id uuid.UUID) (<-chan Envelope, func(), error)
+}
+
+type InMemBus struct {
+	mu          sync.RWMutex
+	subs        map[uuid.UUID][]chan Envelope
+	publishChan chan published
+}
+
+type published struct {
+	id  uuid.UUID
+	env Envelope
+}
+
+const (
+	busBufferSize        = 256 // buffer size for publishChan
+	subscriberBufferSize = 128 // buffer size for each subscriber channel
+)
+
+func NewInMemBus() *InMemBus {
+	b := &InMemBus{
+		subs:        make(map[uuid.UUID][]chan Envelope),
+		publishChan: make(chan published, busBufferSize),
+	}
+	go b.run()
+	return b
+}
+
+func (b *InMemBus) Publish(id uuid.UUID, env Envelope) error {
+	b.publishChan <- published{id: id, env: env}
+	return nil
+}
+
+func (b *InMemBus) Subscribe(id uuid.UUID) (<-chan Envelope, func(), error) {
+	ch := make(chan Envelope, subscriberBufferSize)
+
+	b.mu.Lock()
+	b.subs[id] = append(b.subs[id], ch)
+	b.mu.Unlock()
+
+	cancel := func() {
+		b.mu.Lock()
+		defer b.mu.Unlock()
+
+		chans := b.subs[id]
+		for i, c := range chans {
+			if c == ch {
+				// swap-delete
+				chans[i] = chans[len(chans)-1]
+				chans = chans[:len(chans)-1]
+				break
+			}
+		}
+
+		if len(chans) == 0 {
+			delete(b.subs, id)
+		} else {
+			b.subs[id] = chans
+		}
+	}
+
+	return ch, cancel, nil
+}
+
+func (b *InMemBus) run() {
+	for pub := range b.publishChan {
+		id := pub.id
+
+		b.mu.RLock()
+		chans := b.subs[id]
+		for _, ch := range chans {
+			select {
+			case ch <- pub.env:
+			default:
+				<-ch
+				ch <- pub.env
+			}
+		}
+		b.mu.RUnlock()
+	}
+}

core/data/data.go

@@ -1,33 +0,0 @@
-// Package data ...
-package data
-
-import "time"
-
-type Envelope struct {
-	Identifier string
-	Timestamp  time.Time
-	Sequence   uint64
-	Payload    []byte
-}
-
-type Source interface {
-	Start(config any) error
-	Stop()
-	Name() string
-
-	Subscribe(key string) (<-chan Envelope, error)
-	Unsubscribe(key string) error
-
-	IsValidKey(key string) bool
-	GetSubscriptions() []string
-}
-
-type Processor interface {
-	Start(config any, send chan<- Envelope, receive <-chan Envelope) error
-	Stop()
-}
-
-type Sink interface {
-	Start(config any, receive <-chan Envelope) error
-	Stop()
-}

core/data/descriptor.go

@@ -0,0 +1,9 @@
+package data
+
+type Descriptor struct {
+	Type       string
+	Key        string
+	Schema     string
+	Attributes map[string]string
+	Tags       []string
+}

core/data/egress.go

@@ -0,0 +1,18 @@
+package data
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+)
+
+type Egress interface {
+	Name() string
+	Configure(cfg map[string]any) error
+	Connect(ctx context.Context, actions EgresActions) error
+	Disconnect(ctx context.Context) error
+}
+
+type EgresActions interface {
+	Subscribe(ctx context.Context, namespace string, id uuid.UUID) (<-chan Envelope, func(), error)
+}

core/data/envelope.go

@@ -0,0 +1,13 @@
+// Package data ...
+package data
+
+import (
+	"time"
+)
+
+type Envelope struct {
+	SendTime time.Time
+
+	Descriptor Descriptor
+	Payload    any
+}

core/data/errors.go

@@ -1,10 +0,0 @@
-package data
-
-import "errors"
-
-var (
-	ErrUnknownKey        = errors.New("unknown key")
-	ErrAlreadySubscribed = errors.New("already subscribed")
-	ErrNotSubscribed     = errors.New("not subscribed")
-	ErrDroppedMessage    = errors.New("dropped message: buffer full")
-)

core/data/events/bar.go

@@ -0,0 +1,12 @@
+package events
+
+import "time"
+
+type Bar struct {
+	Open     float64
+	High     float64
+	Low      float64
+	Close    float64
+	Volume   float64
+	Interval time.Duration
+}

core/data/events/custom.go

@@ -0,0 +1,6 @@
+package events
+
+type Custom struct {
+	Bytes       []byte
+	ContentType string
+}

core/data/events/domain.go

@@ -0,0 +1,22 @@
+// Package events ...
+package events
+
+type DataType uint8
+
+const (
+	TradeType DataType = iota
+	QuoteType
+	BarType
+	MBPDeltaType
+	MBPSnapshotType
+	MBODeltaType
+	MBOSnapshotType
+	CustomType
+)
+
+type Side uint8
+
+const (
+	Bid Side = iota
+	Ask
+)

core/data/events/mbodelta.go

@@ -0,0 +1,20 @@
+package events
+
+type MBODelta struct {
+	Operation MBOOrderOp
+	OrderID   string
+	Side      Side
+	Price     float64
+	Size      float64
+	IsMaker   bool
+	Seq       uint64
+	ParentID  string
+}
+
+type MBOOrderOp uint8
+
+const (
+	OrderAdd MBOOrderOp = iota
+	OrderMod
+	OrderDel
+)

core/data/events/mbosnapshot.go

@@ -0,0 +1,14 @@
+package events
+
+type MBOSnapshot struct {
+	Orders []OrderEntry
+	Seq    uint64
+}
+
+type OrderEntry struct {
+	OrderID string
+	Side    Side
+	Price   float64
+	Size    float64
+	IsMaker bool
+}

core/data/events/mbpdelta.go

@@ -0,0 +1,8 @@
+package events
+
+type MBPDelta struct {
+	Side  Side
+	Price float64
+	Size  float64
+	Seq   uint64
+}

core/data/events/mbpsnapshot.go

@@ -0,0 +1,13 @@
+package events
+
+type MBPSnapshot struct {
+	Bids  []PriceLevel
+	Asks  []PriceLevel
+	Depth int
+	Seq   uint64
+}
+
+type PriceLevel struct {
+	Price float64
+	Size  float64
+}

core/data/events/quote.go

@@ -0,0 +1,8 @@
+package events
+
+type Quote struct {
+	BidPrice float64
+	BidSize  float64
+	AskPrice float64
+	AskSize  float64
+}

core/data/events/trade.go

@@ -0,0 +1,16 @@
+package events
+
+type Trade struct {
+	Price     float64
+	Qty       float64
+	Aggressor AggressorSide
+	TradeID   string
+}
+
+type AggressorSide uint8
+
+const (
+	AggUnknown AggressorSide = iota
+	AggBuy
+	AggSell
+)

core/data/ingress.go

@@ -0,0 +1,34 @@
+package data
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+type Ingress interface {
+	Name() string
+	Configure(cfg map[string]any) error
+	Connect(ctx context.Context, actions IngressActions) error
+	Disconnect(ctx context.Context) error
+}
+
+type IngressActions interface {
+	Emit(namespace string, id uuid.UUID, envelope Envelope) error
+	EmitBatch(namespace string, id uuid.UUID, envelopes []Envelope) error
+}
+
+type LiveIngress interface {
+	Subscribe(ctx context.Context, key ...string) error
+	Unsubscribe(ctx context.Context, key ...string) error
+	ListAvailableKeys(ctx context.Context) []string
+	IsValidKey(ctx context.Context, key string) bool
+}
+
+type HistoricalIngress interface {
+	Fetch(ctx context.Context, key string, start time.Time, end time.Time) ([]Envelope, error)
+	ListAvailableKeys(ctx context.Context) []string
+	IsValidKey(ctx context.Context, key string) bool
+	// some other method that gives the avalible time period for a key
+}

core/data/processor.go

@@ -0,0 +1,14 @@
+package data
+
+import "context"
+
+type Processor interface {
+	Name() string
+	Configure(cfg map[string]any) error
+	Run(ctx context.Context, actions ProcessorActions) error
+}
+
+type ProcessorActions interface {
+	IngressActions
+	EgresActions
+}

core/data/registry.go

@@ -0,0 +1 @@
+package data

core/data/registry/processor.go

@@ -1,34 +0,0 @@
-// Package registry ...
-package registry
-
-import (
-	"fmt"
-	"sync"
-
-	"gitlab.michelsen.id/phillmichelsen/tessera/core/data"
-)
-
-type ProcessorFactory func(string) (data.Processor, error)
-
-type Processors struct {
-	mu sync.RWMutex
-	m  map[string]ProcessorFactory
-}
-
-func NewProcessors() *Processors { return &Processors{m: make(map[string]ProcessorFactory)} }
-
-func (r *Processors) Register(name string, f ProcessorFactory) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	r.m[name] = f
-}
-
-func (r *Processors) New(name string, params string) (data.Processor, error) {
-	r.mu.RLock()
-	f, ok := r.m[name]
-	r.mu.RUnlock()
-	if !ok {
-		return nil, fmt.Errorf("unknown processor: %s", name)
-	}
-	return f(params)
-}

core/data/registry/sinks.go

@@ -1,33 +0,0 @@
-package registry
-
-import (
-	"fmt"
-	"sync"
-
-	"gitlab.michelsen.id/phillmichelsen/tessera/core/data"
-)
-
-type SinkFactory func(string) (data.Sink, error)
-
-type Sinks struct {
-	mu sync.RWMutex
-	m  map[string]SinkFactory
-}
-
-func NewSinks() *Sinks { return &Sinks{m: make(map[string]SinkFactory)} }
-
-func (r *Sinks) Register(name string, f SinkFactory) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	r.m[name] = f
-}
-
-func (r *Sinks) New(name string, params string) (data.Sink, error) {
-	r.mu.RLock()
-	f, ok := r.m[name]
-	r.mu.RUnlock()
-	if !ok {
-		return nil, fmt.Errorf("unknown sink: %s", name)
-	}
-	return f(params)
-}

core/data/registry/sources.go

@@ -1,33 +0,0 @@
-package registry
-
-import (
-	"fmt"
-	"sync"
-
-	"gitlab.michelsen.id/phillmichelsen/tessera/core/data"
-)
-
-type SourceFactory func(string) (data.Source, error)
-
-type Sources struct {
-	mu sync.RWMutex
-	m  map[string]SourceFactory
-}
-
-func NewSources() *Sources { return &Sources{m: make(map[string]SourceFactory)} }
-
-func (r *Sources) Register(name string, f SourceFactory) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	r.m[name] = f
-}
-
-func (r *Sources) New(name string, params string) (data.Source, error) {
-	r.mu.RLock()
-	f, ok := r.m[name]
-	r.mu.RUnlock()
-	if !ok {
-		return nil, fmt.Errorf("unknown source: %s", name)
-	}
-	return f(params)
-}

core/data/router.go

@@ -0,0 +1,16 @@
+package data
+
+import "github.com/google/uuid"
+
+type Router struct {
+	busses map[string]Bus
+}
+
+func NewRouter() *Router {
+	return &Router{
+		busses: make(map[string]Bus),
+	}
+}
+
+func (r *Router) Emit(namespace string, id uuid.UUID, envelope Envelope) error         { return nil }
+func (r *Router) EmitBatch(namespace string, id uuid.UUID, enveloeps []Envelope) error { return nil }

go.mod

@@ -1,3 +1,5 @@
 module gitlab.michelsen.id/phillmichelsen/tessera
 
 go 1.25.1
+
+require github.com/google/uuid v1.6.0 // indirect

go.sum

@@ -0,0 +1,2 @@
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=