Recentemente em dois vídeos onde abordei a técnica de event sourcing, fiz uso de uma espécie de base de dados que armazena eventos que ocorrem na aplicação, e que é chamada de Event Store. Naqueles exemplos, a ideia era armazenar os eventos gerados pelo domínio e que eram, mais tarde, utilizando para reconstruir o estado atual do objeto em questão.
Como lá abordamos o Event Store de forma bem superficial, pois o foco era no conceito e não na ferramenta, neste artigo vamos aprofundar um pouco mais para explorar a API e os recursos que este repositório pode oferecer para armazenarmos e extrairmos os eventos para uso em cenários de event sourcing ou também em cenários que demandam um service bus, onde podemos publicar conteúdo e os assinantes possam ser notificados e/ou interrogar a base para saber se algo ocorreu e tomar alguma decisão sobre isso.
O primeiro passo é fazer o download através do site oficial. Este pacote trará o executável que é a “parte servidor”, ou seja, a própria base onde os eventos serão armazenados. Além disso, há também uma console administrativa web, que é acessada através de um navegador e permitirá interagir com a base e realizar diversas configurações de funcionamento e monitoramento.
Para exemplificar o uso de alguns recursos o Event Store, vamos criar três aplicações: Loja, Nota Fiscal e Transportadora. Na primeira, o cliente irá postar o pedido de compra. A segunda, Nota Fiscal, irá interrogar periodicamente o servidor procurando por novos pedidos adicionados e, consequentemente, emitir a nota fiscal. Por fim, a Transportadora irá monitorar qualquer nova nota fiscal emitida, e irá imediatamente iniciar o processo de entrega do produto.
O primeiro passo é iniciar o servidor que passará a receber as solicitações. Basicamente depois de extraído o pacote do download, basta executar o comando abaixo no prompt de comando e ele passará a receber as requisições. E depois do servidor inicializado, você pode ir até a console web administrativa através do seguinte endereço: http://localhost:2113, login “admin” e a senha “changeit”.
C:EventStore.ClusterNode.exe
Depois do servidor rodando, precisamos começar a codificar as aplicações, e para isso, o projeto também fornece diversas APIs para diferentes tecnologias, incluindo para .NET que é chamada de EventStore.Client. Utilize o comando abaixo para baixar e instalar ele através do Nuget:
PM> Install-Package EventStore.Client
Depois das três aplicações estarem com o pacote do cliente devidamente instalado, chega o momento de estabelecermos a conexão com o servidor para postar o evento de novo pedido criado na loja. A classe que serve para estabelecer a ligação com o servidor é a EventStoreConnection, e pode (e deve) ser mantida uma por aplicação para melhor reutilização dos recursos que a mesma utiliza para executar suas atividades. É importante notar que os métodos expostos são assíncronos, mas como estou utilizando aplicações console para o exemplo, forçarei a execução síncrona de todos eles.
A conexão se dá através do protocolo TCP, onde você pode utilizar o método estático Create para informar o endereço a partir de uma URI, e a porta padrão para o servidor é a 1113. Como dito acima, estou abrindo a conexão explicitamente e aguardando que a mesma seja concluída para dar sequência na utilização.
using (var conn = EventStoreConnection.Create(new Uri(“tcp://localhost:1113”)))
{
conn.ConnectAsync().Wait();
}
Voltando ao exemplo de testes, a primeira aplicação se resume a postar um novo evento informando que um novo pedido foi criado. Note através do código abaixo como isso é definido. Não vou me preocupar em colocar a definição das classes aqui porque são muito simples e não há qualquer informação relevante dentro delas; são classes POCOs com as propriedades que refletem nosso negócio.
var novoPedido = new NovoPedidoCriado(new Pedido()
{
NomeDoCliente = “Israel Aece”,
ValorTotal = 1200
});
Depois do evento criado, precisamos postar o mesmo, mas antes precisamos entender um conceito do Event Store, chamado de Stream. Como já suspeitamos, é a forma que temos de agrupar os eventos por qualquer regra ou seção da aplicação que desejamos. Para o nosso caso, vamos chamar o stream de “Ecommerce.Pedidos”, que irá concentrar todos os eventos relacionados aos pedidos realizados em nosso loja.
A classe de conexão fornece um método chamado AppendToStreamAsync, e além de especificarmos o nome do stream onde queremos armazenar, temos que passar o objeto (evento) que deve ser persistido. Para encapsular e descrever o evento do ponto de vista da infraestrutura, o Event Store possui uma classe chamada EventData. Essa classe além de ter algumas características relacionadas a infraestrutura, recebe também em seu construtor o Id do evento, o nome e o objeto que representa o mesmo, e que no nosso caso está armazenado na variável “novoPedido”.
A serialização pode ser binária, mas dependendo com quem vamos interagir/integrar, utilizar JSON pode ser uma opção bem mais interessante. E para isso, estou recorrendo ao Json.NET (Newtonsoft) para serializar e deserializar os objetos.
conn.AppendToStreamAsync(
“Ecommerce.Pedidos”,
ExpectedVersion.Any,
GerarEvento(novoPedido)).Wait();
private static EventData GerarEvento(Evento evento)
{
return new EventData(
evento.Id,
evento.GetType().Name,
true,
Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(evento)), null);
}
Depois que o evento for postado, se acessarmos a console administrativa do Event Store, veremos o stream e o evento criado, conforme é possível visualizar na imagem abaixo:
Na sequência vamos recorrer ao projeto de Nota Fiscal, que ficará interrogando periodicamente o servidor em busca de eventos do tipo NovoPedidoCriado. Como os eventos são acumulativos do lado do servidor, nós, o cliente, devemos controlar os eventos que já foram processados pela aplicação, e por isso, estamos armazenando em uma variável o número do último evento, para que em uma próxima execução não corra o risco de gerar notas fiscais para os pedidos já gerados. E vale ressaltar que isso é um exemplo, e que em uma cenário real você precisa armazenar este contador em uma local seguro que resista a possíveis reinicializações da aplicação.
Da mesma forma que fazemos na criação do evento, para ler também precisamos criar a conexão com o servidor, e neste caso, vamos utilizar o método ReadStreamEventsForwardAsync, e como o próprio nome diz, lê os eventos na ordem em que eles foram postados.
var ultimoEventoProcessado = 0;
using (var conn = EventStoreConnection.Create(new Uri(“tcp://localhost:1113”)))
{
conn.ConnectAsync().Wait();
while ((Console.ReadLine() != null))
{
var itens = conn.ReadStreamEventsForwardAsync(
“Ecommerce.Pedidos”,
ultimoEventoProcessado + 1, 200, false).Result;
if (itens.Events.Any())
{
foreach (var e in itens.Events)
Processar(conn, e);
}
else
{
Log.Yellow(“Não há eventos para processar.”);
}
}
}
Os eventos extraídos são representados pela classe ResolvedEvent, que entre várias informações sobre o evento, temos o evento gerado e serializado pela nossa aplicação de novo pedido criado. Como podemos notar abaixo, o método Processar identifica que se trata de um evento deste tipo, deserializa o mesmo, o processa e depois disso, dispara um evento dizendo que a respectiva nota fiscal foi emitida, através de um novo evento.
internal static void Processar(
IEventStoreConnection conexao,
ResolvedEvent dadosDoEvento)
{
if (dadosDoEvento.Event.EventType == “NovoPedidoCriado”)
{
var novoPedido = Extrair<NovoPedidoCriado>(dadosDoEvento);
Log.Green(“NOVO PEDIDO”);
Log.Green(“Id: ” + novoPedido.Id, 4);
Log.Green(“Data: ” + novoPedido.Data, 4);
Log.Green(“Cliente: ” + novoPedido.NomeDoCliente, 4);
Log.Green(“Valor: ” + novoPedido.ValorTotal.ToString(“N2”), 4);
Log.Yellow(“Emitindo a Nota Fiscal do Pedido”, 4);
Log.NewLine();
conexao.AppendToStreamAsync(
“Ecommerce.Pedidos”,
ExpectedVersion.Any,
GerarEvento(
new NotaFiscalEmitida(
novoPedido.NomeDoCliente,
novoPedido.ValorTotal,
“0001.0292.2999-2881-9918.11.9999/99”))).Wait();
ultimoEventoProcessado = dadosDoEvento.Event.EventNumber;
}
}
private static TEvento Extrair<TEvento>(ResolvedEvent dadosDoEvento)
where TEvento : Evento
{
return JsonConvert.DeserializeObject<TEvento>(
Encoding.UTF8.GetString(dadosDoEvento.Event.Data));
}
E, para finalizar o processo, temos agora a aplicação da transportadora, que ao invés de periodicamente procurar por um evento de nota fiscal emitida, assina o stream e tão logo quando ele for incluído, a transportadora será automaticamente notificada do evento gerado. E da mesma forma que antes, abrimos a conexão, e agora utilizamos o método SubscribeToStreamAsync, informando o nome do stream que desejamos monitorar e qualquer novo evento, disparamos o método Processar, conforme pode ser visualizado abaixo:
using (var conn = EventStoreConnection.Create(new Uri(“tcp://localhost:1113”)))
{
conn.ConnectAsync().Wait();
conn.SubscribeToStreamAsync(
“Ecommerce.Pedidos”,
false,
(a, e) => Processar(e)).Wait();
Console.ReadLine();
}
internal static void Processar(ResolvedEvent dadosDoEvento)
{
if (dadosDoEvento.Event.EventType == “NotaFiscalEmitida”)
{
var novoPedido = Extrair<NotaFiscalEmitida>(dadosDoEvento);
Log.Green(“NOVO PEDIDO – NOTA FISCAL EMITIDA”);
Log.Green(“Id: ” + novoPedido.Id, 4);
Log.Green(“Cliente: ” + novoPedido.NomeDoCliente, 4);
Log.Green(“Valor: ” + novoPedido.ValorTotal.ToString(“N2”), 4);
Log.Green(“NF-e: ” + novoPedido.ChaveDaNotaFiscalEletronica, 4);
Log.NewLine();
}
}
E agora, se colocarmos as três aplicações lado a lado, é possível visualizar o efeito do processamento nas três, onde uma gera o pedido, a segunda emite a nota fiscal e a terceira é notificada para iniciar o processo de transporte.
É claro que isso é um exemplo simplista, mas pode ser considerado algo parecido em uma escala maior, como por exemplo, utilizar este recurso para diálogo entre contextos do domínio, sincronização de bases de (escrita e leitura) em ambiente de CQRS ou para desempenhar qualquer outra tarefa em modo assíncrono.
Israel Aece 