CoreLib EventDriven Design Reference

Nota importante: o presente assembly encontra-se descontinuado. Novas implementações event-driven devem utilizar os assemblies Primavera.Hydrogen.EventBus.

Keyworks: EventDriven Event Bus Broken Publisher Subscriber Observer Topic

Introdução

A existência de referências físicas entre componentes de uma aplicação contribui muito para a sua complexidade e para os custos de manutenção e evolução desses componentes. Historicamente esse tem sido um dos maiores problemas das aplicações PRIMAVERA e também do Business Suite e da Plataforma Elevation.

As arquiteturas orientadas a eventos apresentam uma abordagem para este problema, entre outros, através da remoção dessas referências entre componentes, substituindo-as por dependências “loosely-coupled” implementadas através de eventos que os componentes “partilham” para transferir informação entre si.

O padrão Publisher/Subscriber é, digamos assim, uma versão simplificada de uma arquitetura orientada a eventos (EDA) em que um componente publica eventos e um ou mais outros componentes os observam e decidem se devem ou não atuar sobre esses eventos.

Este documento descreve a arquitetura do componente EventBus (em Primavera.CoreLib.EventDriven).

Este componente implementa um “event broker” que permite a realização do padrão publisher/subscriber entre quaisquer dois componentes da Plataforma Elevation.

Conceitos

EDA (Event-driven Architecture)

EDA é um padrão de arquitetura que visa a promoção da produção, deteção, consumo e reação a eventos. Na base do seu desenho está a separação física dos componentes que partilham os eventos.

Evento

Um evento é uma alteração significativa do estado de um componente. Os componentes participantes numa EDA produzem e consomem notificações sobre esses eventos (notificação, mensagem ou tópico). É comum, no entanto, referenciar tanto o evento como a sua notificação pelo mesmo nome: evento.

Publisher/subscriber

Este padrão permite que um componente que emite uma mensagem (o publisher) não tenha que estar programado para a enviar diretamente para cada recetor específico (os subscribers). Antes o publisher caracteriza a mensagem através de uma classe e publica-a (de alguma forma) sem conhecimento dos eventuais subscribers. Os subscriber são responsáveis por especificar de alguma forma as mensagens em que estão interessados e só receberão essas. Este padrão pode ser implementado como uma forma de EDA em que as mensagens são as notificações dos eventos.

Publisher

É o componente que publica mensagens numa arquitetura publisher/subscriber.

Subscriber

É o componente que receber mensagem numa arquitetura publisher/subscriber.

Observer

O padrão observer é semelhante ao publisher/subscriber e, por isso, é comum que os subscribers sejam chamados de observers.

Event Broker

Implementações mais evoluídas do padrão publisher/subscriber utilizam um componente que atua como intermediário entre o publisher e os subscribers e que permite a realização de operações mais avançadas: como a filtragem dos eventos ou a sua priorização. Este nível adicional de separação entre os componentes “em comunicação” contribui para a sua maior independência e facilita a implementação de cenários mais avançados como, por exemplo, remoting, configuração declarativa, etc.

Event Bus

Este termo é sinónimo de Event Broker mas remete para o padrão service bus, em que os componentes da arquitetura publisher/subscriber são serviços em máquinas distintas, eventualmente, em qualquer parte do globo.

Event Topic

O tópico descreve de alguma forma o evento ocorrido e atua como uma notificação da sua ocorrência produzida pelo componente onde o evento ocorreu e consumida pelo(s) componente(s) interessado(s) nesse evento.

Pressupostos

  • Pretende-se estabelecer uma arquitetura EDA através do padrão publisher/subscriber implementado com um event bus.
  • Inicialmente esse componente será utilizado para partilhar eventos in-process, ou seja, entre componentes da mesma aplicação dentro do mesmo processo.
  • No entanto, a evolução da Plataforma Elevation vai no sentido de introduzir uma queue de pedidos processados offline e de um sistema de notificações no momento da sua conclusão (para processos long-running como, por exemplo, o processamento de salários). A arquitetura usada no Event Bus tem exatamente o objetivo de suportar esse cenário através da partilha de eventos out of process (mesmo em máquinas distintas).
  • A API do serviço deve ser o mais simples possível para facilitar a sua utilização e deve ser totalmente configurável, tanto em design-time como em runtime.
  • Os publishers não devem ter conhecimento nenhum sobre os subcribers.
  • Os subscribers apenas devem conhecer os event topics em que estiverem interessados.

Desenho

Os componentes do Event Bus da Plataforma Elevation são disponibilizados pelas seguintes assemblies:

  • Primavera.CoreLib: inclui o Event Bus assim como os componentes da sua implementação in-process.
  • Primavera.CoreLib.EventDriven.Azure: inclui os componentes da implementação do event bus que utiliza serviços do Microsoft Azure.

Todos os componentes do Event Bus estão definidos sob o namespace Primavera.Core.EventDriven.

A API principal do serviço Event Bus é descrita pela interface Primavera.Core.EventDriven.IEventBus. Os métodos disponíveis permitem:

  • Configurar e inicializar o serviço.
  • Publicar eventos.
  • Subscrever eventos.

