Observadores

Há situações em que um padrão de mensagem/resposta simples não é suficiente e o cliente precisa receber notificações assíncronas. Por exemplo, talvez seja conveniente notificar um usuário quando uma nova mensagem instantânea for publicada por um amigo.

Observadores de cliente são um mecanismo que permite notificar clientes de forma assíncrona. As interfaces do observador devem herdar de IGrainObserver e todos os métodos devem retornar void, Task, Task<TResult>, ValueTask ou ValueTask<TResult>. Não é recomendado um tipo de retorno void, pois ele pode incentivar o uso de async void na implementação, que é um padrão perigoso que pode resultar em falhas do aplicativo em caso de uma exceção do método. Em vez disso, para cenários de notificação de melhor esforço, considere aplicar OneWayAttribute ao método de interface do observador. Isso fará com que o receptor não envie uma resposta para a invocação do método e fará com que o método retorne imediatamente ao local da chamada, sem esperar uma resposta do observador. Uma granularidade chama um método em um observador, invocando-o como qualquer método de interface de granularidade. O runtime do Orleans garantirá a entrega de solicitações e respostas. Um caso de uso comum para observadores é inscrever um cliente para receber notificações quando um evento ocorrer no aplicativo Orleans. Um grain que publica essas notificações deve fornecer uma API para adicionar ou remover observadores. Além disso, geralmente é conveniente expor um método que permita que uma assinatura existente seja cancelada.

Os desenvolvedores de grain podem usar uma classe de utilitário como ObserverManager<TObserver> para simplificar o desenvolvimento de tipos de grain observados. Ao contrário dos grains, que são automaticamente reativados conforme necessário após a falha, os clientes não são tolerantes a falhas: um cliente que falha pode nunca se recuperar. Por esse motivo, o utilitário ObserverManager<T> remove assinaturas após uma duração configurada. Os clientes que estão ativos devem assinar novamente um temporizador para manter a assinatura ativa.

Para assinar uma notificação, o cliente deve primeiro criar um objeto local que implemente a interface do observador. Em seguida, ele chama um método na fábrica de observadores, CreateObjectReference, para transformar o objeto em uma referência de grain, que poderá, então, ser passado para o método de assinatura no grain da notificação.

Esse modelo também pode ser usado por outros grains para receber notificações assíncronas. Os grains também podem implementar interfaces IGrainObserver. Ao contrário do caso da assinatura do cliente, o grain de assinatura simplesmente implementa a interface do observador e passa uma referência para si mesmo (por exemplo, this.AsReference<IMyGrainObserverInterface>()). Não há necessidade de CreateObjectReference(), pois os grains já são endereçáveis.

Exemplo de código

Vamos supor que temos um grain que envia periodicamente mensagens aos clientes. Para simplificar, a mensagem do nosso exemplo será uma cadeia de caracteres. Primeiramente, definimos a interface no cliente que receberá a mensagem.

A interface terá esta aparência:

public interface IChat : IGrainObserver
{
    Task ReceiveMessage(string message);
}

A única coisa especial é que a interface deve herdar de IGrainObserver. Agora, qualquer cliente que queira observar essas mensagens deve implementar uma classe que implemente IChat.

O caso mais simples seria algo semelhante a:

public class Chat : IChat
{
    public Task ReceiveMessage(string message)
    {
        Console.WriteLine(message);
        return Task.CompletedTask;
    }
}

No servidor, devemos ter um Grain que envie essas mensagens de chat aos clientes. O Grain também deve ter um mecanismo para que os clientes assinem e cancelem a assinatura de notificações. Para assinaturas, a granularidade pode usar uma instância da classe utilitária ObserverManager<TObserver>.

Observação

O ObserverManager<TObserver> faz parte do Orleans desde a versão 7.0. Para versões mais antigas, a implementação a seguir pode ser copiada.

class HelloGrain : Grain, IHello
{
    private readonly ObserverManager<IChat> _subsManager;

    public HelloGrain(ILogger<HelloGrain> logger)
    {
        _subsManager =
            new ObserverManager<IChat>(
                TimeSpan.FromMinutes(5), logger);
    }

    // Clients call this to subscribe.
    public Task Subscribe(IChat observer)
    {
        _subsManager.Subscribe(observer, observer);

        return Task.CompletedTask;
    }

    //Clients use this to unsubscribe and no longer receive messages.
    public Task UnSubscribe(IChat observer)
    {
        _subsManager.Unsubscribe(observer);

        return Task.CompletedTask;
    }
}

Para enviar uma mensagem aos clientes, o método Notify da instância ObserverManager<IChat> pode ser usado. O método usa um método Action<T> ou expressão lambda (e que T é do tipo IChat aqui). Você pode chamar qualquer método na interface para enviá-lo aos clientes. No nosso caso, só temos um método, ReceiveMessage, e o nosso código de envio no servidor seria semelhante a este:

public Task SendUpdateMessage(string message)
{
    _subsManager.Notify(s => s.ReceiveMessage(message));

    return Task.CompletedTask;
}

Agora, nosso servidor tem um método para enviar mensagens para clientes observadores, dois métodos para assinatura/cancelamento de assinatura e o cliente implementou uma classe capaz de observar as mensagens de grain. A última etapa é criar uma referência de observador no cliente usando nossa classe Chat implementada anteriormente e deixá-lo receber as mensagens depois de assinar.

O código seria assim:

//First create the grain reference
var friend = _grainFactory.GetGrain<IHello>(0);
Chat c = new Chat();

//Create a reference for chat, usable for subscribing to the observable grain.
var obj = _grainFactory.CreateObjectReference<IChat>(c);

//Subscribe the instance to receive messages.
await friend.Subscribe(obj);

Agora, sempre que nosso grain no servidor chamar o método SendUpdateMessage, todos os clientes inscritos receberão a mensagem. Em nosso código cliente, a instância Chat na variável c receberá a mensagem e a gerará para o console.

Importante

Os objetos passados para CreateObjectReference são mantidos por meio de um WeakReference<T> e, portanto, serão coletados lixos se não existirem outras referências.

Os usuários devem manter uma referência de cada observador que não queiram coletar.

Observação

Observadores não são inerentemente confiáveis, pois um cliente que hospeda um observador pode falhar, e os observadores criados após a recuperação têm identidades diferentes (aleatórias). O ObserverManager<TObserver> depende de novas assinaturas periódicas pelos observadores, conforme discutido acima, para que observadores inativos possam ser removidos.

Modelo de execução

As implementações de IGrainObserver são registradas por meio de uma chamada para IGrainFactory.CreateObjectReference e cada chamada para esse método cria uma nova referência que aponta para essa implementação. O Orleans Orleans executará solicitações enviadas a cada uma dessas referências, uma por uma, até a conclusão. Observadores não são reentrantes e, portanto, solicitações simultâneas para um observador não serão intercaladas pelo Orleans. Se houver vários observadores recebendo solicitações simultaneamente, essas solicitações poderão ser executadas em paralelo. A execução de métodos do observador não é afetada por atributos como AlwaysInterleaveAttribute ou ReentrantAttribute: o modelo de execução não pode ser personalizado por um desenvolvedor.