NATS broker

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## master
 
+- Add NATS-based broker. ([@palkan][])
+
 ## 1.4.6 (2023-10-25)
 
 - Add `@anycable/anycable-go` NPM package to install `anycable-go` as a JS project dependency. ([@palkan][])

Makefile

@@ -153,7 +153,7 @@ test-conformance-broker-redis: tmp/anycable-go-test
 	ANYCABLE_BROKER=memory ANYCABLE_BROADCAST_ADAPTER=redisx ANYCABLE_HTTP_BROADCAST_SECRET=any_secret ANYCABLE_PUBSUB=redis bundle exec anyt -c "tmp/anycable-go-test --headers=cookie,x-api-token" --target-url="ws://localhost:8080/cable" --require=etc/anyt/**/*.rb
 
 test-conformance-broker-nats: tmp/anycable-go-test
-	ANYCABLE_BROKER=memory ANYCABLE_EMBED_NATS=true ANYCABLE_ENATS_ADDR=nats://127.0.0.1:4343 ANYCABLE_PUBSUB=nats ANYCABLE_BROADCAST_ADAPTER=http ANYCABLE_HTTP_BROADCAST_SECRET=any_secret bundle exec anyt -c "tmp/anycable-go-test --headers=cookie,x-api-token" --target-url="ws://localhost:8080/cable" --require=etc/anyt/**/*.rb
+	ANYCABLE_BROKER=nats ANYCABLE_EMBED_NATS=true ANYCABLE_ENATS_ADDR=nats://127.0.0.1:4343 ANYCABLE_PUBSUB=nats ANYCABLE_BROADCAST_ADAPTER=http ANYCABLE_HTTP_BROADCAST_SECRET=any_secret bundle exec anyt -c "tmp/anycable-go-test --headers=cookie,x-api-token" --target-url="ws://localhost:8080/cable" --require=etc/anyt/**/*.rb
 
 test-conformance-all: test-conformance test-conformance-ssl test-conformance-http
 

broker/broker.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"time"
 
 	"github.com/anycable/anycable-go/common"
 )
@@ -57,6 +58,16 @@ type Broker interface {
 	FinishSession(sid string) error
 }
 
+// LocalBroker is a single-node broker that can used to store streams data locally
+type LocalBroker interface {
+	Start() error
+	Shutdown(ctx context.Context) error
+	SetEpoch(epoch string)
+	HistoryFrom(stream string, epoch string, offset uint64) ([]common.StreamMessage, error)
+	HistorySince(stream string, ts int64) ([]common.StreamMessage, error)
+	Store(stream string, msg []byte, seq uint64, ts time.Time) (uint64, error)
+}
+
 type StreamsTracker struct {
 	store map[string]uint64
 

broker/memory.go

@@ -43,6 +43,39 @@ func (ms *memstream) add(data string) uint64 {
 		data:      data,
 	}
 
+	ms.appendEntry(entry)
+
+	return ms.offset
+}
+
+func (ms *memstream) insert(data string, offset uint64, t time.Time) (uint64, error) {
+	ms.mu.Lock()
+	defer ms.mu.Unlock()
+
+	if t == (time.Time{}) {
+		t = time.Now()
+	}
+
+	ts := t.Unix()
+
+	if ms.offset >= offset {
+		return 0, fmt.Errorf("Offset %d is already taken", offset)
+	}
+
+	ms.offset = offset
+
+	entry := &entry{
+		offset:    offset,
+		timestamp: ts,
+		data:      data,
+	}
+
+	ms.appendEntry(entry)
+
+	return ms.offset, nil
+}
+
+func (ms *memstream) appendEntry(entry *entry) {
 	ms.data = append(ms.data, entry)
 
 	if len(ms.data) > ms.limit {
@@ -57,8 +90,6 @@ func (ms *memstream) add(data string) uint64 {
 	// Update memstream expiration deadline on every added item
 	// We keep memstream alive for 10 times longer than ttl (so we can re-use it and its offset)
 	ms.deadline = time.Now().Add(time.Duration(ms.ttl*10) * time.Second).Unix()
-
-	return ms.offset
 }
 
 func (ms *memstream) expire() {
@@ -286,6 +317,24 @@ func (b *Memory) HistorySince(name string, ts int64) ([]common.StreamMessage, er
 	return history, nil
 }
 
+func (b *Memory) Store(name string, data []byte, offset uint64, ts time.Time) (uint64, error) {
+	b.streamsMu.Lock()
+
+	if _, ok := b.streams[name]; !ok {
+		b.streams[name] = &memstream{
+			data:  []*entry{},
+			ttl:   b.config.HistoryTTL,
+			limit: b.config.HistoryLimit,
+		}
+	}
+
+	stream := b.streams[name]
+
+	b.streamsMu.Unlock()
+
+	return stream.insert(string(data), offset, ts)
+}
+
 func (b *Memory) CommitSession(sid string, session Cacheable) error {
 	b.sessionsMu.Lock()
 	defer b.sessionsMu.Unlock()

broker/memory_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestExpire(t *testing.T) {
+func TestMemory_Expire(t *testing.T) {
 	config := NewConfig()
 	config.HistoryTTL = 1
 
@@ -44,7 +44,7 @@ func TestExpire(t *testing.T) {
 	assert.Empty(t, history)
 }
 
-func TestLimit(t *testing.T) {
+func TestMemory_Limit(t *testing.T) {
 	config := NewConfig()
 	config.HistoryLimit = 2
 
@@ -64,7 +64,7 @@ func TestLimit(t *testing.T) {
 	assert.Equal(t, "c", history[1].Data)
 }
 
-func TestFromOffset(t *testing.T) {
+func TestMemory_FromOffset(t *testing.T) {
 	config := NewConfig()
 
 	broker := NewMemoryBroker(nil, &config)
@@ -95,6 +95,33 @@ func TestFromOffset(t *testing.T) {
 	})
 }
 
+func TestMemory_Store(t *testing.T) {
+	config := NewConfig()
+
+	broker := NewMemoryBroker(nil, &config)
+	broker.SetEpoch("2023")
+
+	ts := time.Now()
+
+	offset, err := broker.Store("test", []byte("a"), 10, ts)
+	require.NoError(t, err)
+	assert.EqualValues(t, 10, offset)
+
+	_, err = broker.Store("test", []byte("b"), 11, ts)
+	require.NoError(t, err)
+
+	_, err = broker.Store("tes2", []byte("c"), 1, ts)
+	require.NoError(t, err)
+
+	history, err := broker.HistoryFrom("test", broker.epoch, 10)
+	require.NoError(t, err)
+	assert.Len(t, history, 1)
+	assert.EqualValues(t, 11, history[0].Offset)
+
+	_, err = broker.Store("test", []byte("c"), 3, ts)
+	assert.Error(t, err)
+}
+
 func TestMemstream_filterByOffset(t *testing.T) {
 	ms := &memstream{
 		ttl:   1,