Наблюдатели

Иногда простой шаблон сообщения или ответа недостаточно, и клиенту необходимо получать асинхронные уведомления. Например, пользователю может потребоваться уведомление, когда друг публикует новое мгновенное сообщение.

Клиентские наблюдатели — это механизм, разрешающий асинхронное уведомление клиентов. Интерфейсы наблюдателя должны наследоваться от IGrainObserver, и все методы должны возвращать либо void, , TaskTask<TResult>ValueTaskValueTask<TResult>либо . Мы не рекомендуем возвращать тип void , так как он может поощрять использование async void в реализации. Это опасный шаблон, так как это может привести к сбою приложения, если исключение выбрасывается из метода. Вместо этого для сценариев уведомления о лучших усилиях рассмотрите возможность применения OneWayAttribute метода интерфейса наблюдателя. Это приводит к тому, что получатель не отвечает на вызов метода, и метод возвращается немедленно в месте вызова, не ожидая ответа от наблюдателя. Зерно вызывает метод наблюдателя, вызывая его как любой метод интерфейса зерна. Среда выполнения Orleans гарантирует доставку запросов и ответов. Распространенный вариант использования для наблюдателей состоит в том, чтобы клиент получал уведомления при возникновении события в Orleans приложении. Такие уведомления должны предоставлять API для добавления или удаления наблюдателей. Кроме того, обычно удобно предоставлять метод, разрешающий отмену существующей подписки.

Вы можете использовать служебный класс, подобный ObserverManager<TObserver>, для упрощения разработки наблюдаемых типов зерна. В отличие от зерен, которые Orleans автоматически активируются по мере необходимости после сбоя, клиенты не являются отказоустойчивыми и могут никогда не восстановиться. По этой причине ObserverManager<T> программа удаляет подписки после заданной длительности. Активные клиенты должны повторно подписаться на таймер, чтобы сохранить свои подписки активными.

Чтобы подписаться на уведомление, клиент должен сначала создать локальный объект, реализующий интерфейс наблюдателя. Затем он вызывает метод на фабрике наблюдателей CreateObjectReference, чтобы превратить объект в ссылку на зерно. Затем эту ссылку можно передать методу подписки на оповещение.

Другие зерна также могут использовать эту модель для получения асинхронных уведомлений. Зерна могут реализовывать IGrainObserver интерфейсы. В отличие от случая подписки клиента, подписывающееся зерно просто реализует интерфейс наблюдателя и передает ссылку на себя (например, this.AsReference<IMyGrainObserverInterface>()). Нет необходимости в CreateObjectReference(), потому что зерна уже адресуемы.

Пример кода

Предположим, что у вас есть зерно, которое периодически отправляет сообщения клиентам. Для простоты сообщение в нашем примере представляет собой строку. Сначала определите интерфейс на клиенте, который получает сообщение.

Интерфейс выглядит следующим образом:

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

Единственное специальное требование заключается в том, что интерфейс должен наследоваться от IGrainObserver. Теперь любой клиент, желающий наблюдать за этими сообщениями, должен реализовать класс, реализующий IChat.

Самый простой случай выглядит примерно так:

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

На сервере вам нужно установить компонент, который отправляет эти сообщения чата клиентам. Механизм должен также предусматривать возможность подписки и отмены подписки клиентов на уведомления. Для подписок грейн может использовать экземпляр служебного класса ObserverManager<TObserver>.

Примечание.

ObserverManager<TObserver> является частью Orleans начиная с версии 7.0. Для более старых версий можно скопировать следующую реализацию .

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;
    }
}

Чтобы отправить сообщение клиентам, используйте Notify метод экземпляра ObserverManager<IChat> . Метод принимает Action<T> метод или лямбда-выражение (где T имеет тип IChat здесь). Вы можете вызвать любой метод в интерфейсе, чтобы отправить его клиентам. В нашем случае у нас есть только один метод, ReceiveMessageи наш код отправки на сервере выглядит следующим образом:

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

    return Task.CompletedTask;
}

Теперь наш сервер имеет метод отправки сообщений клиентам наблюдателей и два метода для подписки или отмены подписки. Клиент реализовал класс, способный наблюдать за сообщениями зерна. Последний шаг — создать ссылку наблюдателя на клиент с помощью ранее реализованного Chat класса и разрешить ему получать сообщения после подписки.

Код выглядит следующим образом:

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

Теперь, когда наш grain на сервере вызывает метод SendUpdateMessage, все клиенты-подписчики получают сообщение. В нашем клиентском коде экземпляр Chat в переменной c получает сообщение и выводит его в консоль.

Это важно

Объекты, переданные в CreateObjectReference, удерживаются с помощью WeakReference<T> и сборка мусора происходит, если не существует других ссылок.

Вы должны поддерживать ссылку для каждого наблюдателя, которого вам не нужно собирать.

Примечание.

Наблюдатели по своей природе ненадежны, так как хост клиента, на котором размещен наблюдатель, может выйти из строя, и наблюдатели, созданные после восстановления, имеют разные (случайно выбранные) удостоверения. ObserverManager<TObserver> использует периодическую повторную подписку наблюдателями, как упомянуто выше, для удаления неактивных наблюдателей.

Модель выполнения

Реализации IGrainObserver регистрируются с помощью вызова IGrainFactory.CreateObjectReference. Каждый вызов этого метода создает новую ссылку, указывающую на реализацию. Orleans выполняет запросы, отправленные каждому из этих ссылочных объектов последовательно до завершения. Наблюдатели не являются повторно-входящими; таким образом, Orleans не перемежает параллельные запросы к наблюдателю. Если несколько наблюдателей одновременно получают запросы, эти запросы могут выполняться параллельно. Такие атрибуты, как AlwaysInterleaveAttribute или ReentrantAttribute не влияют на выполнение методов наблюдателя; невозможно настроить модель выполнения.