Os membros principais são os seguintes:

  • Id (get) - disponibiliza o identificador único de uma instância do event bus.
  • Configuration (get) – permite inicializar a configuração do bus e/ou alterá-la em runtime.
  • Start – permite, opcionalmente, iniciar o serviço antes de realizar qualquer outra operação.
  • Stop – permite, opcionalmente, parar o serviço.
  • Reset – permite, opcionalmente, reinicializar o serviço.
  • Publish – permite publicar eventos para o bus (de forma síncrona).
  • PublishAsync – permite publicar eventos para o bus (de forma assíncrona).
  • Subscribe – permite adicionar subscrições de eventos (de forma síncrona).
  • SubscribeAsync – permite adicionar subscrições de eventos.
  • Unsubscribe – permite remover subscrições de eventos (de forma síncrona).
  • UnsubscribeAsync – permite remover subscrições de eventos (de forma assíncrona).

O publisher usará os métodos Publish e/ou PublishAsync e não os métodos Subscribe, SubscribeAsync, Unsubscribe ou UnsubscribeAsync.

O subscriber usará os métodos Subscribe, SubscribeAsync, Unsubscribe e/ou UnsubscribeAsync e não os métodos Publish ou PublishAsync.

O bus disponibiliza ainda os seus próprios eventos que permitem observar quando determinadas operações acontecem:

  • Subscribing
  • Subscribed
  • Unsubscribing
  • Unsubscribed
  • Publishing
  • Published

A implementação concreta do event bus é realizada pela classe Primavera.Core.EventDriven.EventBus.

Visão Geral

CoreLib EventDriven

Notem-se os seguintes aspetos fundamentais da arquitetura do event bus:

  • O publisher desconhece em absoluto a existência de subscribers e não tem qualquer dependência para eles.
  • O subscriber apenas conhece o publisher na medida em que sabe que existe um componente que publica um determinado tópico (definido por uma classe). O subscriber deve ser capaz de criar instâncias dessa classe, logo, daí resulta que o tópico não pode ser alterado arbitrariamente pelo publisher porque podem existir subscribers dependentes da versão anterior do tópico.
  • O event bus é o único componente que conhece, digamos assim, tanto o publisher como o subscriber. No entanto, a arquitetura garante ainda uma grande independência do bus em relação a esses componentes porque:
    • Não são guardadas quaisquer referências para os publishers. Depois de publicado um evento, o bus deixa de depender do que quer que seja do publisher. Por essa razão é que a arquitetura não tipifica os publishers (seja por um interface ou uma classe base).
    • Os subscribers são tipificados pelo interface Primavera.Core.EventDriven.IEventSubscriber<TEventTopic> o que permite guardar as suas subscrições sem depender da classe concreta que implementa cada subscriber.
    • Os tópicos são reconhecidos pelo interface Primavera.Core.EventDriven.IEventTopic e, por isso, são transferidos pelo bus sem dependerem da classe concreta de cada um.

Visão Interna

Internamente o event bus é um pouco mais complexo. As suas funcionalidades principais são disponibilizadas por quatro componentes.

CoreLib EventDriven

A persistência das subscrições (Primavera.Core.EventDriven.IEventBusSubscriptionStore) define o mecanismo usado para armazenar as subscrições.

Estão disponíveis as seguintes implementações:

  • Primavera.Core.EventDriven.InProcess.InProcessEventBusSubscriptionStore – armazena as subscrições em memória.
  • Primavera.Core.EventDriven.Azure.AzureTopicsEventBusSubscriptionStore - utiliza o Microsoft Azure Service Bus para armazenar as subscrições.

O transporte (Primavera.Core.EventDriven.IEventBusTransport), como o nome indica, é responsável pelo envio do tópico, após publicação, para cada um dos subscritores.

Estão disponíveis as seguintes implementações:

  • Primavera.Core.EventDriven.InProcess.InProcessEventBusTransport – os eventos são enviados diretamente para a instância de cada subscriber .
  • Primavera.Core.EventDriven.Azure.AzureTopicsEventBusTransport - utiliza os tópicos do Microsoft Azure Service Bus como transporte.

O filtro (Primavera.Core.EventDriven.IEventBusFilter) é o componente responsável por filtrar os eventos enviados para cada subscriber.

Estão disponíveis as seguintes implementações:

  • Primavera.Core.EventDriven.Internal.NoOpEventBusFilter - implementação que nunca filtra os eventos, delegando a filtragem dos tópicos ao mecanismo de transporte (por exemplo: o transporte que utiliza o Azure Service Bus já disponibiliza esse comportamento).
  • Primavera.Core.EventDriven.InProcess.InProcessEventBusFilter - os tópicos são filtrados em memória.

A configuração (Primavera.Core.EventDriven.EventBusConfiguration) permite adaptar o comportamento do bus sob vários aspetos, entre os quais definir o transporte, o suporte de persistência das subscrições e o filtro usados pela instância do bus.

