sendc_test.go

package rchan

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

func TestSendCBufferedRoundtrip(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, err := New[string](ctx, rdb, name, WithCapacity(8))
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	sc := ch.SendC()
	go func() {
		sc <- "a"
		sc <- "b"
		close(sc) // must map to Close()
	}()
	rch, err := New[string](ctx, rdb, name, WithCapacity(8))
	if err != nil {
		t.Fatal(err)
	}
	defer rch.Stop()
	rctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	for _, want := range []string{"a", "b"} {
		v, ok, err := rch.Recv(rctx)
		if err != nil || !ok || v != want {
			t.Fatalf("want %q: v=%q ok=%v err=%v", want, v, ok, err)
		}
	}
	// close(sc) propagated: drained receiver sees ok=false.
	if _, ok, err := rch.Recv(rctx); ok || err != nil {
		t.Fatalf("after close(sc): ok=%v err=%v, want ok=false err=nil", ok, err)
	}
}

// TestSendCRendezvousGatesOnReader is the reader-gate contract: with no
// receiver anywhere, sc <- v must stay blocked; the moment a receiver is
// present the handoff completes.
func TestSendCRendezvousGatesOnReader(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, err := New[string](ctx, rdb, name) // capacity 0
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	sc := ch.SendC()
	sent := make(chan struct{})
	go func() {
		sc <- "gated"
		close(sent)
	}()
	// No reader beacon exists: the native send must still be blocked.
	select {
	case <-sent:
		t.Fatal("sc <- v completed with no receiver present")
	case <-time.After(1500 * time.Millisecond):
	}
	// A receiver arrives: beacon + pickup unblock the native send.
	rch, err := New[string](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	defer rch.Stop()
	rctx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	v, ok, err := rch.Recv(rctx)
	if err != nil || !ok || v != "gated" {
		t.Fatalf("v=%q ok=%v err=%v", v, ok, err)
	}
	select {
	case <-sent:
	case <-time.After(10 * time.Second):
		t.Fatal("native send still blocked after pickup")
	}
}

func TestLenCap(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	ch, err := New[string](ctx, rdb, testName(t, rdb), WithCapacity(4))
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	if ch.Cap() != 4 {
		t.Fatalf("Cap()=%d, want 4", ch.Cap())
	}
	if n, err := ch.Len(ctx); err != nil || n != 0 {
		t.Fatalf("Len=%d err=%v, want 0", n, err)
	}
	ch.Send(ctx, "a")
	ch.Send(ctx, "b")
	if n, err := ch.Len(ctx); err != nil || n != 2 {
		t.Fatalf("Len=%d err=%v, want 2", n, err)
	}
	if _, _, err := ch.Recv(ctx); err != nil {
		t.Fatal(err)
	}
	if n, err := ch.Len(ctx); err != nil || n != 1 {
		t.Fatalf("Len=%d err=%v, want 1", n, err)
	}
	rz, err := New[string](ctx, rdb, testName(t, rdb))
	if err != nil {
		t.Fatal(err)
	}
	if rz.Cap() != 0 {
		t.Fatalf("rendezvous Cap()=%d, want 0", rz.Cap())
	}
}

// TestTrySendDefaultBuffered: native try-send must take default while the
// remote buffer is full, and succeed after a slot frees.
func TestTrySendDefaultBuffered(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, err := New[string](ctx, rdb, name, WithCapacity(1))
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	if err := ch.Send(ctx, "fill"); err != nil {
		t.Fatal(err)
	}
	sc := ch.SendC()
	time.Sleep(400 * time.Millisecond) // let the pump reach its space gate
	for i := 0; i < 5; i++ {
		select {
		case sc <- "nope":
			t.Fatal("try-send succeeded on a full remote buffer")
		default:
		}
		time.Sleep(50 * time.Millisecond)
	}
	// Free the slot; the transition publish wakes the pump's gate.
	if v, _, err := ch.Recv(ctx); err != nil || v != "fill" {
		t.Fatalf("v=%q err=%v", v, err)
	}
	sent := false
	deadline := time.Now().Add(5 * time.Second)
	for !sent && time.Now().Before(deadline) {
		select {
		case sc <- "yep":
			sent = true
		default:
			time.Sleep(50 * time.Millisecond)
		}
	}
	if !sent {
		t.Fatal("try-send never succeeded after slot freed")
	}
	rctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	if v, _, err := ch.Recv(rctx); err != nil || v != "yep" {
		t.Fatalf("v=%q err=%v", v, err)
	}
}

// TestTrySendDefaultRendezvous: try-send takes default with no receiver,
// succeeds once one is parked.
func TestTrySendDefaultRendezvous(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, err := New[string](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	sc := ch.SendC()
	time.Sleep(400 * time.Millisecond) // pump parks at the reader gate
	select {
	case sc <- "nope":
		t.Fatal("try-send succeeded with no receiver anywhere")
	default:
	}
	rch, err := New[string](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	defer rch.Stop()
	got := make(chan string, 1)
	go func() {
		rctx, cancel := context.WithTimeout(ctx, 15*time.Second)
		defer cancel()
		if v, ok, err := rch.Recv(rctx); err == nil && ok {
			got <- v
		}
		close(got)
	}()
	sent := false
	deadline := time.Now().Add(10 * time.Second)
	for !sent && time.Now().Before(deadline) {
		select {
		case sc <- "yep":
			sent = true
		default:
			time.Sleep(100 * time.Millisecond)
		}
	}
	if !sent {
		t.Fatal("try-send never succeeded with a parked receiver")
	}
	if v, ok := <-got; !ok || v != "yep" {
		t.Fatalf("receiver got %q ok=%v", v, ok)
	}
}

func TestSendCRendezvousSequential(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, err := New[int](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	rch, err := New[int](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	defer rch.Stop()

	sc := ch.SendC()
	go func() {
		for i := 0; i < 5; i++ {
			sc <- i
		}
	}()
	rctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	for i := 0; i < 5; i++ {
		v, ok, err := rch.Recv(rctx)
		if err != nil || !ok || v != i {
			t.Fatalf("i=%d: v=%d ok=%v err=%v", i, v, ok, err)
		}
	}
}