pump_test.go

package rchan

import (
	"context"
	"testing"
	"time"
)

func TestNativeSelect(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	a, _ := New[string](ctx, rdb, testName(t, rdb), WithCapacity(2))
	b, _ := New[string](ctx, rdb, testName(t, rdb), WithCapacity(2))
	defer a.Stop()
	defer b.Stop()
	if err := b.Send(ctx, "from-b"); err != nil {
		t.Fatal(err)
	}
	select {
	case v := <-a.C():
		t.Fatalf("unexpected value on a: %q", v)
	case v := <-b.C():
		if v != "from-b" {
			t.Fatalf("v=%q", v)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("select timed out")
	}
}

func TestProxyClosesOnDrain(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	ch, _ := New[int](ctx, rdb, testName(t, rdb), WithCapacity(4))
	defer ch.Stop()
	ch.Send(ctx, 1)
	ch.Send(ctx, 2)
	if err := ch.Close(ctx); err != nil {
		t.Fatal(err)
	}
	var got []int
	for v := range ch.C() { // ends only if the proxy chan closes
		got = append(got, v)
	}
	if len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("got %v", got)
	}
}

func TestPumpAcksOnConsume(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, _ := New[string](ctx, rdb, name, WithCapacity(2))
	defer ch.Stop()
	ch.Send(ctx, "a")
	proxy := ch.C()
	// Pump claims the value but must not ack until the app reads it.
	eventually(t, 2*time.Second, "pump claims entry", func() bool {
		p, err := rdb.XPending(ctx, "rchan:{"+name+"}:s", "recv").Result()
		return err == nil && p.Count == 1
	})
	if n, _ := rdb.XLen(ctx, "rchan:{"+name+"}:s").Result(); n != 1 {
		t.Fatalf("XLEN=%d, want 1 while unconsumed", n)
	}
	if v := <-proxy; v != "a" {
		t.Fatalf("v=%q", v)
	}
	eventually(t, 2*time.Second, "acked after consume", func() bool {
		n, _ := rdb.XLen(ctx, "rchan:{"+name+"}:s").Result()
		return n == 0
	})
}

func TestStopClosesProxy(t *testing.T) {
	rdb := testClient(t)
	ch, _ := New[int](context.Background(), rdb, testName(t, rdb), WithCapacity(2))
	proxy := ch.C()
	ch.Stop()
	select {
	case _, ok := <-proxy:
		if ok {
			t.Fatal("expected closed proxy after Stop")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("proxy not closed after Stop")
	}
}