Method StreamingCall

benchmark/benchmark.go:92–137

(stream testgrpc.BenchmarkService_StreamingCallServer)

const PreloadMsgSizeHeader = "preload-msg-size"

func (s *testServer) StreamingCall(stream testgrpc.BenchmarkService_StreamingCallServer) error {
	preloadMsgSize := 0
	if md, ok := metadata.FromIncomingContext(stream.Context()); ok && len(md[PreloadMsgSizeHeader]) != 0 {
		val := md[PreloadMsgSizeHeader][0]
		var err error
		preloadMsgSize, err = strconv.Atoi(val)
		if err != nil {
			return fmt.Errorf("%q header value is not an integer: %s", PreloadMsgSizeHeader, err)
		}
	}

	if md, ok := metadata.FromIncomingContext(stream.Context()); ok && len(md[UnconstrainedStreamingHeader]) != 0 {
		return s.UnconstrainedStreamingCall(stream, preloadMsgSize)
	}
	response := &testpb.SimpleResponse{
		Payload: new(testpb.Payload),
	}
	preloadedResponse := &grpc.PreparedMsg{}
	if preloadMsgSize > 0 {
		setPayload(response.Payload, testpb.PayloadType_COMPRESSABLE, preloadMsgSize)
		if err := preloadedResponse.Encode(stream, response); err != nil {
			return err
		}
	}
	in := new(testpb.SimpleRequest)
	for {
		// use ServerStream directly to reuse the same testpb.SimpleRequest object
		err := stream.(grpc.ServerStream).RecvMsg(in)
		if err == io.EOF {
			// read done.
			return nil
		}
		if err != nil {
			return err
		}
		if preloadMsgSize > 0 {
			err = stream.SendMsg(preloadedResponse)
		} else {
			setPayload(response.Payload, in.ResponseType, int(in.ResponseSize))
			err = stream.Send(response)
		}
		if err != nil {
			return err
		}
	}
}

func (s *testServer) UnconstrainedStreamingCall(stream testgrpc.BenchmarkService_StreamingCallServer, preloadMsgSize int) error {
	maxSleep := 0