A arquitetura define ainda um conjunto de interfaces que tipificam os elementos principais da mesma:

  • IEventTopic – identifica uma classe como sendo um tópico. O publisher só pode publicar eventos representados por classes que implementem esse interface e o subscriber só pode subscrever topicos que implementem esse interface.
  • IEventSubscriber<TEventTopic> where TEventTopic : IEventTopic – identifica um subscriber e ainda permite ao subscriber indicar qual é o tópico que subscreve. Um subscriber poder subscrever vários eventos mas para isso deve implementar o interface para cada um dos eventos (IEventSubscriber, IEventSubscriber, etc.).
  • IEventFilter<TEventTopic> where TEventTopic : IEventTopic – define uma classe como sendo um filtro aplicável ao respetivo evento. É esta classe que um subscriber deve fornecer no momento da subscrição para filtrar os eventos do tipo subscrito.

A mecânica detalhada de cada um destes componentes e o papel de cada um dos interfaces é descrito nos capítulos seguintes.

Inicialização

A inicialização do event bus é automática na primeira invocação do componente. Essa inicialização pode ser despoletada das seguintes formas:

  • EventBus.Default()
  • EventBus.Instance(string name)
  • EventBus.Create(EventBusConfiguration configuration)

A primeira utilizará a configuração por defeito (os valores por defeito da configuração e a configuração realizada através do ficheiro de configuração da aplicação). A instância terá um nome por defeito.

A segunda utilizará a configuração por defeito também mas criará uma instância diferente do bus para cada nome especificado.

A terceira utilizará explicitamente a configuração fornecida como parâmetro.

Regra geral deverá utilizar-se a primeira forma.

Instâncias do Event Bus

As instâncias do event bus são estáticas por configuração, o que significa que, para cada processo, existirá uma única instância do bus por cada configuração distinta.

A forma como o event bus é invocado define a instância que será utilizada.

(1) Neste caso ambas as invocações resultam na utilização da mesma instância:

EventBus.Default().Subscribe<Event1, Subscriber1>();
EventBus.Default().Subscribe<Event2, Subscriber2>();

(2) Neste caso serão utilizadas instâncias diferentes:

EventBus.Instance("B1").Subscribe<Event1, Subscriber1>();
EventBus.Instance("B2").Subscribe<Event2, Subscriber2>();

(3) Neste caso também será utilizada a mesma instância uma vez que a configuração é igual (apesar de criada de uma forma distinta):

EventBus.Default().Subscribe<Event1, Subscriber1>();
EventBusConfiguration config = EventBusConfiguration.Default();
EventBus.Create(config).Subscribe<Event2, Subscriber2>();

(4) Também neste caso a instância será a mesma, porque a configuração inicial é a mesma apesar de ser depois modificada em runtime (afetando a mesma instância):

EventBus.Default().Subscribe<Event1, Subscriber1>();
EventBus.Configuration.WithSetting(EventBusSettingsBuilder.TraceData, true);
EventBus.Default().Subscribe<Event2, Subscriber2>();

(5) Neste caso são criadas duas estâncias distintas porque as duas configurações são diferentes:

EventBusConfiguration config1 = EventBusConfiguration.Default();
IEventBus bus1 = EventBus.Create(config1);
BusConfiguration config2 = EventBusConfiguration.Default().WithSetting(EventBusSettingsBuilder.TraceData, true);
IEventBus bus2 = EventBus.Create(config2);

Start()

A invocação explicita do método Start() não é sempre necessária porque todos os outros métodos da API a fazem implicitamente.

Este código:

IEventBus bus = EventBus.Default();
bus.PublishAsync<Event1>();

Tem exatamente o mesmo efeito que:

IEventBus bus = EventBus.Default();
bus.Start();
bus.PublishAsync<Event1>();

No entanto, o método pode ser invocado quando se pretende que o bus esteja pronto a receber subscrições sem realizar nenhuma outra operação:

EventBus.Default().Start();
// Wait for subscriptions…

Configuração

O event bus é altamente configurável e permite a realização da configuração de várias formas:

  • Em design time, usando a secção de configuração EventBusConfigurationSection.
  • Em runtime, criando uma instância da classe EventBusConfiguration e passando-a no método EventBus.Create().

O fluxo de inicialização do bus tratará de carregar a configuração correta da seguinte forma:

  • Se existir a secção de configuração no ficheiro de configuração da aplicação, serão usados os settings aí definidos. Os que não estiverem definidos assumirão os valores por defeito.
  • Caso contrário, será usada a configuração por defeito (com os valores por defeito para todos os settings).
  • As configurações que sejam fornecidas na instância EventBusConfiguration (no método Create) sobrepõem-se sempre às anteriores.

Em qualquer dos casos a configuração do bus pode depois ser alterada através da propriedade Configuration da API.

Settings Disponíveis

