Transient fault handling in gRPC with retries in C#

gRPC calls can be interrupted by transient faults. Transient faults include:

  • Momentary loss of network connectivity.
  • Temporary unavailability of a service.
  • Timeouts due to server load.

When a gRPC call is interrupted, the client throws an RpcException with details about the error. The client app must catch the exception and choose how to handle the error.

var client = new Greeter.GreeterClient(channel);
try
{
    var response = await client.SayHelloAsync(
        new HelloRequest { Name = ".NET" });

    Console.WriteLine("From server: " + response.Message);
}
catch (RpcException ex)
{
    // Write logic to inspect the error and retry
    // if the error is from a transient fault.
}

Duplicating retry logic throughout an app is verbose and error-prone. Fortunately, the .NET gRPC client now has built-in support for automatic retries. Retries are centrally configured on a channel, and there are many options for customizing retry behavior using a RetryPolicy.

var defaultMethodConfig = new MethodConfig
{
    Names = { MethodName.Default },
    RetryPolicy = new RetryPolicy
    {
        MaxAttempts = 5,
        InitialBackoff = TimeSpan.FromSeconds(1),
        MaxBackoff = TimeSpan.FromSeconds(5),
        BackoffMultiplier = 1.5,
        RetryableStatusCodes = { StatusCode.Unavailable }
    }
};

// Clients created with this channel will automatically retry failed calls.
var channel = GrpcChannel.ForAddress("https://localhost:5001", new GrpcChannelOptions
{
    ServiceConfig = new ServiceConfig { MethodConfigs = { defaultMethodConfig } }
});

References
https://devblogs.microsoft.com/dotnet/grpc-in-dotnet-6/
https://docs.microsoft.com/en-us/aspnet/core/grpc/retries?view=aspnetcore-6.0

gRPC client-side load balancing in C#

The following code example configures a channel to use DNS service discovery with round-robin load balancing:

var channel = GrpcChannel.ForAddress(
    "dns:///my-example-host",
    new GrpcChannelOptions
    {
        Credentials = ChannelCredentials.Insecure,
        ServiceConfig = new ServiceConfig { LoadBalancingConfigs = { new RoundRobinConfig() } }
    });
var client = new Greet.GreeterClient(channel);

var response = await client.SayHelloAsync(new HelloRequest { Name = "world" });

References
https://devblogs.microsoft.com/dotnet/grpc-in-dotnet-6/
https://docs.microsoft.com/en-us/aspnet/core/grpc/loadbalancing?view=aspnetcore-6.0

Parallel.ForEachAsync in .NET 6

using System.Net.Http.Headers;
using System.Net.Http.Json;

var userHandlers = new []  { "users/VahidN", "users/shanselman", "users/jaredpar", "users/davidfowl" };

using HttpClient client = new()
{
    BaseAddress = new Uri("https://api.github.com"),
};
client.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue("DotNet", "6"));

ParallelOptions parallelOptions = new() { MaxDegreeOfParallelism = 3 };

await Parallel.ForEachAsync(userHandlers, parallelOptions, async (uri, token) =>
{
    var user = await client.GetFromJsonAsync<GitHubUser>(uri, token);
    Console.WriteLine($"Name: {user.Name}\nBio: {user.Bio}\n");
});

public class GitHubUser
{
    public string Name { get; set; }
    public string  Bio { get; set; }
}

References
https://www.hanselman.com/blog/parallelforeachasync-in-net-6

Dependency Injection in WPF Application using Generic HostBuilder

Creating Generic HostBuilder

Nuget :

PM> Install-Package Microsoft.Extensions.Hosting -Version 3.1.2

The HostBuilder class is available from the following namespace,

using Microsoft.Extensions.Hosting;

Initialize the Host

Within Constructor of App class(file: App.xaml.cs) please add below code to build Host,

host = new HostBuilder()
           .ConfigureServices((hostContext, services) =>
           {
 
           }).Build();

DI Container

host = new HostBuilder()
          .ConfigureServices((hostContext, services) =>
          {
              //Add business services as needed
              services.AddScoped<IEmployeeViewModel, EmployeeViewModel>();
 
              //Lets Create single tone instance of Master windows
              services.AddSingleton<EmployeeWindow>();
 
          }).Build();

