Server stream service - ResponseStatus

#1

Hello,
I have implemented a server stream service, as described here:
https://docs.servicestack.net/grpc#implementing-server-stream-services

Dtos:

[DataContract]
public class StreamDocumens : IHasBearerToken, IReturn<StreamDocumentsResponse>
{
    [DataMember(Order = 1)] 
    public string? BearerToken { get; set; }
}

[DataContract]
public class StreamDocumentsResponse: IHasResponseStatus
{
    [DataMember(Order = 1)] 
    public Guid Id { get; set; }

    [DataMember(Order = 2)] 
    public string? InvoiceNumber { get; set; }

    [DataMember(Order = 3)] 
    public ResponseStatus? ResponseStatus { get; set; }
}

On the client-side is dart (flutter):

try {
    final stream = grpcCient.streamDocuments(request);

    await for (var r in stream) {
      if (r.id.isEmpty) continue;

      print("Response received | ${r.invoiceNumber}");
}
on GrpcError catch (e) {
    printError(e);
}

Everything is working but in case of exception, is there a way to access ResponseStatus property?
Also, I’m not sure, will the ResponseStatus property be automatically populated, as in normal ServiceStack service?

#2

You’d only get a ServiceStack API ResponseStatus Exception for the entire request if using the C#/.NET generic GrpcServiceClient, e.g:

var client = new GrpcServiceClient(baseUrl);
try
{
    await foreach (var file in client.StreamAsync(request))
    {
        //...
    }
}
catch (WebServiceException ex)
{
   //...
}

Not when using any other gRPC client.

If you want to return an error for a single item you would need to populate it the ResponseStatus yourself in one of the returned response DTOs:

public class StreamFileService : Service, IStreamService<StreamFiles,FileContent>
{
    public async IAsyncEnumerable<FileContent> Stream(StreamFiles request, CancellationToken cancel = default)
    {
        var i = 0;
        var paths = request.Paths ?? TypeConstants.EmptyStringList;
        while (!cancel.IsCancellationRequested)
        {
            var file = VirtualFileSources.GetFile(paths[i]);
            var bytes = file?.GetBytesContentsAsBytes();
            var to = file != null
                ? new FileContent {
                    Name = file.Name,
                    Type = MimeTypes.GetMimeType(file.Extension),
                    Body = bytes,
                    Length = bytes.Length,
                }
                : new FileContent {
                    Name = paths[i],
                    ResponseStatus = new ResponseStatus {
                        ErrorCode = nameof(HttpStatusCode.NotFound),
                        Message = "File does not exist",
                    }
                };
            
            yield return to;

            if (++i >= paths.Count)
                yield break;
        }
    }
}

See the Implementing Server Stream Services example.

#3

Thank you for very quick response as always.

1 Like
#4

It turns out that my exceptions on the server are not coming from service.
Error is generated on deserializing grpc request:

ProtoBuf.ProtoException: Invalid wire-type (String); this usually means you have over-written a file without truncating or setting the length; see https://stackoverflow.com/q/215297>
at ProtoBuf.ProtoReader.State.ThrowProtoException(String message) in /_/src/protobuf-net.Core/ProtoReader.State.ReadMethods.cs:line 764
at ProtoBuf.ProtoReader.State.ThrowWireTypeException() in /_/src/protobuf-net.Core/ProtoReader.State.ReadMethods.cs:line 759
at proto_100(State& , AppStreamUpPosDokument )
at ProtoBuf.ProtoReader.State.ReadMessage[TSerializer,T](SerializerFeatures features, T value, TSerializer& serializer)
at ProtoBuf.ProtoReader.State.FillBuffer[TSerializer,T](SerializerFeatures features, TSerializer& serializer, T initialValue)
at ProtoBuf.Serializers.RepeatedSerializer`2.ReadRepeated(State& state, SerializerFeatures features, TCollection values, ISerializer`1 serializer)
at proto_54(State& , AppStreamUpPosDokumenta )
at ProtoBuf.ProtoReader.State.ReadAsRoot[T](T value, ISerializer`1 serializer)
at ProtoBuf.ProtoReader.State.DeserializeRoot[T](T value, ISerializer`1 serializer)
at ProtoBuf.Internal.DynamicStub.ConcreteStub`1.TryDeserializeRoot(TypeModel model, State& state, Object& value, Boolean autoCreate)
at ProtoBuf.Meta.TypeModel.DeserializeRootAny(State& state, Type type, Object value, Boolean autoCreate)

Is there a way to inspect grpc requests on / before deserialization?
I have tried the error request logger, but there are no errors there.

#5

This looks like a serialization exception of an invalid type, if this is from the streaming service try catching the exception in the service then returning the populated Response DTO instead as done in the example above.

#6

I did tried to do that, but request never reach service. Exception is thrown before, that is why I asked, where to inspect request before service.

#7

This Exception happens within the ProtoBuf serializer, i.e. the binary request that’s sent can’t be deserialized into the DTO Type. The actual deserialization happens in ASP .NET Core’s gRPC MapGrpcService implementation so it’s not something you can debug in ServiceStack.

If it doesn’t hit your debug breakpoint in your Service implementation then it must be an issue deserializing into the Request DTO, in which case I would start with an empty DTO and add properties until you find the property that’s causing the issue.