As configurações que podem ser alteradas estão disponíveis na classe EventBusConfiguration. São as seguintes:

  • Name: define o nome da instância do bus.
  • SubscriptionStoreType: define o tipo que implementa o mecanismo de persistência das subscrições. Valor por defeito: Primavera.Core.EventDriven.InProcess.InProcessEventBusSubscriptionStore.
  • TransportType: define o tipo que implementa o mecanismo de transporte de eventos. Valor por defeito: Primavera.Core.EventDriven.InProcess.InProcessEventBusTransport.
  • FilterType: define o tipo de implementa o mecanismo de filtragem de eventos. Valor por defeito: Primavera.Core.EventDriven.InProcess.InProcessEventBusFilter.
  • Settings: Conjunto de outros settings.

As configurações que podem ser definidas através dos Settings dependem dos tipos usados nos componentes do bus.

Estão disponíveis duas classes que definem settings diferentes:

  • Primavera.Core.EventDriven.EventBusSettingsBuilder: define settings gerais.
    • TraceData: indica se os dados dos tópicos devem ser registrados no trace do event bus. O valor por defeito é falso.
  • Primavera.Core.EventDriven.Azure.AzureTopicsEventBusSettingsBuilder: define settings associados à utilização dos Azure Service Bus Topics.
    • ConnectionString: define a connection string para ligação ao Azure Service Bus.
    • OperationTimeout: define o tempo máximo de execução de qualquer operação no service bus. O valor por defeito é 1 minuto.
    • RetryPolicy: define a política que deve ser seguida para ultrapassar erros transientes ao aceder à API do service bus. O valor por defeito é AzureServiceBusRetryPolicy.Exponential.
    • RetryPolicyMaximumAttempts: define o número máximo de vezes que a política de retry deve ser aplicada. O valor por defeito é 3.
    • RetryPolicyMinBackoffTime: define o tempo mínimo entre tentativas de retry. O valor por defeito é 2 segundos.
    • RetryPolicyMaxBackoffTime: define o tempo máximo entre tentativas de retry. O valor por defeito é 6 segundos.
    • MessageAutoRenewTimeout: define a duração máxima do lock de mensagens. O valor por defeito é 1 minuto.
    • MessageMaxConcurrentCalls: define o número máximo de chamadas concorrentes ao callback de mensagem. O valor por defeito é 1.
    • DefaultMessageTimeToLive: define o tempo máximo de vida de uma mensagem no service bus. O valor por defeito é igual ao definido no Azure Service Bus.
    • SubscriptionMaxDeliveryCount: define o número máximo de entregas para uma subscrição. O valor por defeito é 10.

Ficheiro de Configuração

A configuração do event bus pode ser declarada no ficheiro de configuração através da secção EventBusConfigurationSection (que é uma secção overridable ) como no exemplo seguinte:

<?xml version="1.0" encoding="utf-8"?>
<configuration>
  <configSections>
    <section name="eventBusConfiguration" type="Primavera.Core.EventDriven.EventBusConfigurationSection, Primavera.CoreLib.EventDriven, Version=4.0.0.0, Culture=neutral, PublicKeyToken=33086db60a481256"/>
  </configSections>
  <eventBusConfiguration>
    <elements>
      <add name="AZSB"
           subscriptionStoreType="Primavera.Core.EventDriven.Azure.AzureTopicsEventBusSubscriptionStore, Primavera.CoreLib.EventDriven.Azure, Version=4.0.0.0, Culture=neutral, PublicKeyToken=33086db60a481256"
           transportType="Primavera.Core.EventDriven.Azure.AzureTopicsEventBusTransport, Primavera.CoreLib.EventDriven.Azure, Version=4.0.0.0, Culture=neutral, PublicKeyToken=33086db60a481256"
           filterType="Primavera.Core.EventDriven.Internal.NoOpEventBusFilter, Primavera.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=33086db60a481256">
           <settings>
             <add key="AzureServiceBus.ConnectionString"
                  value="Endpoint=sb://intdev.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=Mg8rUs/78Cv95B6qAU6Ukv6F+FNDThSNdIHJ7PWr/Tc="/>
           </settings>
      </add>
    </elements>
  </eventBusConfiguration>
</configuration>

Note-se que, como já foi referido, nenhum dos settings é obrigatório dado que assumirá o seu valor por defeito caso não seja definido.

Configuração em Runtime

Em runtime, a configuração é definida pela classe EventBusConfiguration que disponibiliza uma API fluente para definir o valor dos vários settings.

Exemplo:

EventBusConfiguration configuration = EventBusConfiguration
    .Default("name")
    .WithSubscriptionStore<Primavera.Core.EventDriven.Azure.AzureTopicsEventBusSubscriptionStore>()
    .WithTransport<Primavera.Core.EventDriven.Azure.AzureTopicsEventBusTransport>()
    .WithFilter<Primavera.Core.EventDriven.Internal.NoOpEventBusFilter>()
    .WithSetting(AzureTopicsEventBusSettingsBuilder.ConnectionString, connectionString);
IEventBus bus = EventBus.Create(configuration);
bus.Start();

Depois de inicializado o bus, a configuração pode ser alterada usando a mesma API. Como no exemplo seguinte:

