Observadores

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

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

Os desenvolvedores de grãos podem usar uma classe de utilidade, como ObserverManager<TObserver> para simplificar o desenvolvimento de tipos de grãos observados. Ao contrário dos grãos, que são reativados automaticamente 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 ObserverManager<T> utilitário remove assinaturas após uma duração configurada. Os clientes que estão ativos devem se inscrever novamente em um temporizador para manter sua assinatura ativa.

Para assinar uma notificação, o cliente deve primeiro criar um objeto local que implemente a interface do observador. Em seguida, CreateObjectReferencechama um método na fábrica do observador, ', para transformar o objeto em uma referência de grão, que pode então ser passada para o método de assinatura no grão notificador.

Este modelo também pode ser usado por outros grãos para receber notificações assíncronas. Grãos também podem implementar IGrainObserver interfaces. Ao contrário do caso da assinatura do cliente, o grão 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 porque CreateObjectReference() os grãos já são endereçáveis.

Exemplo de código

Vamos supor que temos um grão que periodicamente envia mensagens aos clientes. Para simplificar, a mensagem no nosso exemplo será uma cadeia de caracteres. Primeiro, 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 do IGrainObserver. Agora, qualquer cliente que queira observar essas mensagens deve implementar uma classe que implemente IChato .

O caso mais simples seria algo assim:

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

No servidor, devemos ter em seguida um grão que envia essas mensagens de chat para os clientes. O Grão também deve ter um mecanismo para que os clientes se inscrevam e cancelem a assinatura para notificações. Para assinaturas, o grão pode usar uma instância da classe ObserverManager<TObserver>de utilitário.

Nota

ObserverManager<TObserver> faz parte da Orleans versão 7.0. Para versões mais antigas, a seguinte implementação 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 Notify método da ObserverManager<IChat> instância pode ser usado. O método usa um Action<T> método ou expressão lambda (onde T é do tipo IChat aqui). Você pode chamar qualquer método na interface para enviá-lo aos clientes. No nosso caso, temos apenas um método, ReceiveMessagee o nosso código de envio no servidor ficaria assim:

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 assinar/cancelar a assinatura, e o cliente implementou uma classe capaz de observar as mensagens de grão. O último passo é criar uma referência de observador no cliente usando nossa classe implementada Chat anteriormente e deixá-lo receber as mensagens depois de assiná-lo.

O código ficaria 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 grão no servidor chama o SendUpdateMessage método, todos os clientes inscritos receberão a mensagem. Em nosso código de cliente, a Chat instância em variável c receberá a mensagem e a enviará para o console.

Importante

Os objetos passados para CreateObjectReference são mantidos através de um WeakReference<T> e, portanto, serão recolhidos de lixo se não existirem outras referências.

Os utilizadores devem manter uma referência para cada observador que não desejem que seja recolhida.

Nota

Os observadores são inerentemente não confiáveis, uma vez que um cliente que hospeda um observador pode falhar e os observadores criados após a recuperação têm identidades diferentes (aleatórias). ObserverManager<TObserver> depende da reinscrição periódica dos observadores, como discutido acima, para que os observadores inativos possam ser removidos.

Modelo de execução

As implementações de são registradas por meio de IGrainObserver uma chamada para IGrainFactory.CreateObjectReference e cada chamada para esse método cria uma nova referência que aponta para essa implementação. Orleans executará os pedidos enviados a cada uma destas referências, um a um, até à sua conclusão. Os observadores não são reentrantes e, por conseguinte, os pedidos simultâneos a um observador não serão intercalados pela Orleans. Se houver vários observadores que estão recebendo solicitações simultaneamente, essas solicitações podem ser executadas em paralelo. A execução de métodos de observação não é afetada por atributos como AlwaysInterleaveAttribute ou ReentrantAttribute: o modelo de execução não pode ser personalizado por um desenvolvedor.