SignalR Streaming — Real-Time Data Feeds

Hub method (single result):
  Client: invoke("GetWardCapacity", wardId)
  Server: returns WardCapacityDto — one response

Server-to-client streaming:
  Client: stream("StreamVitals", patientId)
  Server: yields VitalSignDto every 30 seconds until cancelled

Client-to-server streaming:
  Client: send("UploadObservations", observationStream)
  Server: receives ObservationDto items as the client sends them

// Hub method returns IAsyncEnumerable<T> — simplest streaming pattern
public sealed class VitalsMonitorHub : Hub<IVitalsMonitorClient>
{
    private readonly IVitalsRepository _vitals;

    // Stream vital signs to the connected client
    public async IAsyncEnumerable<VitalSignDto> StreamVitalSigns(
        Guid patientId,
        [EnumeratorCancellation] CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            var vital = await _vitals.GetLatestAsync(patientId, ct);
            if (vital is not null)
                yield return new VitalSignDto(
                    vital.HeartRate,
                    vital.BloodPressureSystolic,
                    vital.BloodPressureDiastolic,
                    vital.SpO2,
                    vital.Temperature,
                    vital.RecordedAt);

            // Poll every 30 seconds
            await Task.Delay(TimeSpan.FromSeconds(30), ct);
        }
    }

    // Stream INR readings for anticoagulation monitoring
    public async IAsyncEnumerable<INRReadingDto> StreamINRReadings(
        Guid patientId,
        [EnumeratorCancellation] CancellationToken ct)
    {
        await foreach (var reading in _vitals
            .GetINRStreamAsync(patientId, ct)
            .WithCancellation(ct))
        {
            yield return new INRReadingDto(
                reading.Value,
                reading.TherapeuticRange,
                reading.IsInRange,
                reading.Timestamp);
        }
    }
}

const stream = connection.stream("StreamVitalSigns", patientId);
stream.subscribe({
    next:     vital => updateVitalsDisplay(vital),
    error:    err   => console.error("Stream error:", err),
    complete: ()    => console.log("Stream ended"),
});

// Stop streaming
const subscription = connection.stream("StreamVitalSigns", patientId)
    .subscribe(/* ... */);
subscription.dispose();  // triggers CancellationToken on server

// ChannelReader — more control over buffering and back-pressure
public ChannelReader<DrugOrderUpdateDto> StreamDrugOrderUpdates(
    string wardId,
    CancellationToken ct)
{
    var channel = Channel.CreateUnbounded<DrugOrderUpdateDto>();

    _ = Task.Run(async () =>
    {
        try
        {
            await foreach (var update in _pharmacy
                .GetWardUpdatesAsync(wardId, ct))
            {
                await channel.Writer.WriteAsync(update, ct);
            }
        }
        catch (OperationCanceledException) { /* normal cancellation */ }
        finally
        {
            channel.Writer.Complete();
        }
    }, ct);

    return channel.Reader;
}

// Server receives items streamed from the client
public async Task UploadObservations(
    IAsyncEnumerable<ObservationDto> observations,
    CancellationToken ct)
{
    await foreach (var observation in observations.WithCancellation(ct))
    {
        // Process each observation as it arrives
        await _observationService.SaveAsync(
            Context.ConnectionId,
            observation,
            ct);
    }

    // All observations received — finalize
    await Clients.Caller.UploadCompleted(new UploadResultDto(
        TotalSaved: count,
        Timestamp: DateTime.UtcNow));
}

// Stream observations from client to server
const observationSubject = new signalR.Subject();
connection.send("UploadObservations", observationSubject);

// Send observations as they become available
observationSubject.next({ heartRate: 72, timestamp: new Date() });
observationSubject.next({ heartRate: 74, timestamp: new Date() });
observationSubject.complete();  // signals end of stream

Polling (setInterval approach):
  Client calls the server every N seconds
  Every poll is a full HTTP request → response
  N seconds of latency in the best case
  Wasted requests when no data changes

SignalR streaming:
  Server pushes data as it changes
  No wasted requests — only sends when data exists
  Near-zero latency (sub-second)
  One persistent WebSocket connection (cheaper than repeated HTTP)

Use streaming for:
  ✓ Real-time vital signs monitoring
  ✓ Drug order status updates
  ✓ Alert/alarm feeds
  ✓ Live ward capacity

Use polling for:
  ✓ Non-real-time data (reports, audit logs)
  ✓ Client environments that do not support WebSockets
  ✓ Simple data that changes infrequently

// The CancellationToken is cancelled when:
// 1. Client calls subscription.dispose() / stream.cancel()
// 2. Client disconnects
// 3. Request times out

public async IAsyncEnumerable<VitalSignDto> StreamVitalSigns(
    Guid patientId,
    [EnumeratorCancellation] CancellationToken ct)
{
    while (!ct.IsCancellationRequested)
    {
        VitalSignDto? vital = null;
        try
        {
            vital = await _vitals.GetLatestAsync(patientId, ct);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            // Log but continue streaming — do not crash the stream on transient errors
            _logger.LogWarning(ex, "Failed to fetch vitals for {PatientId}", patientId);
        }

        if (vital is not null)
            yield return vital;

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