IEventBus bus = EventBus.Create(configuration);
// ...
bus
    .Configuration
    .WithSubscriptionStore<Primavera.Core.EventDriven.Azure.AzureTopicsEventBusSubscriptionStore>()
    .WithTransport<Primavera.Core.EventDriven.Azure.AzureTopicsEventBusTransport>()
    .WithFilter<Primavera.Core.EventDriven.Internal.NoOpEventBusFilter>()
    .WithSetting(AzureTopicsEventBusSettingsBuilder.ConnectionString, connectionString);

Lifetime

Como foi referido anteriormente cada instância do bus “arranca” automaticamente caso seja chamado o método Start explicitamente ou, implicitamente, se for chamado qualquer outro método da API (Publish, Subscribe, etc.).

Da mesma forma que não é necessário iniciar o bus também não é obrigatório pará-lo explicitamente porque isso é realizado automaticamente quando a instância do bus é libertada pelo runtime.

No entanto, esse processo não é totalmente thread-safe (porque não se quis implementar o event bus como IDisposable) o que pode significar, em determinados cenários, que esses threads associados às queues não sejam libertados de forma graciosa.

Nos casos em que isso seja imperioso, o bus disponibiliza um mecanismo específico que controla o seu tempo de vida e trata de libertar a instância de forma totalmente controlada, garantindo a finalização de todos os threads antes da finalização do bus. A implementação assenta no padrão IDisposable naturalmente e deve ser utilizado da seguinte forma:

IEventBus bus = EventBus.Default();
using (IEventBusLifetimeScope scope = bus.BeginLifetimeScope())
{
    // perform operations on the bus here
}

// here the bus still exists but it is stopped

Definição de Event Topics

Passemos então ao comportamento fundamental do event bus e do padrão publisher/subscriber.

Como foi descrito antes, o publisher e os subscribers partilham (por assim dizer) eventos através das notificações da sua ocorrência. A essas notificações chamamos de “event topics” ou simplesmente “tópicos”.

Um tópico é definido como uma classe, declarada pelo publisher, que deve implementar o interface IEventTopic para ser aceite pelo bus e que pode declarar os dados que descrevem esse evento e que os subscribers utilizarão para os processar.

O tópico mais simples é apenas uma notificação de que algo aconteceu. Esse algo é apenas definido pela classe e não existem dados adicionais.

/// <summary>
/// Describes that something particular happenned in the app.
/// </summary>
/// <seealso cref="IEventTopic" />
public class SomethingHappened : IEventTopic
{
}

Tipicamente, no entanto, há interesse em descrever o evento com dados adicionais que podem ser tantos
quantos necessários. Esses dados devem ser propriedades da classe.

/// <summary>
/// Indicates that a new customer was created.
/// </summary>
/// <seealso cref="IEventTopic" />
public class CustomerCreated : IEventTopic
{
    #region Public Properties

    /// <summary>
    /// Gets or sets the customer identifier.
    /// </summary>
    public string CustomerId
    {
        get;
        set;
    }

    #endregion
}

Serialização

Exceto no caso do transporte em memória, os tópicos são serializados em JSON para envio para cada subscriber. Portanto, é crucial que as classes que definem os tópicos estejam devidamente preparadas para serem corretamente serializadas.

Filtragem de Eventos

Por defeito, como nos dois exemplos anteriores, o bus não faz nada aos tópicos recebidos de um publisher exceto enviá-los para os subscribers registados.

Se para um dado evento existirem 3 subscribers registados:

EventBus.Default().Subscribe<CustomerCreated>(subscriber1);
EventBus.Default().Subscribe<CustomerCreated>(subscriber2);
EventBus.Default().Subscribe<CustomerCreated>(subscriber3);

E se um publisher realizer 3 publicações desse evento:

await EventBus.PublishAsync<CustomerCreated>(new CustomerCreated(“C1”));
await EventBus.PublishAsync<CustomerCreated>(new CustomerCreated(“C2”));
await EventBus.PublishAsync<CustomerCreated>(new CustomerCreated(“C3”));

Cada um dos subscribers receberá os 3 eventos, para os clientes C1, C2 e C3.

Obviamente que o bus suporta que determinado subscriber só receba determinados eventos de um dado tipo, com base num filtro que ele próprio define no momento da subscrição.

Todos os tópicos são filtráveis, no sentido, que não é preciso especificar nada no tópico para que os respetivos subscribers forneçam filtros que definirão em runtime se o tópico é ou não recebido por cada subscriber particular.

Consideremos o seguinte exemplo de um tópico:

/// <summary>
/// Indicates that a given entity was modified.
/// </summary>
/// <seealso cref="IFilterable" />
public class EntityModified : IEventTopic
{
    #region Public Properties

    /// <summary>
    /// Gets or sets the name of the entity that was modified.
    /// </summary>
    public string EntityName
    {
        get;
        set;
    }

    /// <summary>
    /// Gets or sets the key of the entity that was modified.
    /// </summary>
    public string EntityKey
    {
        get;
        set;
    }

    #endregion
}

Determinado subscriber pode simplesmente estar interessado em todas as ocorrências de um evento. Para isso bastará que não indice nenhum filtro no momento da subscrição.

