gRPC Clients and Streaming

syntax = "proto3";

package shop.inventory.v1;

option csharp_namespace = "Shop.Inventory.V1";

import "google/protobuf/timestamp.proto";

service Inventory {
  rpc GetStock (StockRequest) returns (StockReply);
  rpc WatchStock (WatchStockRequest) returns (stream StockChanged);
  rpc ImportStockCounts (stream StockCount) returns (ImportSummary);
}

message StockRequest {
  string sku = 1;
}

message StockReply {
  string sku = 1;
  int32 available = 2;
  int32 reserved = 3;
}

message WatchStockRequest {
  string warehouse_code = 1;
}

message StockChanged {
  string sku = 1;
  int32 available = 2;
  int32 reserved = 3;
  google.protobuf.Timestamp changed_at = 4;
}

message StockCount {
  string warehouse_code = 1;
  string sku = 2;
  int32 quantity = 3;
}

message ImportSummary {
  int32 accepted = 1;
  int32 rejected = 2;
}

using Grpc.Core;
using Shop.Inventory.V1;

public interface IStockChanges
{
    IAsyncEnumerable<StockChanged> SubscribeAsync(string warehouseCode, CancellationToken cancellationToken);
}

public interface IStockCounts
{
    Task SetCountAsync(string warehouseCode, string sku, int quantity, CancellationToken cancellationToken);
}

public sealed class InventoryService(
    IStockChanges changes, IStockCounts counts, IHostApplicationLifetime lifetime, ILogger<InventoryService> logger)
    : Inventory.InventoryBase
{
    public override async Task WatchStock(
        WatchStockRequest request, IServerStreamWriter<StockChanged> responseStream, ServerCallContext context)
    {
        // context.CancellationToken fires when the client cancels or disposes the call or its deadline passes.
        // On shutdown it fires only after ShutdownTimeout, so link ApplicationStopping to end the feed promptly.
        using var stop = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, lifetime.ApplicationStopping);
        try
        {
            await foreach (var change in changes.SubscribeAsync(request.WarehouseCode, stop.Token))
            {
                // One write at a time: the writer throws if a second WriteAsync starts before the first finishes.
                await responseStream.WriteAsync(change, stop.Token);
            }
        }
        catch (OperationCanceledException) when (lifetime.ApplicationStopping.IsCancellationRequested)
        {
            // Returning ends the stream with status OK; the client reconnects to another instance.
        }
    }

    public override async Task<ImportSummary> ImportStockCounts(
        IAsyncStreamReader<StockCount> requestStream, ServerCallContext context)
    {
        var summary = new ImportSummary();
        await foreach (var count in requestStream.ReadAllAsync(context.CancellationToken))
        {
            if (string.IsNullOrWhiteSpace(count.Sku) || count.Quantity < 0)
            {
                summary.Rejected++;
                continue;
            }

            await counts.SetCountAsync(count.WarehouseCode, count.Sku, count.Quantity, context.CancellationToken);
            summary.Accepted++;
        }

        logger.LogInformation("Imported {Accepted} stock counts, rejected {Rejected}", summary.Accepted, summary.Rejected);
        return summary;
    }
}

using Grpc.Core;
using Grpc.Net.Client;
using Grpc.Net.Client.Balancer;
using Grpc.Net.Client.Configuration;
using Shop.Inventory.V1;

// By default DNS is re-resolved only after a connection fails; refresh it so scaled-out pods get traffic.
var services = new ServiceCollection();
services.AddSingleton<ResolverFactory>(new DnsResolverFactory(refreshInterval: TimeSpan.FromSeconds(30)));

// dns:/// resolves every pod IP behind a headless Service and spreads calls across them.
using var channel = GrpcChannel.ForAddress("dns:///inventory-grpc.shop.svc.cluster.local:8081", new GrpcChannelOptions
{
    Credentials = ChannelCredentials.Insecure,
    ServiceConfig = new ServiceConfig { LoadBalancingConfigs = { new RoundRobinConfig() } },
    ServiceProvider = services.BuildServiceProvider()
});
var client = new Inventory.InventoryClient(channel);

(string Sku, int Quantity)[] counts =
[
    ("KETTLE-BLUE-1L", 140),
    ("MUG-WHITE-350", 820),
    ("TEAPOT-CAST-IRON", 36)
];

using var call = client.ImportStockCounts(deadline: DateTime.UtcNow.AddMinutes(2));
foreach (var (sku, quantity) in counts)
{
    await call.RequestStream.WriteAsync(new StockCount { WarehouseCode = "AMS-1", Sku = sku, Quantity = quantity });
}
await call.RequestStream.CompleteAsync();

var summary = await call;
Console.WriteLine($"Accepted {summary.Accepted}, rejected {summary.Rejected}");