Setting up OnStartup and OnExit Application events

private void OnStartup(object sender, StartupEventArgs e)
        {
            var mainWindow = host.Services.GetService<EmployeeWindow>();
            mainWindow.Show();
        }
protected override async void OnExit(ExitEventArgs e)
       {
           using (host)
           {
               await host.StopAsync();
           }
 
           base.OnExit(e);
       }

App.XAML file used as below,

<Application x:Class="WPFDesktopApp.App"
             xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
             xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
            Startup="OnStartup">
    <Application.Resources>
         
    </Application.Resources>
</Application>

References
https://www.thecodebuzz.com/dependency-injection-wpf-generic-hostbuilder-net-core/
https://laurentkempe.com/2019/09/03/WPF-and-dotnet-Generic-Host-with-dotnet-Core-3-0/

Serialize objects in file with protobuf in C#

Install-Package protobuf-net

Decorate your classes

[ProtoContract]
class Person {
    [ProtoMember(1)]
    public int Id {get;set;}
    [ProtoMember(2)]
    public string Name {get;set;}
    [ProtoMember(3)]
    public Address Address {get;set;}
}
[ProtoContract]
class Address {
    [ProtoMember(1)]
    public string Line1 {get;set;}
    [ProtoMember(2)]
    public string Line2 {get;set;}
}

Serialize your data

var person = new Person {
    Id = 12345, Name = "Fred",
    Address = new Address {
        Line1 = "Flat 1",
        Line2 = "The Meadows"
    }
};
using(var file = new FileStream(path, FileMode.Truncate)) {
    Serializer.Serialize(file, person);
    file.SetLength(file.Position);
}
if (!File.Exists(Statics.UserProtobufFile))
{
    using (var file = new FileStream(Statics.UserProtobufFile, FileMode.Create))
    {
        Serializer.Serialize(file, Statics.User);
        file.SetLength(file.Position);
    }
}
else
{
    using (var file = new FileStream(Statics.UserProtobufFile, FileMode.Truncate))
    {
        Serializer.Serialize(file, Statics.User);
        file.SetLength(file.Position);
    }
}

Deserialize your data

Person newPerson;
using (var file = File.OpenRead("person.bin")) {
    newPerson = Serializer.Deserialize<Person>(file);
}

References
https://github.com/protobuf-net/protobuf-net
https://stackoverflow.com/questions/20369302/how-to-serialize-several-objects-in-one-file-in-c-sharp-with-protobuf
https://stackoverflow.com/questions/2152978/using-protobuf-net-i-suddenly-got-an-exception-about-an-unknown-wire-type

How to use Task.WhenAll in C#

using System;
using System.Collections.Generic;
using System.Net.NetworkInformation;
using System.Threading;
using System.Threading.Tasks;

public class Example
{
   public static void Main()
   {
      int failed = 0;
      var tasks = new List<Task>();
      String[] urls = { "www.adatum.com", "www.cohovineyard.com",
                        "www.cohowinery.com", "www.northwindtraders.com",
                        "www.contoso.com" };
      
      foreach (var value in urls) {
         var url = value;
         tasks.Add(Task.Run( () => { var png = new Ping();
                                     try {
                                        var reply = png.Send(url);
                                        if (! (reply.Status == IPStatus.Success)) {
                                           Interlocked.Increment(ref failed);
                                           throw new TimeoutException("Unable to reach " + url + ".");
                                        }
                                     }
                                     catch (PingException) {
                                        Interlocked.Increment(ref failed);
                                        throw;
                                     }
                                   }));
      }
      Task t = Task.WhenAll(tasks);
      try {
         t.Wait();
      }
      catch {}   

      if (t.Status == TaskStatus.RanToCompletion)
         Console.WriteLine("All ping attempts succeeded.");
      else if (t.Status == TaskStatus.Faulted)
         Console.WriteLine("{0} ping attempts failed", failed);      
   }
}
// The example displays output like the following:
//       5 ping attempts failed
private static async Task Main(string[] args)
{
    var stopwatch = new Stopwatch();
    stopwatch.Start();
 
    // this task will take about 2.5s to complete
    var sumTask = SlowAndComplexSumAsync();
 
    // this task will take about 4s to complete
    var wordTask = SlowAndComplexWordAsync();
 
    // running them in parallel should take about 4s to complete
    await Task.WhenAll(sumTask, wordTask);

    // The elapsed time at this point will only be about 4s
    Console.WriteLine("Time elapsed when both complete..." + stopwatch.Elapsed);
 
    // These lines are to prove the outputs are as expected,
    // i.e. 300 for the complex sum and "ABC...XYZ" for the complex word
    Console.WriteLine("Result of complex sum = " + sumTask.Result);
    Console.WriteLine("Result of complex letter processing " + wordTask.Result);
 
    Console.Read();
}