Por outro lado, quando se pretender que um subscriber receba apenas determinados eventos, este deve fornecer uma instância de IEventTopicFilter<TEventTopic> na subscrição:

EntityModifiedFilter filter = new EntityModifiedFilter(“Customers”);
EventBus.Default().Subscribe<EntityModified>(filter);

Neste caso, a classe EntityModifiedFilter, o filtro, poderia ser declarada da seguinte forma:

/// <summary>
/// Provides a filter for the <see cref="EntityModified"/> event topic.
/// </summary>
/// <seealso cref="IEventTopicFilter{EntityModified}" />
public class EntityModifiedFilter : IEventTopicFilter<EntityModified>
{
    #region Private Properties

    private string EntityName
    {
        get;
        set;
    }

    #endregion

    #region Constructors

    /// <summary>
    /// Initializes a new instance of the <see cref="EntityModifiedFilter"/> class.
    /// </summary>
    /// <param name="entityName">The name of the entity.</param>
    public EntityModifiedFilter(string entityName)
    {
        this.EntityName = entityName;
    }

    #endregion

    #region Public Methods

    /// <summary>
    /// Determines whether the specified topic should be received by the subscriber.
    /// </summary>
    /// <param name="topic">The event topic.</param>
    /// <returns>
    /// A value indicating whether the specified topic should be received by the subscriber.
    /// If the result is false, the event topic will NOT be received by the subscriber.
    /// </returns>
    public bool Receive(EntityModified topic)
    {
        return this.EntityName.Equals(topic.EntityName, StringComparison.CurrentCultureIgnoreCase);
    }

    #endregion
}

Note-se que o event bus invocará o método Receive para tópico publicado para determinar se deve ou não entregar a notificação a cada subscriber. Cada subscriber fornece a sua própria instância do filtro e por isso controla os eventos que recebe efetivamente.

Filtragem de Eventos por Expressões

Determinados transportes como o Azure Service Bus incluem mecanismos de filtragem baseados em expressões. Esses mecanismos determinam se determinada subscrição receberá ou não determinada mensagem (evento) dependendo do resultado da avaliação da expressão.

A classe Primavera.Core.EventDriven.IEventTopicExpressionFilter<TEventTopic> define um filtro desse tipo que tem uma propriedade com o nome Expression para definir a expressão de filtragem como no exemplo seguinte:

public sealed class ReceiveIfStartsWithAorZ : IEventTopicExpressionFilter<FilteredEventEx>
{
    #region Public Properties

    /// <summary>
    /// Gets the expression that should be used to filter the event topics.
    /// The syntax used in this expression depends on the concrete implementation of the event bus
    /// being used.
    /// </summary>
    public string Expression
    {
        get
        {
            return "Column LIKE 'A%' OR Column LIKE 'Z%'";
        }
    }

    #endregion

    #region Public Methods

    /// <summary>
    /// Determines whether the specified event topic should be received by the subscriber.
    /// </summary>
    /// <param name="topic">The event topic instance.</param>
    /// <returns>
    /// A value indicating whether the specified event topic should be received by the subscriber.
    /// If the result is false, the event topic will NOT be received by the subscriber.
    /// </returns>
    public bool Receive(FilteredEventEx topic)
    {
        return false;
    }

    #endregion
}

Neste caso o bus utilizará a expressão para filtrar o evento em vez do método Receive.

Um filtro deste tipo seria então utilizado pelo subscriber como outro filtro qualquer:

FilteredEventSubscriberEx subscriber1 = new FilteredEventSubscriberEx();
ReceiveIfStartsWithAorZ filter1 = new ReceiveIfStartsWithAorZ();
AzureTopicsEventBus.Instance("AZSB").Subscribe(subscriber1, filter1);

Nomes dos Eventos

Por defeito o event bus usará o nome do tipo que define o tópico para atribuir o nome do tópico (que é utilizado internamente). O nome gerado será um hash sobre o nome desse tipo.

Em determinados cenários se, por exemplo, for ncessário que o nome do tópico seja identificável e legível, pode utilizar-se o atributo Primavera.Core.EventDriven.EventTopicNameAttribute para especificar o nome interno do tópico, como no exemplo seguinte:

[EventTopicName("Cenario1.Event")]
public sealed class FilteredEventEx : IEventTopicWithProperties

Publicação de Eventos

A publicação de um tópico é realizada pelo método Publish (ou PublishAsync) que tem vários overloads diferentes para suportar vários cenários de instanciação do tópico concreto.

MyEvent data = new MyEvent();
// (…)
await EventBus.Default().Publish(data);

A publicação pode ser síncrona (se for utilizado o método Publish) ou assíncrona (se for utilizado o método PublishAsync).

É importante reconhecer as seguintes propriedades do bus:

  • Não há qualquer garantia de que o objeto (a instância) recebido por cada subscriber é o mesmo que foi enviado pelo publisher. No caso dos transportes que usam serialização isso não acontece garantidamente.
  • A garantia é que o objeto é idêntico (tem os mesmos valores nas propriedades serializadas).

