Function TestBrokerFlowControl

pkg/streamclient/streambroker_test.go:132–177

(t *testing.T)

}

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

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

	largeData := make([]byte, 100)
	for i := range largeData {
		largeData[i] = byte(i)
	}

	writeDone := make(chan error)
	go func() {
		_, err := writer.Write(largeData)
		writeDone <- err
	}()

	received := make([]byte, 0, 100)
	buf := make([]byte, 20)
	for len(received) < len(largeData) {
		n, err := reader.Read(buf)
		if err != nil {
			t.Fatalf("Read failed: %v", err)
		}
		received = append(received, buf[:n]...)
	}

	select {
	case err := <-writeDone:
		if err != nil {
			t.Fatalf("Write failed: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("Write didn't complete in time")
	}

	if !bytes.Equal(received, largeData) {
		t.Fatal("Received data doesn't match sent data")
	}

	writer.Close()
}

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