Function TestBrokerCancel

pkg/streamclient/streambroker_test.go:201–223

(t *testing.T)

}

func TestBrokerCancel(t *testing.T) {
	broker1, broker2 := setupBrokerPair()

	reader, meta := broker1.CreateStreamReader("reader1", "writer1", 1024)
	writer, err := broker2.CreateStreamWriter(meta)
	if err != nil {
		t.Fatalf("CreateStreamWriter failed: %v", err)
	}

	reader.Close()

	select {
	case <-writer.GetCanceledChan():
		// Success
	case <-time.After(1 * time.Second):
		t.Fatal("Writer not notified of cancellation")
	}

	_, _, canceled := writer.GetAckState()
	if !canceled {
		t.Fatal("Writer should be in canceled state")
	}
}

func TestBrokerMultipleWrites(t *testing.T) {
	broker1, broker2 := setupBrokerPair()