cached_test.go

package sseserver

import (
	"bytes"
	"errors"
	"io"
	"log/slog"
	"net/http/httptest"
	"strconv"
	"testing"
	"time"
)

var (
	_ MultiStream = &CachedStream{}
	_ Stream      = &CachedStream{}
)

var (
	topic      = "benchmark_topic"
	localCache = newCache(time.Minute, time.Minute)
)

func init() {
	slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))

	for n := 0; n < 10000; n++ {
		event := &Event{
			ID: strconv.Itoa(n + 1),
		}

		localCache.add(topic, strconv.Itoa(n), event)
	}
}

func BenchmarkLocalCacheAdd(b *testing.B) {
	for n := 0; n < b.N; n++ {
		localCache.add(topic, strconv.Itoa(n-1), &Event{
			ID: strconv.Itoa(n),
		})
	}
}

func BenchmarkLocalCacheGet(b *testing.B) {
	for n := 0; n < b.N; n++ {
		events, _ := localCache.get(topic, "0", "5000", 5000, nil)

		for range events {
		}
	}
}

func assertReceivedEvents(t *testing.T, resp *httptest.ResponseRecorder, events ...Event) {
	t.Helper()

	var buf bytes.Buffer
	for _, event := range events {
		_ = write(&buf, &event)
	}

	if buf.String() != resp.Body.String() {
		t.Errorf("Expected response body: %q, got: %q", buf.String(), resp.Body.String())
	}
}

func TestCachedResync(t *testing.T) {
	t.Parallel()

	stream := NewCached(Config{
		Reconnect:             0,
		KeepAlive:             0,
		Lifetime:              10 * time.Millisecond,
		QueueLength:           32,
		ResyncEventsThreshold: 2,
	}, "first", time.Minute, time.Minute)
	defer stream.Stop()

	// Publish two events
	event1 := Event{ID: "16"}
	stream.Publish(&event1)

	event2 := Event{ID: "32"}
	stream.Publish(&event2)

	w := httptest.NewRecorder()
	// connect with initial last event ID to receive both cached events
	_ = stream.Subscribe(t.Context(), w, "first")

	// Assert both events were received
	assertReceivedEvents(t, w, event1, event2)
}

func TestCachedResyncWithBroadcast(t *testing.T) {
	t.Parallel()

	stream := NewCached(Config{
		Reconnect:             0,
		KeepAlive:             0,
		Lifetime:              10 * time.Millisecond,
		QueueLength:           32,
		ResyncEventsThreshold: 2,
	}, "first", time.Minute, time.Minute)
	defer stream.Stop()

	// Publish two events, with broadcast in between
	event1 := Event{ID: "16"}
	stream.Publish(&event1)
	stream.PublishBroadcast(&Event{ID: "999"})
	event2 := Event{ID: "32"}
	stream.Publish(&event2)

	w := httptest.NewRecorder()
	// connect with initial last event ID to receive both cached events,
	// broadcasted event should be excluded
	_ = stream.Subscribe(t.Context(), w, "first")

	// Assert both events were received
	assertReceivedEvents(t, w, event1, event2)
}

func TestCachedError(t *testing.T) {
	t.Parallel()

	stream := NewCached(Config{
		Reconnect:   0,
		KeepAlive:   0,
		Lifetime:    10 * time.Millisecond,
		QueueLength: 32,
	}, "8", time.Minute, time.Minute)
	defer stream.Stop()

	w := httptest.NewRecorder()
	// resyncing from non-existent event ID should return error
	err := stream.Subscribe(t.Context(), w, "non-existent")
	assertReceivedEvents(t, w, ResyncRequiredErrorEvent())
	if !errors.Is(err, ErrCacheMiss) {
		t.Errorf("Expected error: %v, got: %v", ErrCacheMiss, err)
	}
}

func TestCachedResyncTopics(t *testing.T) {
	t.Parallel()

	stream := NewCachedMultiStream(Config{
		Reconnect:             0,
		KeepAlive:             0,
		Lifetime:              10 * time.Millisecond,
		QueueLength:           32,
		ResyncEventsThreshold: 5,
	}, map[string]string{
		"topic1": "first1",
		"topic2": "first2",
	}, time.Minute, time.Minute)
	defer stream.Stop()

	// Generate two sub-streams of events
	var events1, events2 []Event

	for i := 0; i < 5; i++ {
		event1 := Event{ID: strconv.Itoa(i * 10)}
		events1 = append(events1, event1)
		stream.PublishTopic("topic1", &event1)

		event2 := Event{ID: strconv.Itoa(i * 20)}
		events2 = append(events2, event2)
		stream.PublishTopic("topic2", &event2)
	}

	t.Run("with topic1", func(t *testing.T) {
		w := httptest.NewRecorder()
		_ = stream.SubscribeTopic(t.Context(), w, "topic1", "first1")
		assertReceivedEvents(t, w, events1...)
	})

	t.Run("with topic2", func(t *testing.T) {
		w := httptest.NewRecorder()
		_ = stream.SubscribeTopic(t.Context(), w, "topic2", "first2")
		assertReceivedEvents(t, w, events2...)
	})
}