References
https://docs.microsoft.com/en-us/dotnet/api/system.threading.tasks.task.whenall?view=net-5.0
https://jeremylindsayni.wordpress.com/2019/03/11/using-async-await-and-task-whenall-to-improve-the-overall-speed-of-your-c-code/

Bidirectional Streaming with gRPC in C#

.proto

syntax = "proto3";

import "google/protobuf/timestamp.proto";

option csharp_namespace = "GrpcBidiStearming";

package greet;

service Greeter {
  rpc SayHello (stream HelloRequest) returns (stream HelloReply);
}

message HelloRequest {
  string name = 1;
  google.protobuf.Timestamp timestamp = 2;
}

message HelloReply {
  string message = 1;
  google.protobuf.Timestamp timestamp = 2;
}

Server

public override async Task SayHello(IAsyncStreamReader<HelloRequest> requestStream,
    IServerStreamWriter<HelloReply> responseStream, ServerCallContext context)
{
    while (await requestStream.MoveNext() && !context.CancellationToken.IsCancellationRequested)
    {
        Console.WriteLine("Request : {0}, Timestamp : {1}", requestStream.Current.Name,
            requestStream.Current.Timestamp);

        await responseStream.WriteAsync(new HelloReply
        {
            Message = $"Hello {requestStream.Current.Name}",
            Timestamp = Timestamp.FromDateTime(DateTime.UtcNow)
        });
        
        Console.WriteLine("Response : {0}, Timestamp : {1}", requestStream.Current.Name,
            requestStream.Current.Timestamp);
    }
}

Client

static async Task Main(string[] args)
{
    using var channel = GrpcChannel.ForAddress("http://localhost:5000");
    var client = new Greeter.GreeterClient(channel);

    CancellationToken cancellationToken = new CancellationToken(false);
    var response = client.SayHello(new CallOptions(null, null, cancellationToken));


    for (int i = 0; i < 100; i++)
    {
        var timestamp = Timestamp.FromDateTime(DateTime.UtcNow);
        await response.RequestStream.WriteAsync(new HelloRequest
        {
            Name = "Mahmood",
            Timestamp = timestamp
        });

        Console.WriteLine(String.Format("Request : {0}", timestamp.ToString()));

        await response.ResponseStream.MoveNext();

        Console.WriteLine(String.Format("Response : {0}, Timestamp : {1}",
            response.ResponseStream.Current.Message,
            response.ResponseStream.Current.Timestamp));

        await Task.Delay(1000);
    }
}

References
https://www.youtube.com/watch?v=wY4nMSUF9e0&list=PLUOequmGnXxPOlhyA57ijmEyOeVmYQt32&index=4

Client Streaming with gRPC in C#

,proto

syntax = "proto3";

option csharp_namespace = "GrpcClientStreaming";

import "google/protobuf/timestamp.proto";

package greet;

service Greeter {
  rpc SayHello (stream HelloRequest) returns (HelloReply);
}

message HelloRequest {
  string name = 1;
  google.protobuf.Timestamp timestamp = 2;
}

message HelloReply {
  repeated string messages = 1;
}

Server

public override async Task<HelloReply> SayHello(IAsyncStreamReader<HelloRequest> requestStream,
    ServerCallContext context)
{
    HelloReply response = new HelloReply();

    while (await requestStream.MoveNext() && !context.CancellationToken.IsCancellationRequested)
    {
        response.Messages.Add(requestStream.Current.Timestamp.ToString());
        Console.WriteLine(requestStream.Current.Timestamp.ToString());
    }

    return response;
}

