eventos

Home > Barramento de eventos 🚧

Barramento de eventos

Important

Em desenvolvimento (2.0). Ainda não está numa versão lançada e pode mudar.

Vale para a 2.0 · Delphi XE em diante e Lazarus/FPC 3.2 em diante · Nível: intermediário

Em uma frase: o TRALEventBus deixa partes do programa publicarem eventos em tópicos (pedido/criado, estoque/sp/baixo) e outras assinarem esses tópicos, sem uma conhecer a outra; com a ponte MQTT, os mesmos eventos saem para outros programas.

Você vai precisar: o pacote RALModuleEventBusPackage (só requer PascalRAL); para a ponte, RALModuleEventBusMQTTPackage (requer o MQTT).

Publicar e assinar

uses RALEventBus;

  FEventos := TRALEventBus.Create(Self);
  FEventos.Subscribe('pedido/#', PedidoMudou);        // pedido e tudo abaixo dele
  FEventos.Subscribe('estoque/+/baixo', EstoqueBaixo); // + é um nível

  FEventos.Publish('pedido/criado', '{"id":1234}');

procedure TForm1.PedidoMudou(Sender: TObject; AEvent: TRALEvent);
begin
  Log(AEvent.Topic + ': ' + AEvent.Text);
end;

• Os tópicos são como os do MQTT: níveis separados por /, + vale um nível e # vale o nível em que está e tudo abaixo.
• O evento leva Payload (bytes; Text lê como texto UTF-8) e, opcionalmente, um objeto em Data (não copiado nem liberado pelo barramento), Origin (quem publicou) e Timestamp. O handler lê o evento durante a chamada; não guarde o objeto.
• Publish devolve para quantos assinantes o evento foi.
• No Object Inspector, Subscriptions lista assinaturas com Filter, OnEvent e Enabled.

Onde os handlers rodam

• Delivery = edSync (padrão): na thread de quem publicou, um assinante depois do outro, antes de Publish voltar.
• Delivery = edAsync: em Workers threads do barramento (0 = uma por processador). Cada assinante recebe os eventos na ordem em que foram publicados e uma vez só; assinantes diferentes rodam ao mesmo tempo. Na tela, use TThread.Queue.
• MaxQueue limita a fila de cada thread: o que não cabe é descartado para aquele assinante e OnDropped avisa.
• Uma exceção num handler vai para OnError; os outros assinantes recebem o evento normalmente.

Saindo da assinatura

destructor TPainel.Destroy;
begin
  FEventos.UnsubscribeObject(Self);   // tudo que este objeto assinou
  inherited;
end;

Unsubscribe(id) e UnsubscribeObject esperam terminar uma chamada do handler que esteja rodando em outra thread, e eventos que ainda estavam na fila não chegam mais: depois deles, o objeto pode ser liberado com segurança. WaitIdle espera as threads entregarem tudo.

Handlers anônimos (Delphi 2010 em diante)

uses RALEventBus, RALEventBusAnon;

  vId := FEventos.SubscribeProc('pedido/#',
    procedure(AEvent: TRALEvent)
    begin
      Log(AEvent.Text);
    end);
  ...
  FEventos.UnsubscribeProc(vId);

No Lazarus/FPC 3.2 a unit fica vazia: use métodos.

Ponte com o MQTT

uses RALEventBusMQTT;

  FPonte := TRALEventBusMQTT.Create(Self);
  FPonte.Bus := FEventos;
  FPonte.Client := FMQTT;                 // um TRALMQTTClient conectado ao broker
  FPonte.OutFilters.Add('pedido/#');      // o que sai do programa
  FPonte.InFilters.Add('estoque/#');      // o que entra no barramento
  FMQTT.Subscriptions.Add('estoque/#', mqQoS1);

• O cliente MQTT assina o que quer receber (Subscriptions); a ponte só escolhe o que entra no barramento.
• O que entrou pela ponte nunca sai de novo por ela. Se o cliente assina um tópico em que ele mesmo publica, a mensagem volta do broker: use filtros que não se cruzem, ou a opção NoLocal do MQTT 5 na assinatura.

Veja também

• MQTT, Serviços (contêiner), Fila de jobs