Method TestInflightStreamClosing

internal/transport/transport_test.go:691–732

TestInflightStreamClosing ensures that closing in-flight stream sends status error to concurrent stream reader.

(t *testing.T)

// TestInflightStreamClosing ensures that closing in-flight stream
// sends status error to concurrent stream reader.
func (s) TestInflightStreamClosing(t *testing.T) {
	serverConfig := &ServerConfig{
		BufferPool: mem.DefaultBufferPool(),
	}
	copts := ConnectOptions{
		BufferPool: mem.DefaultBufferPool(),
	}
	server, client, cancel := setUpWithOptions(t, 0, serverConfig, suspended, copts)
	defer cancel()
	defer server.stop()
	defer client.Close(fmt.Errorf("closed manually by test"))

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	stream, err := client.NewStream(ctx, &CallHdr{}, nil)
	if err != nil {
		t.Fatalf("Client failed to create RPC request: %v", err)
	}

	donec := make(chan struct{})
	serr := status.Error(codes.Internal, "client connection is closing")
	go func() {
		defer close(donec)
		if _, err := stream.readTo(make([]byte, defaultWindowSize)); err != serr {
			t.Errorf("unexpected Stream error %v, expected %v", err, serr)
		}
	}()

	// should unblock concurrent stream.Read
	stream.Close(serr)

	// wait for stream.Read error
	timeout := time.NewTimer(5 * time.Second)
	select {
	case <-donec:
		if !timeout.Stop() {
			<-timeout.C
		}
	case <-timeout.C:
		t.Fatalf("Test timed out, expected a status error.")
	}
}

// Tests that when streamID > MaxStreamId, the current client transport drains.
func (s) TestClientTransportDrainsAfterStreamIDExhausted(t *testing.T) {