Method UnconstrainedStreamingCall

benchmark/benchmark.go:139–210

(stream testgrpc.BenchmarkService_StreamingCallServer, preloadMsgSize int)

}

func (s *testServer) UnconstrainedStreamingCall(stream testgrpc.BenchmarkService_StreamingCallServer, preloadMsgSize int) error {
	maxSleep := 0
	if md, ok := metadata.FromIncomingContext(stream.Context()); ok && len(md[UnconstrainedStreamingDelayHeader]) != 0 {
		val := md[UnconstrainedStreamingDelayHeader][0]
		d, err := time.ParseDuration(val)
		if err != nil {
			return fmt.Errorf("can't parse %q header: %s", UnconstrainedStreamingDelayHeader, err)
		}
		maxSleep = int(d)
	}

	in := new(testpb.SimpleRequest)
	// Receive a message to learn response type and size.
	err := stream.RecvMsg(in)
	if err == io.EOF {
		// read done.
		return nil
	}
	if err != nil {
		return err
	}

	response := &testpb.SimpleResponse{
		Payload: new(testpb.Payload),
	}
	setPayload(response.Payload, in.ResponseType, int(in.ResponseSize))

	preloadedResponse := &grpc.PreparedMsg{}
	if preloadMsgSize > 0 {
		if err := preloadedResponse.Encode(stream, response); err != nil {
			return err
		}
	}

	go func() {
		for {
			// Using RecvMsg rather than Recv to prevent reallocation of SimpleRequest.
			err := stream.RecvMsg(in)
			switch status.Code(err) {
			case codes.Canceled:
				return
			case codes.OK:
			default:
				log.Fatalf("server recv error: %v", err)
			}
		}
	}()

	go func() {
		for {
			if maxSleep > 0 {
				time.Sleep(time.Duration(rand.IntN(maxSleep)))
			}
			var err error
			if preloadMsgSize > 0 {
				err = stream.SendMsg(preloadedResponse)
			} else {
				err = stream.Send(response)