Method TestRetryStreaming

test/retry_test.go:181–498

(t *testing.T)

}

func (s) TestRetryStreaming(t *testing.T) {
	req := func(b byte) *testpb.StreamingOutputCallRequest {
		return &testpb.StreamingOutputCallRequest{Payload: &testpb.Payload{Body: []byte{b}}}
	}
	res := func(b byte) *testpb.StreamingOutputCallResponse {
		return &testpb.StreamingOutputCallResponse{Payload: &testpb.Payload{Body: []byte{b}}}
	}

	largePayload, _ := newPayload(testpb.PayloadType_COMPRESSABLE, 500)

	type serverOp func(stream testgrpc.TestService_FullDuplexCallServer) error
	type clientOp func(stream testgrpc.TestService_FullDuplexCallClient) error

	// Server Operations
	sAttempts := func(n int) serverOp {
		return func(stream testgrpc.TestService_FullDuplexCallServer) error {
			const key = "grpc-previous-rpc-attempts"
			md, ok := metadata.FromIncomingContext(stream.Context())
			if !ok {
				return status.Errorf(codes.Internal, "server: no header metadata received")
			}
			if got := md[key]; len(got) != 1 || got[0] != strconv.Itoa(n) {
				return status.Errorf(codes.Internal, "server: metadata = %v; want <contains %q: %q>", md, key, n)
			}
			return nil
		}
	}
	sReq := func(b byte) serverOp {
		return func(stream testgrpc.TestService_FullDuplexCallServer) error {
			want := req(b)
			if got, err := stream.Recv(); err != nil || !proto.Equal(got, want) {
				return status.Errorf(codes.Internal, "server: Recv() = %v, %v; want %v, <nil>", got, err, want)
			}
			return nil
		}
	}
	sReqPayload := func(p *testpb.Payload) serverOp {
		return func(stream testgrpc.TestService_FullDuplexCallServer) error {
			want := &testpb.StreamingOutputCallRequest{Payload: p}
			if got, err := stream.Recv(); err != nil || !proto.Equal(got, want) {
				return status.Errorf(codes.Internal, "server: Recv() = %v, %v; want %v, <nil>", got, err, want)
			}
			return nil
		}
	}
	sHdr := func() serverOp {
		return func(stream testgrpc.TestService_FullDuplexCallServer) error {
			return stream.SendHeader(metadata.Pairs("test_header", "test_value"))
		}
	}
	sRes := func(b byte) serverOp {
		return func(stream testgrpc.TestService_FullDuplexCallServer) error {
			msg := res(b)
			if err := stream.Send(msg); err != nil {
				return status.Errorf(codes.Internal, "server: Send(%v) = %v; want <nil>", msg, err)
			}
			return nil
		}