Process Stream data from gRPC Server using Flutter/Dart

In the earlier post, we addressed how to create a simple Grpc server in C# and consume it using a client developed using Flutter/Dart.

We will now take the next step and address how to create expose a streaming remote method in the server and consume the same in the client.

Streaming method in Server (C#)

In the first part, we will create the streaming method in the Server. We will continue with the earlier example and use the same server which we developed using C#. We will begin by updating our Protocol Buffer definition to include a method which returns a stream of data.

service Instrument{
	rpc Subscribe (RawDataRequest) returns (stream RawDataResponse);
}

message RawDataRequest{
	int32 maxItems = 1;
}

message RawDataResponse{
	int32 id  = 1;
	string description = 2;
}

As you can notice in the code above, we have added a Subscribe method which accepts a single parameter RawDataRequest and returns a stream of RawDataResponse. The stream keyword indicates the response is a collection of data and not a single entity.

We will now build the solution to ensure the corresponding C# files are generated. Once the auto-generated files are generated/updated, we can now update our InstrumentService defined in the earlier post to include the new Subscribe method.

public override async Task Subscribe(RawDataRequest request, IServerStreamWriter<RawDataResponse> responseStream, ServerCallContext context)
{
    for(int i = 0; i < request.MaxItems; i++)
    {
        await Task.Delay(1000);
        await responseStream.WriteAsync(new RawDataResponse
        {
            Id = i,
            Description = $"Description {i}"
        });
    }
}

For the sake of example, we will keep it simple. The method would accept a maxItems, which indicates the number of data the method would return. The method inturn returns a stream of RawDataResponse by writing to the output stream using the responseStream.WriteAsync method.

That would complete our requirement for server. We can move towards the client codebase and understand how to process a stream of data using dart.

Client Code (Dart/Flutter)

The client is supposed to invoke the remote method call in the server and process the stream of response data by keeping the connection alive. Before we begin defining the method for processing the stream of data, ensure you have copied over the latest definition of Protocol buffers and use the protoc compiler to generate the corresponding dart files.

protoc --dart_out=grpc:lib\generated -Iprotos protos\greet.proto

For the sake of this example, we will process the stream of data and display the items in a List. We will use a List<> to store the items returned by the method and use it to fill the ListView.

List<RawDataResponse> _streamData = <RawDataResponse>[
  ];

  ListView.builder(
                  padding: const EdgeInsets.all(8),
                  itemCount: _streamData.length,
                  itemBuilder: (BuildContext context, int index) {
                    return Container(
                        height: 50,
                        margin: const EdgeInsets.all(2),
                        child: Text(
                            '[${_streamData[index].id}] : ${_streamData[index].description} '));
                  })

With that in place, the final step required is to fill the List _streamData with the values returned from the Server.

Future<void> subscribeStream() async {
    final channel = ClientChannel('localhost',
        port: 5280,
        options:
            const ChannelOptions(credentials: ChannelCredentials.insecure()));

    stub = InstrumentClient(channel,
        options: CallOptions(timeout: constDuration(seconds: 30)));

    try {
      var subscribeRequest = RawDataRequest(maxItems: 10);
      var response = stub.subscribe(subscribeRequest);

      awaitfor (var data in response) {
        setState(() {
          _streamData.add(data);
        });
      }
    } catch (e) {
      print(e);
    }
  }

That’s all we require to call a streaming gRPC method using Flutter/Dart. Complete source code of the example is available in my Github.