Se a publicação dos eventos pode ser síncrona ou assíncrona, ao critério do publisher, isso não tem influência na forma como os subscribers receberão as notificações, que também podem ser recebidas sincrona ou assincronamente, ao critério de cada subscriber. Este aspeto é importante porque leva a que não possa ser assumida nenhuma regra quanto à ordem e o momento em que em cada subscriber receberá um determinado evento.

Subscrição de Eventos

Um subscriber regista-se junto do event bus através do método Subscribe ou SubscribeAsync (e o inverso através de Unsubscribe). Existem vários overloads disponíveis para suportar cenários diferentes de declaração do tópico e do subscriber.

MySubscriber subscriber = new MySubscriber();
EventBus.Default().Subscribe(subscriber).

Declaração do Subscriber

No entanto, para que uma classe possa atuar como subscriber deve implementar a interface IEventSubscriber para o evento.

Esta interface garante a existência de duas prioriedades que determinadam como o evento será recebido pelo subscriber. O subscriber deve apenas devolver um valor numa delas (se devolver nas duas será considerada apenas a primeira).

  • HandleAction: define a ação (o código síncrono) a executar para receber o evento.
  • HandleTask: define a tarefe (o código assíncrono) a executar para receber o evento.
/// <summary>
/// Defines the event subscriber.
/// </summary>
/// <seealso cref="IEventSubscriber{Request}" />
/// <seealso cref="IEventSubscriber{Response}" />
public class Subscriber : IEventSubscriber<Request>, IEventSubscriber<Response>
{
    #region Public Properties

    /// <summary>
    /// Gets the action that will handle the event topic synchronously.
    /// </summary>
    Action<Request> IEventSubscriber<Request>.HandleAction
    {
        get
        {
            return null;
        }
    }

    /// <summary>
    /// Gets the task that will handle the event topic synchronously.
    /// </summary>
    Func<Request, Task> IEventSubscriber<Request>.HandleTask
    {
        get
        {
            return this.HandleRequestAsync;
        }
    }

    /// <summary>
    /// Gets the action that will handle the event topic synchronously.
    /// </summary>
    Action<Response> IEventSubscriber<Response>.HandleAction
    {
        get
        {
            return null;
        }
    }

    /// <summary>
    /// Gets the task that will handle the event topic synchronously.
    /// </summary>
    Func<Response, Task> IEventSubscriber<Response>.HandleTask
    {
        get
        {
            return this.HandleResponseAsync;
        }
    }

    #endregion

    #region Public Methods

    private async Task HandleRequestAsync(Request topic)
    {
        (...)
    }

    private async Task HandleResponseAsync(Response topic)
    {
        (...)
    }

    #endregion
}

Note-se neste exemplo que uma determinada classe pode ser subscriber de N eventos, bastanto para isso implementar o interface IEventSubscriber para cada um desses tópicos.

Sequência de Execução Normal

O diagrama de sequência seguinte ilustra um exemplo de execução normal do event bus que reflete os vários aspetos documentados antes.

CoreLib EventDriven

Persistência de Subscrições

O componente de persistência de subscrições é responsável por armazenar as subscrições ativas para uma instância do bus.

IEventBusSubscriptionStore

O interface deste componente está definido em Primavera.Core.EventDriven.IEventBusSubscriptionStore e podem ser acrescentadas tantas implementações quantas necessárias, ativando depois a implementação concreta através da configuração EventBusConfiguration.SubscriptionStoreType.

InProcessEventBusSubscriptionStore

Esta implementação (Primavera.Core.EventDriven.InProcess.InProcessEventBusSubscriptionStore em Primavera.CoreLib) da persistência de subscrições armazena-as em memória. O que significa que as subscrições apenas estarão registadas durante o tempo de vida da instância do event bus.

AzureTopicsEventBusSubscriptionStore

Esta implementação (Primavera.Core.EventDriven.Azure.AzureTopicsEventBusSubscriptionStore) em Primavera.CoreLib.EventDriven.Azure) regista as subscrições usando tópicos do Microsoft Azure Service Bus.

Implementação de um Novo Provider

Para implementar um novo provider bastará implementar o interface IEventBusSubscriptionStore.

Transporte

O componente de transporte é responsável pelo envio de tópicos entre os publishers e os subscribers.

IEventBusTransport

O interface deste componente é definido em Primavera.Core.EventDriven.IEventBusTransport. Como no caso das subscrições, podem ser adicionadas novas implementações, sendo a ativa identificada pela configuração EventBusConfiguration.TransportType.

InProcessEventBusTransport

Esta implementação (Primavera.Core.EventDriven.InProcess.InProcessEventBusTransport em Primavera.CoreLib.EventDriven) do transporte invoca diretamente as instâncias dos subscribers em memória (por isso diz-se "in process").

AzureTopicsEventBusTransport

Esta implementação (Primavera.Core.EventDriven.Azure.AzureTopicsEventBusTransport em Primavera.CoreLib.EventDriven.Azure) do transporte utiliza tópicos do Microsoft Azure para transferir eventos entre o publisher e os subscribers.

