Since lesson 4.4.5, the Kladde's pipeline board shows all deals of an office, column by column by stage. Since lesson 4.5.5, the sales commands run through the DealGrain. What is missing: when one colleague qualifies a deal, another colleague does not see it until he reloads the page. The board should update by itself.

For that, the system has to tell someone who is not asking right now. Orleans offers three ways: observers, streams and the broadcast channel. Add to them what the Kladde already has: the event store, from which the projector reads starting at its remembered position. This lesson puts the ways side by side and answers the real question: Which delivery is enough for what, and what may be lost along the way?

Grain observers

An observer is an object outside a grain that a grain can call as if it were a grain itself. A typical one is an object in the client, for example in a Blazor session, that receives messages from the cluster. Three building blocks are needed:

  1. An interface that inherits from IGrainObserver. Its methods return Task.
  2. An object that implements the interface, and an object reference to it, which IGrainFactory.CreateObjectReference creates. This reference can be passed to a grain like a grain reference.
  3. A grain that manages the references of its subscribers and calls them when needed. Orleans brings ObserverManager<T> for this.

ObserverManager has one important property: every subscription expires. A client can crash without unsubscribing, and a grain would otherwise send it messages forever. That is why the manager forgets every subscription that has not been renewed for longer than ExpirationDuration, and a client that wants to keep listening subscribes again regularly.

The following program shows the whole behavior in four steps. One feed grain per office manages the subscribers, and two “browser windows” listen. Create observer.cs:

C#
#:package Microsoft.Orleans.Server@10.4.2
#:property PublishAot=false
#:property ServerGarbageCollection=true

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Orleans.Concurrency;
using Orleans.Utilities;

var builder = Host.CreateApplicationBuilder(args);
builder.Logging.SetMinimumLevel(LogLevel.Warning);
builder.UseOrleans(silo => silo.UseLocalhostClustering());

using var host = builder.Build();
await host.StartAsync();

var grains = host.Services.GetRequiredService<IGrainFactory>();
var feed = grains.GetGrain<IPipelineFeedGrain>("office-north");
var deal = Guid.Parse("0199a1bd-2f0b-7dd3-b443-2fce002520f8");

var windowA = new BoardWindow("Window A");
var windowB = new BoardWindow("Window B");
var referenceA = grains.CreateObjectReference<IPipelineObserver>(windowA);
var referenceB = grains.CreateObjectReference<IPipelineObserver>(windowB);
await feed.SubscribeAsync(referenceA);
await feed.SubscribeAsync(referenceB);

await Step("1. Deal qualified", () => feed.ChangedAsync(deal, "Qualified"));

await feed.UnsubscribeAsync(referenceB);
grains.DeleteObjectReference<IPipelineObserver>(referenceB);
await Step("2. Window B closed, deal won", () => feed.ChangedAsync(deal, "Won"));

await feed.ForgetAsync();
await Step("3. Feed grain activated anew", () => feed.ChangedAsync(deal, "Lost"));

await feed.SubscribeAsync(referenceA);
await Step("4. Window A renewed its subscription", () => feed.ChangedAsync(deal, "Won"));

grains.DeleteObjectReference<IPipelineObserver>(referenceA);
await host.StopAsync();

static async Task Step(string text, Func<Task> change)
{
    Console.WriteLine(text);
    await change();
    await Task.Delay(TimeSpan.FromMilliseconds(300)); // one-way messages arrive a moment later
}

public interface IPipelineObserver : IGrainObserver
{
    [OneWay]
    Task ChangedAsync(Guid deal, string stage);
}

public interface IPipelineFeedGrain : IGrainWithStringKey
{
    Task SubscribeAsync(IPipelineObserver observer);

    Task UnsubscribeAsync(IPipelineObserver observer);

    [OneWay]
    Task ChangedAsync(Guid deal, string stage);

    Task ForgetAsync();
}

public sealed class PipelineFeedGrain(ILogger<PipelineFeedGrain> logger) : Grain, IPipelineFeedGrain
{
    private readonly ObserverManager<IPipelineObserver> _observers = new(TimeSpan.FromMinutes(2), logger);

    public Task SubscribeAsync(IPipelineObserver observer)
    {
        _observers.Subscribe(observer, observer);
        return Task.CompletedTask;
    }

    public Task UnsubscribeAsync(IPipelineObserver observer)
    {
        _observers.Unsubscribe(observer);
        return Task.CompletedTask;
    }

    public Task ChangedAsync(Guid deal, string stage)
    {
        _observers.Notify(observer => observer.ChangedAsync(deal, stage));
        return Task.CompletedTask;
    }

    public Task ForgetAsync()
    {
        this.DeactivateOnIdle();
        return Task.CompletedTask;
    }
}

public sealed class BoardWindow(string name) : IPipelineObserver
{
    public Task ChangedAsync(Guid deal, string stage)
    {
        Console.WriteLine($"   {name}: Deal {deal.ToString()[..8]} is now {stage}");
        return Task.CompletedTask;
    }
}
Bash
dotnet run observer.cs

1. Deal qualified
   Window B: Deal 0199a1bd is now Qualified
   Window A: Deal 0199a1bd is now Qualified
2. Window B closed, deal won
   Window A: Deal 0199a1bd is now Won
3. Feed grain activated anew
4. Window A renewed its subscription
   Window A: Deal 0199a1bd is now Won

The order of the two windows in step 1 may be different on your machine; for two different receivers there is no order. The four steps show everything you need to know about observers:

  1. One call reaches all subscribers. Notify calls the method for each of them. ObserverManager lives in Orleans.Utilities.
  2. Unsubscribing and releasing belong together. UnsubscribeAsync takes the window out of the grain's list. DeleteObjectReference releases the reference in the client. Without the second, the registration stays behind in the client, one for every closed window, until memory is full.
  3. Subscriptions are not durable. ObserverManager keeps its list in the memory of the activation. If the feed grain is deactivated or its silo fails, the list is empty, and the message in step 3 reaches nobody. No error, no notice.
  4. Renewing heals. Because every subscriber subscribes again regularly, it is back after one beat at the latest. The same renewal that expiry demands also repairs the loss.

Three more details:

  • [OneWay] from Orleans.Concurrency means: the caller does not wait for a reply, the call returns immediately, and the caller never learns about errors at the receiver. For a notification “something has changed” that is right. The price: a subscriber whose process is no longer running stays in the list until its subscription expires, and for as long as that lasts Orleans writes an error to the log for every message to it. Without [OneWay] and with Notify(Func<T, Task>), which waits, the manager recognizes a dead subscriber by the error and removes it immediately; but then the feed grain waits for the slowest receiver. For the Kladde, [OneWay] with a short expiry is the better choice.
  • The client holds on to the object. CreateObjectReference remembers the object only weakly (WeakReference). If nobody else holds the BoardWindow, garbage collection removes it, and messages go nowhere.
  • Observers handle one call after another, like grains. Two windows do get their messages at the same time, though.