Observadores

A veces, un patrón de mensaje o respuesta simple no es suficiente y el cliente debe recibir notificaciones asincrónicas. Por ejemplo, un usuario podría querer notificaciones cuando un amigo publica un nuevo mensaje instantáneo.

Los observadores de cliente son un mecanismo que permite la notificación asincrónica de los clientes. Las interfaces de observador deben heredar de IGrainObservery todos los métodos deben devolver void, Task, Task<TResult>, ValueTasko ValueTask<TResult>. No recomendamos un tipo de retorno de void porque podría fomentar el uso de async void en la implementación. Se trata de un patrón peligroso, ya que puede provocar fallos de la aplicación si se produce una excepción en el método. En su lugar, para escenarios de notificación de mejor esfuerzo, considere aplicar el OneWayAttribute al método de la interfaz del observador. Esto hace que el receptor no envíe una respuesta para la invocación del método y hace que el método vuelva inmediatamente en el sitio de llamada, sin esperar una respuesta del observador. Un grain llama a un método en un observador invocándolo como cualquier método de la interfaz de grain. El Orleans tiempo de ejecución garantiza la entrega de solicitudes y respuestas. Un caso de uso común para los observadores es inscribir a un cliente para recibir notificaciones cuando se produce un evento en la Orleans aplicación. Una unidad que publica dichas notificaciones debe proporcionar una API para agregar o quitar observadores. Además, suele ser conveniente exponer un método que permita la cancelación de una suscripción existente.

Puede usar una clase de utilidad como ObserverManager<TObserver> para simplificar el desarrollo de tipos de grano observados. A diferencia de los granos, que Orleans se reactivan automáticamente según sea necesario después del fallo, los clientes no son tolerantes a fallos: un cliente que falla podría no recuperarse nunca. Por este motivo, la ObserverManager<T> utilidad quita las suscripciones después de una duración configurada. Los clientes activos deben renovar su suscripción periódicamente para mantenerla activa.

Para suscribirse a una notificación, el cliente primero debe crear un objeto local que implemente la interfaz de observador. A continuación, llama a un método en el generador de observador, CreateObjectReference', para convertir el objeto en una referencia de grano. Luego, puede pasar esta referencia al método de suscripción en el grano de notificación.

Otros granos también pueden usar este modelo para recibir notificaciones asincrónicas. Los granos son capaces de implementar IGrainObserver interfaces. A diferencia del caso de la suscripción de cliente, el grano de suscripción simplemente implementa la interfaz de observador y pasa una referencia a sí misma (por ejemplo, this.AsReference<IMyGrainObserverInterface>()). No es necesario CreateObjectReference() porque los granulados ya son direccionables.

Ejemplo de código

Vamos a suponer que tiene una entidad que envía mensajes periódicamente a los clientes. Por motivos de simplicidad, el mensaje de nuestro ejemplo es una cadena. En primer lugar, defina la interfaz en el cliente que recibe el mensaje.

La interfaz tiene este aspecto:

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

El único requisito especial es que la interfaz debe heredar de IGrainObserver. Ahora, cualquier cliente que quiera observar estos mensajes debe implementar una clase que implemente IChat.

El caso más sencillo tiene un aspecto similar al siguiente:

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

En el servidor, a continuación, debería tener un grano que envíe estos mensajes de chat a los clientes. El sistema también debe proporcionar un mecanismo para que los clientes puedan suscribirse y cancelar su suscripción a las notificaciones. En el caso de las suscripciones, el grano puede usar una instancia de la clase de utilidad ObserverManager<TObserver>.

Nota:

ObserverManager<TObserver> forma parte de Orleans desde la versión 7.0. En el caso de las versiones anteriores, se puede copiar la siguiente implementación .

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 un mensaje a los clientes, use el Notify método de la ObserverManager<IChat> instancia. El método toma un Action<T> método o una expresión lambda (donde T es de tipo IChat aquí). Puede llamar a cualquier método de la interfaz para enviarlo a los clientes. En nuestro caso, solo tenemos un método, ReceiveMessagey nuestro código de envío en el servidor tiene este aspecto:

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

    return Task.CompletedTask;
}

Ahora, nuestro servidor tiene un método para enviar mensajes a los clientes de observador y dos métodos para suscribirse o anular la suscripción. El cliente ha implementado una clase que puede observar los mensajes de grano. El último paso consiste en crear una referencia de observador en el cliente mediante nuestra clase implementada Chat anteriormente y dejar que reciba mensajes después de suscribirse.

El código tiene este aspecto:

//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);

Ahora, cada vez que nuestra instancia en el servidor llama al método SendUpdateMessage, todos los clientes suscritos reciben el mensaje. En nuestro código de cliente, la Chat instancia de la variable c recibe el mensaje y la envía a la consola.

Importante

Los objetos pasados a CreateObjectReference se mantienen a través de WeakReference<T> y, por tanto, son recogidos por el recolector de basura si no existen otras referencias.

Debe mantener una referencia para cada observador que no quiera recopilar.

Nota:

Los observadores son intrínsecamente poco confiables porque un cliente que hospeda un observador podría producir un error y los observadores creados después de la recuperación tienen identidades diferentes (aleatorias). ObserverManager<TObserver> se basa en la suscripción periódica por parte de los observadores, como se ha descrito anteriormente, para que pueda quitar observadores inactivos.

Modelo de ejecución

Las implementaciones de IGrainObserver se registran mediante una llamada a IGrainFactory.CreateObjectReference. Cada llamada a ese método crea una nueva referencia que apunta a esa implementación. Orleans ejecuta las solicitudes enviadas a cada una de estas referencias una por una, hasta su finalización. Los observadores no son reentrantes; por lo tanto, Orleans no intercala las solicitudes simultáneas a un observador. Si varios observadores reciben solicitudes simultáneamente, esas solicitudes se pueden ejecutar en paralelo. Atributos como AlwaysInterleaveAttribute o ReentrantAttribute no afectan a la ejecución de métodos de observador; no se puede personalizar el modelo de ejecución.