Implementação de um Novo Provider

Para implementar um novo provider bastará implementar o interface IEventBusTransport.

Filtros

Este componente é responsável por filtrar os eventos recebidos pelos subscribers de acordo com os filtros que estes forneceram no momento de subscrição.

IEventBusFilter

O interface deste componente é definido em Primavera.Core.EventDriven.IEventBusFilter. Como no caso do transporte, podem ser adicionadas novas implementações, sendo a ativa identificada pela configuração EventBusConfiguration.FilterType.

InProcessEventBusFilter

Esta implementação (Primavera.Core.EventDriven.InProcess.InProcessEventBusFilter em Primavera.CoreLib) filtra os eventos em memória, invocando diremente o método Receive dos filtros disponibilizados pelos subscribers.

NoOpEventBusFilter

Determinados transportes, como os Azure Service Bus Topics, incluem nativamente processos de filtragem dos eventos pelos subscritores pelo que não é necessário que o event bus trate dessa operação explicitamente.

Esta implementação (Primavera.Core.EventDriven.InternalNoOpEventBusFilter) em (Primavera.CoreLib) deve ser usada nesses casos, para garantir que o event bus não realiza nenhuma operação de filtragem.

In-process versus Out-of-process

Como já se viu nos pontos anteriores, o event bus suporta dois grandes cenários de utilização do padrão publish/subscribe:

  • In-process – quando o publisher e o subscriber executam no mesmo processo.
  • Out-of-process – quando o publisher e o subscriber executam em processos diferentes (na mesma ou em máquinas diferentes).

In-process

Neste cenário, embora não seja obrigatório, o que faz sentido é que tanto o publisher e os subscribers usem a mesma instância do event bus.

Note-se que o transporte em memória apenas suporta este cenário porque a memória não pode ser partilhada entre processos diferentes.

Portanto, o fluxo de execução é basicamente o apresentado anteriormente:

CoreLib EventDriven

Out-of-process

No cenário out-of-process cada “participante” tem a sua própria instância do event bus e, portanto, é necessário que essas instâncias partilhem de alguma forma a informação sobre as subscrições existentes e os eventos publicados.

A instância do publisher armazenará as subscrições e será capaz de entregar os eventos aos subscritores adequados. Quando se utilizarem tópicos do Azure Service Bus, por exemplo, um evento será registado no respetivo tópico e assim encaminhado automaticamente para os subscritores registados através do callback da mensagem (registado pela instância do event bus que lida com cada subscriber).

Exatamente porque este tipo de transporte dos eventos suporta um grande lapso de tempo entre uma subscrição e a receção de um evento desse tipo, sempre que um novo tópico é recebido, é criada uma nova instância do subscriber. Este é um aspeto muito importante do funcionamento do bus out-of-process.

O diagrama seguinte ilustra um possível fluxo de execução neste cenário, detalhando os aspetos referidos anteriormente.

CoreLib EventDriven

Azure Service Bus Topics

Está disponível, como já foi referido antes, uma versão do event bus que utiliza os tópicos do Microsoft Azure Service Bus para troca de eventos entre o publisher e os subscribers.

Apesar do event bus poder ser configurado facilmente para usar os tipos de subscrições e de transporte que utilizam o Azure Service Bus, existe uma classe que simplica a utilização de uma instância do bus com essas características. Trata-se do tipo Primavera.Core.EventDriven.Azure.AzureTopicsEventBus, que pode ser usado com no exemplo seguinte:

FilteredEventSubscriberEx subscriber1 = new FilteredEventSubscriberEx();
FilteredEventSubscriberEx subscriber2 = new FilteredEventSubscriberEx();

ReceiveIfStartsWithAorZ filter1 = new ReceiveIfStartsWithAorZ();
ReceiveIfStartsWithBorZ filter2 = new ReceiveIfStartsWithBorZ();

AzureTopicsEventBus.Instance("AZSB").Subscribe(subscriber1, filter1);
AzureTopicsEventBus.Instance("AZSB").Subscribe(subscriber2, filter2);

AzureTopicsEventBus.Instance("AZSB").Publish(new FilteredEventEx("A200"));

Erros

Regra geral o event bus lancará exceções para todos os erros que comprometem o funcionamento esperado. Normalmente do tipo Primavera.Core.EventDriven.EventBusException. Os erros são tipificados na propriedade ErrorCode para permitir o seu tratamento especializado.

Tracing

O event bus regista um trace detalhado de todas as operações realizadas. Determinados eventos podem corresponder a situações inesperadas (por exemplo, um evento que não está subscrito por nenhum subscriber) pelo que a informação do tracing pode ser muito importante para analisar o comportamento do bus.

Primavera.Core.Events

Esta assembly define os eventos utilizados da Plataforma Elevation e, em conjunto, com os componentes da plataforma que os utilizam, são bons exemplos da correta utilização do event bus.

A documentação do referido assembly poderá ser consultada aqui.

Questões em Aberto

N/A.

FAQ

N/A.