Client

static async Task Main(string[] args)
{
    using var channel = GrpcChannel.ForAddress("http://localhost:5000");
    var client = new Greeter.GreeterClient(channel);

    CancellationToken cancellationToken = new CancellationToken(false);
    var response = client.SayHello(new CallOptions(null, null, cancellationToken));

    for (int i = 0; i < 5; i++)
    {
        var timestamp = Timestamp.FromDateTime(DateTime.UtcNow);
        await response.RequestStream.WriteAsync(new HelloRequest
        {
            Name = "Mahmood",
            Timestamp = timestamp
        });

        Console.WriteLine(String.Format("Request : {0}", timestamp.ToString()));

        await Task.Delay(1000);
    }

    await response.RequestStream.CompleteAsync();

    Console.WriteLine("--------------------------------------");

    var result = response.ResponseAsync.Result;

    foreach (string message in result.Messages)
    {
        Console.WriteLine("Response : {0}", message);
    }
}

References
https://www.youtube.com/watch?v=DNxdvRQ4qRQ&list=PLUOequmGnXxPOlhyA57ijmEyOeVmYQt32&index=3

Server Streaming with gRPC in C#

.proto

syntax = "proto3";

import "google/protobuf/timestamp.proto";

option csharp_namespace = "GrpcServerStreaming";

package greet;

service Greeter {
  rpc SayHello (HelloRequest) returns (stream HelloReply);
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
  google.protobuf.Timestamp timestamp = 2;
}

Server

public override async Task SayHello(HelloRequest request, IServerStreamWriter<HelloReply> responseStream,
    ServerCallContext context)
{
    while (!context.CancellationToken.IsCancellationRequested)
    {
        await responseStream.WriteAsync(new HelloReply
        {
            Message = String.Format("Hello {0}", request.Name),
            Timestamp = Timestamp.FromDateTime(DateTime.UtcNow)
        });
        
        Console.WriteLine(Timestamp.FromDateTime(DateTime.UtcNow));
        await Task.Delay(1000);
    }
}

Client

static async Task Main(string[] args)
{
    // The port number(5001) must match the port of the gRPC server.
    using var channel = GrpcChannel.ForAddress("http://localhost:5000");
    var client = new Greeter.GreeterClient(channel);
    var response = client.SayHello(new HelloRequest {Name = "Mahmood"});
    
    while (await response.ResponseStream.MoveNext(new CancellationToken(false)))
    {
        Console.WriteLine(response.ResponseStream.Current.Message);
        Console.WriteLine(response.ResponseStream.Current.Timestamp);
        Console.WriteLine("----------------------------------------------------------------");
    }

    Console.ReadKey();
}

References
https://www.youtube.com/watch?v=F2T6xNRoa1E&list=PLUOequmGnXxPOlhyA57ijmEyOeVmYQt32&index=2

How to use Task.Run in C#

using System;
using System.Threading;
using System.Threading.Tasks;

public class Example
{
   public static void Main()
   {
      ShowThreadInfo("Application");

      var t = Task.Run(() => ShowThreadInfo("Task") );
      t.Wait();
   }

   static void ShowThreadInfo(String s)
   {
      Console.WriteLine("{0} Thread ID: {1}",
                        s, Thread.CurrentThread.ManagedThreadId);
   }
}
// The example displays the following output:
//       Application thread ID: 1
//       Task thread ID: 3
using System;
using System.Threading;
using System.Threading.Tasks;

public class Example
{
   public static void Main()
   {
      Console.WriteLine("Application thread ID: {0}",
                        Thread.CurrentThread.ManagedThreadId);
      var t = Task.Run(() => {  Console.WriteLine("Task thread ID: {0}",
                                   Thread.CurrentThread.ManagedThreadId);
                             } );
      t.Wait();
   }
}
// The example displays the following output:
//       Application thread ID: 1
//       Task thread ID: 3

References
https://docs.microsoft.com/en-us/dotnet/api/system.threading.tasks.task.run?view=net-5.0