Bidirectional Streaming RPC — Full-Duplex gRPC

Client: send PatientMonitorRequest  → Server processes
Server: send PatientStatusResponse  → Client processes
Client: send another request (any time)
Server: send another response (any time)

Both sides read and write concurrently
Stream ends when: either side completes their write stream,
                  deadline is exceeded, or either side cancels

service PatientMonitorService {
    // Both sides stream independently
    rpc MonitorPatients (stream PatientMonitorRequest)
        returns (stream PatientStatusResponse);
}

message PatientMonitorRequest {
    oneof action {
        SubscribeRequest   subscribe   = 1;
        UnsubscribeRequest unsubscribe = 2;
        AcknowledgeRequest acknowledge = 3;
    }
}

message SubscribeRequest   { string patient_id = 1; }
message UnsubscribeRequest { string patient_id = 1; }
message AcknowledgeRequest { string alert_id   = 1; }

message PatientStatusResponse {
    string patient_id  = 1;
    string status_type = 2;  // "vitals_updated", "alert_triggered", "order_changed"
    bytes  payload     = 3;  // serialized type-specific data
    int64  timestamp   = 4;
}

public sealed class PatientMonitorGrpcService
    : PatientMonitorService.PatientMonitorServiceBase
{
    private readonly IVitalsRepository _vitals;
    private readonly IAlertService     _alerts;

    public override async Task MonitorPatients(
        IAsyncStreamReader<PatientMonitorRequest> requestStream,
        IServerStreamWriter<PatientStatusResponse> responseStream,
        ServerCallContext context)
    {
        var ct              = context.CancellationToken;
        var subscribedIds   = new ConcurrentHashSet<Guid>();

        // Read client messages concurrently with writing responses
        // Task A: process incoming requests (subscribe/unsubscribe)
        var readTask = Task.Run(async () =>
        {
            await foreach (var request in requestStream.ReadAllAsync(ct))
            {
                switch (request.ActionCase)
                {
                    case PatientMonitorRequest.ActionOneofCase.Subscribe:
                        if (Guid.TryParse(request.Subscribe.PatientId, out var subId))
                            subscribedIds.Add(subId);
                        break;

                    case PatientMonitorRequest.ActionOneofCase.Unsubscribe:
                        if (Guid.TryParse(request.Unsubscribe.PatientId, out var unsubId))
                            subscribedIds.TryRemove(unsubId);
                        break;

                    case PatientMonitorRequest.ActionOneofCase.Acknowledge:
                        if (Guid.TryParse(request.Acknowledge.AlertId, out var alertId))
                            await _alerts.AcknowledgeAsync(alertId, ct);
                        break;
                }
            }
        }, ct);

        // Task B: push status updates for subscribed patients
        var writeTask = Task.Run(async () =>
        {
            while (!ct.IsCancellationRequested)
            {
                foreach (var patientId in subscribedIds)
                {
                    var status = await _vitals.GetLatestStatusAsync(patientId, ct);
                    if (status is not null)
                    {
                        await responseStream.WriteAsync(new PatientStatusResponse
                        {
                            PatientId  = patientId.ToString(),
                            StatusType = "vitals_updated",
                            Payload    = ByteString.CopyFrom(
                                JsonSerializer.SerializeToUtf8Bytes(status)),
                            Timestamp  = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
                        }, ct);
                    }
                }

                await Task.Delay(TimeSpan.FromSeconds(15), ct);
            }
        }, ct);

        // Wait for both to complete
        try
        {
            await Task.WhenAll(readTask, writeTask);
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested) { }
    }
}

using var channel = GrpcChannel.ForAddress("https://monitor.systemforge.internal");
var client = new PatientMonitorService.PatientMonitorServiceClient(channel);

using var cts  = new CancellationTokenSource();
using var call = client.MonitorPatients(cancellationToken: cts.Token);

// Read responses concurrently
var readTask = Task.Run(async () =>
{
    await foreach (var status in call.ResponseStream.ReadAllAsync(cts.Token))
    {
        switch (status.StatusType)
        {
            case "vitals_updated":
                var vitals = JsonSerializer.Deserialize<VitalsDto>(status.Payload.ToArray());
                UpdateVitalsDisplay(status.PatientId, vitals!);
                break;

            case "alert_triggered":
                var alert = JsonSerializer.Deserialize<AlertDto>(status.Payload.ToArray());
                ShowAlert(alert!);
                break;
        }
    }
}, cts.Token);

// Subscribe to patients
await call.RequestStream.WriteAsync(new PatientMonitorRequest
{
    Subscribe = new SubscribeRequest { PatientId = patientId.ToString() }
});

// Acknowledge an alert
await call.RequestStream.WriteAsync(new PatientMonitorRequest
{
    Acknowledge = new AcknowledgeRequest { AlertId = alertId.ToString() }
});

// Unsubscribe when done
await call.RequestStream.WriteAsync(new PatientMonitorRequest
{
    Unsubscribe = new UnsubscribeRequest { PatientId = patientId.ToString() }
});

await call.RequestStream.CompleteAsync();
await readTask;

Bidirectional streaming is the most complex gRPC pattern. Use it when:
  ✓ Client needs to send commands AND receive real-time updates over the same connection
  ✓ Collaborative workflows (multiple users editing the same record)
  ✓ Command-response over long-lived connection (chat-style protocol)
  ✓ Interactive monitoring where the client controls what it monitors

Consider simpler alternatives:
  Server streaming: if the client only subscribes once and receives data
  SignalR: if you need browser WebSocket support with group broadcasting
  Unary + polling: if real-time is not critical and request patterns are simple

// IServerStreamWriter is NOT thread-safe for concurrent writes
// If both tasks write concurrently, use a Channel as a buffer

var responseChannel = Channel.CreateUnbounded<PatientStatusResponse>();

// Multiple sources write to the channel
_ = Task.Run(async () =>
{
    while (true)
        await responseChannel.Writer.WriteAsync(await GetVitalStatusAsync());
});

// Single consumer writes to the gRPC stream (thread-safe single writer)
await foreach (var response in responseChannel.Reader.ReadAllAsync(ct))
    await responseStream.WriteAsync(response, ct);