W poprzednich wpisach używałem dwóch sposobów deklaracji wiadomości:

public class PersonAdded : IEvent
{
   public Guid PersonId { get; set; }
}
public class AddPerson:IMessage
{
   public Guid PersonId { get; set; }
   public string FirstName { get; set; }        
   public string LastName { get; set; }
}

Istnieje jeszcze trzeci interfejs, ICommand:

public class AddPerson:ICommand
{
   public Guid PersonId { get; set; }
   public string FirstName { get; set; }        
   public string LastName { get; set; }
}

Wszystkie one służą do definicji wiadomości. Jaka jest więc różnica? IMessage jest najbardziej ogólny. Wiadomość zdefiniowana za pomocą IMessage może być zarówno zdarzeniem jak i komendą.

Komendy nie mogą być publikowane (Publish) i subskrybowane (Subscribe). Mogą z kolei być wysłane za pomocą Bus.Send jak to widzieliśmy w jednym z pierwszych wpisów.

Z kolei IEvent (zdarzenia), analogicznie mogą być publikowane i subskrybowane, ale nie można ich wysyłać za pomocą Bus.Send.

IMessage jest przydatny dla Bus.Reply, gdzie zwrotna wiadomość może być zarówno komendą jak i zdarzeniem.

Dziś o kolejnym, moim zdaniem jednym z najważniejszych elementów programowania rozproszonego, a mianowicie o zdarzeniach. Jak w każdej analogicznej technologii, mamy zaimplementowany wzorzec obserwator. Każdy węzeł może nasłuchiwać konkretnej wiadomości i jak tylko zostanie ona opublikowana, wszyscy subskrybenci zostaną poinformowani.

Zacznijmy od przykładu. W poprzednich wpisach stworzyliśmy Example.API oraz klienta. Dzisiaj rozszerzymy solucję o dwa nowe projekty: Subscriber1 oraz Subscriber2. Skorzystamy z wiadomości AddPerson oraz dodamy pierwsze zdarzenie PersonAdded:

public class PersonAdded : IEvent
{
   public Guid PersonId { get; set; }
}
public class AddPerson:IMessage
{
   public Guid PersonId { get; set; }
   public string FirstName { get; set; }        
   public string LastName { get; set; }
}

Zamiast IMessage, implementujemy IEvent. To właśnie te zdarzenie będziemy publikować.  Następnie modyfikujemy AddPersonHandler z Example.API:

class AddPersonHandler : IHandleMessages<AddPerson>
{
   private readonly IBus _bus;

   public AddPersonHandler(IBus bus)
   {
       _bus = bus;
   }

   public void Handle(AddPerson message)
   {
       Console.WriteLine("Adding person. {0} {1}",message.FirstName,message.LastName);
       _bus.Publish(new PersonAdded(){PersonId = message.PersonId});
   }
}

Jak widać publikacja zdarzeń jest bardzo prosta. Kolejny etap to implementacja subskrybentów. Musimy stworzyć dwa nowe projekty oraz standardowo zainstalować nServiceBus, zainicjalizować EndPointCofig. Niczym to się nie różni od Example.API więc nie będę tego powtarzał. W pliku konfiguracyjnym jednak, musimy określić jakie wiadomości będą nasłuchiwane. Skorzystamy z konfiguracji klienta i powinno to wyglądać następująco:

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <configSections>
    <section name="MessageForwardingInCaseOfFaultConfig" type="NServiceBus.Config.MessageForwardingInCaseOfFaultConfig, NServiceBus.Core" />
    <section name="UnicastBusConfig" type="NServiceBus.Config.UnicastBusConfig, NServiceBus.Core" />
  </configSections>
  <startup>
    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.5" />
  </startup>
  <MessageForwardingInCaseOfFaultConfig ErrorQueue="error" />
  <UnicastBusConfig>
    <MessageEndpointMappings>
      <add Assembly="Example.Contracts" Endpoint="Example.API@localhost" />
    </MessageEndpointMappings>
  </UnicastBusConfig>
</configuration>

Obsługa zdarzeń niczym nie różni się od obsługi innych wiadomości zatem handler wygląda tak:

public class AddPersonEventHandler:IHandleMessages<PersonAdded>
{
   public void Handle(PersonAdded message)
   {                        
       Console.WriteLine("Subscriber 1. Person added: {0}",message.PersonId);
   }
}

Oczywiście analogicznie sytuacja wygląda dla innych subskrybentów.  Możliwa jest również ręczna subskrypcja za pomocą:

Bus.Subscribe<PersonAdded>();    
Bus.Unsubscribe<PersonAdded>();

Jeśli chcemy zablokować autosubskrypcję to:

configuration.DisableFeature<AutoSubscribe>();

W praktyce jednak wygodniej jest korzystać z pliku konfiguracyjnego + autosubskrypcji, jak to zostało pokazane wcześniej.

Na początek taki prosty przykład wystarczy. W przyszłym wpisie, pokażę jak to działa od strony wewnętrznej bo oczywiście nie jest to zwykłe zdarzenie rozproszone…

Dzisiaj o zadaniu, które jest powszechne dla systemów rozproszonych, zwłaszcza tych opartych o kolejki. Załóżmy, że chcemy wysyłać emaile co 24h.  Najpierw definiujemy wiadomość:

public class SendEmail : IMessage
{
   public string Email { get; set; }
}

Kolejnym krokiem jest implementacja handlera:

public class SendEmailHandler : IHandleMessages<SendEmail>
{
   public void Handle(SendEmail message)
   {
       Console.WriteLine("{0}:{1}", DateTime.Now.ToLongTimeString(), message.Email);
   }
}

Jeśli chcemy co 5 sekund umieszczać powyższą wiadomość na kolejce, to możemy:

_schedule.Every(TimeSpan.FromSeconds(10), () => _bus.SendLocal(new SendEmail { Email = Guid.NewGuid().ToString() }));

Zmienna _schedule to instancja klasy Schedule. Najlepszym momentem na wywołanie tego jest prawdopodobnie start aplikacji:

public class ScheduleMyTasks : IWantToRunWhenBusStartsAndStops
{
   private readonly IBus _bus;
   private readonly Schedule _schedule;

   public ScheduleMyTasks(IBus bus, Schedule schedule)
   {
       _bus = bus;
       _schedule = schedule;
   }

   public void Start()
   {
       _schedule.Every(TimeSpan.FromSeconds(10), () => _bus.SendLocal(new SendEmail()));
   }

   public void Stop()
   {
   }
}

Implementując interfejs IWantToRunWhenBusStartsAndStops możemy wykonać własny kod w momencie startu nServiceBus. Nic  więcej nie musimy konfigurować – klasa sama zostanie znaleziona przez framework.

Należy mieć na uwadze, że scheduler umieści wiadomość w kolejce co określony czas, ale nie jest to jednoznaczne z wykonaniem kodu. Jeśli sporo wiadomości czeka na handler’a to oczywiście może to opóźnić się.

Tak jak to w przypadku wszystkich wiadomości, gdy obsługa zakończy się wyjątkiem, wtedy zostanie ona powtórzona kilkakrotnie (pisałem w poprzednich postach o mechanizmie wznowień). Uruchamiając przykład zobaczymy zatem:

Dzisiaj o implementacji wzorca saga w nServiceBus. Korzystamy z niego, jeśli mamy kilka procesów biznesowych, które współtworzą jeden rozbudowany workflow. Załóżmy, że w celu złożenia zamówienia, należy wykonać kilka operacji takich jak weryfikacja danych, zapisanie danych w bazie, wysłanie emaila itp. Jeśli operacje są niezależne od siebie, wtedy wystarczy wysłać odpowiednie wiadomości do odizolowanych od siebie handlerów, tak jak to miało miejsce w poprzednich wpisach.

Co jeśli pewien stan jest współdzielony? Wtedy wspomniane podejście nie zadziałała. Rozwiązaniem jest wzorzec sagi. Można ją porównać do transakcji z tym, że nie ma tutaj koncepcji ACID. Jedynie co mamy to stan, który będzie współdzielony przez wszystkie etapy sagi. Z punktu widzenia nServiceBus, jest to klasa, która obsługuje kilka wiadomości naraz. Będziemy mieli zatem kilka implementacji metody Handle, dla różnych wiadomości. Oprócz tego, stan jest współdzielony między wszystkie handlery i zapisywany odpowiednio w bazie danych przez nServiceBus. W programowaniu rozproszonym nie wiemy, ile czasu ten stan będzie musiał być przechowywany w pamięci. Wywoływanie zewnętrznych serwisów może potrwać od sekundy po kilka godzin lub dni. Jak wspomniałem wcześniej, główna koncepcja nServiceBus i kolejek to  zagwarantowanie, że każda wiadomość zostanie dostarczona, nawet w przypadku tymczasowej awarii sieci. Z tego względu, niezbędne jest zapisanie stanu w bazie danych, a nie tylko do pamięci podręcznej.

Myślę, że przykład wyjaśni najlepiej powyższy wzorzec, który wciąż nie jest bardzo znany. Zacznijmy od deklaracji stanu, który będzie współdzielony przez wszystkie etapy sagi:

public class MySagaData : ContainSagaData
{
   [Unique]
   public Guid SagaId { get; set; }

   public string CustomField1 { get; set; }
   public string CustomField2 { get; set; }
   public string CustomField3 { get; set; }
}

W nServiceBus każdy stan musi dziedziczyć po ContainSagaData. Saga również powinna mieć identyfikator. W końcu chcemy współdzielić stan w ramach konkretnej sagi. Innymi słowy,  gdy rozpoczniemy dwie niezależne od siebie sagi, powinny one mieć osobne stany.

Samą sagę definiujemy, poprzez dziedziczenie po klasie Saga:

class SagaExample : Saga<MySagaData>
{
   protected override void ConfigureHowToFindSaga(SagaPropertyMapper<MySagaData> mapper)
   {
       throw new NotImplementedException();
   }
}

W ConfigureHowToFindSaga definiuje się korelacje między wiadomościami a sagą. Następnie musimy zdefiniować co rozpocznie daną sagę. W nServiceBus będzie to oczywiście wiadomość np.:

public class AddPerson:IMessage
{
   public Guid PersonId { get; set; }
}

Za pomocą interfejsu IAmStartedByMessages, informujemy, że dana wiadomość (AddPerson) zawsze rozpocznie nową sagę:

class SagaExample : Saga<MySagaData>, IAmStartedByMessages<AddPerson>
{
   protected override void ConfigureHowToFindSaga(SagaPropertyMapper<MySagaData> mapper)
   {
       throw new NotImplementedException();
   }

   public void Handle(AddPerson message)
   {
       throw new NotImplementedException();
   }
}

Kolejnym etapem jest zdefiniowanie kilku wiadomości, które będą współtworzyć sagę. W naszym przypadku:

public class FirstNameMessage : IMessage
{
   public Guid PersonId { get; set; }
   public string FirstName { get; set; }
}

public class LastNameMessage : IMessage
{
   public Guid PersonId { get; set; }
   public string LastName { get; set; }
}

Saga zatem aktualnie wygląda następująco:

class SagaExample : Saga<MySagaData>, IAmStartedByMessages<AddPerson>, IHandleMessages<FirstNameMessage>,IHandleMessages<LastNameMessage>
{
   protected override void ConfigureHowToFindSaga(SagaPropertyMapper<MySagaData> mapper)
   {
       throw new NotImplementedException();
   }

   public void Handle(AddPerson message)
   {
       throw new NotImplementedException();
   }

   public void Handle(FirstNameMessage message)
   {
       throw new NotImplementedException();
   }

   public void Handle(LastNameMessage message)
   {
       throw new NotImplementedException();
   }
}

Innymi słowy, saga składa się ze stanu, wiadomości rozpoczynającej proces (sagę) oraz wiadomości składających się na nią. Musimy teraz skonfigurować korelację między poszczególnymi wiadomościami a współdzielonym stanem. W naszym przypadku PersonID będzie określał sagę.

We wspomnianej metodzie ConfigureHowToFindSaga, definiujemy mapowanie:

protected override void ConfigureHowToFindSaga(SagaPropertyMapper<MySagaData> mapper)
{
  mapper.ConfigureMapping<FirstNameMessage>(s => s.PersonId).ToSaga(m => m.SagaId);
  mapper.ConfigureMapping<LastNameMessage>(s => s.PersonId).ToSaga(m => m.SagaId);
}

Następnie, w obsłudze wiadomości, która rozpoczyna sagę inicjalizujemy stan:

public void Handle(AddPerson message)
{
  Data.SagaId = message.PersonId;
  Data.CustomField1 = "any value";
  Data.CustomField2 = "any value 1";
  Data.CustomField3 = "any value 2";
}

Najważniejsze pole to SagaID. Reszta jest opcjonalna i może zostać wypełniona na każdym etapie sagi. Zaimplementujmy pozostałe wiadomości w następujący sposób:

class SagaExample : Saga<MySagaData>, IAmStartedByMessages<AddPerson>, IHandleMessages<FirstNameMessage>, IHandleMessages<LastNameMessage>
{
   protected override void ConfigureHowToFindSaga(SagaPropertyMapper<MySagaData> mapper)
   {
       mapper.ConfigureMapping<FirstNameMessage>(s => s.PersonId).ToSaga(m => m.SagaId);
       mapper.ConfigureMapping<LastNameMessage>(s => s.PersonId).ToSaga(m => m.SagaId);
   }

   public void Handle(AddPerson message)
   {
       Data.SagaId = message.PersonId;
       Data.CustomField1 = "any value";
       Data.CustomField2 = "any value 1";
       Data.CustomField3 = "any value 2";

       Console.WriteLine("Handling AddPerson");
       PrintState();
   }

   public void Handle(FirstNameMessage message)
   {
       Data.CustomField2 = message.FirstName;

       Console.WriteLine("Handling FirstNameMessage");
       PrintState();       
   }

   public void Handle(LastNameMessage message)
   {
       Data.CustomField3 = message.LastName;

       Console.WriteLine("Handling LastNameMessage");
       PrintState();
   }

   private void PrintState()
   {
       Console.WriteLine(Data.SagaId);
       Console.WriteLine(Data.CustomField1);
       Console.WriteLine(Data.CustomField2);
       Console.WriteLine(Data.CustomField3);
       Console.WriteLine();
   }
}

Spróbujmy przetestować naszą sagę wysyłając serie następujących wiadomości:

var busConfiguration = new BusConfiguration();
busConfiguration.UsePersistence<RavenDBPersistence>();
ISendOnlyBus bus = Bus.CreateSendOnly(busConfiguration);

Guid personId = Guid.NewGuid();
bus.Send(new AddPerson() {PersonId = personId});
bus.Send(new FirstNameMessage() { PersonId = personId,FirstName = "Piotr"});
bus.Send(new LastNameMessage() { PersonId = personId, LastName = "Zielinski" });

Prześledźmy co po kolei ma miejsce:

Najpierw wysyłamy AddPerson, co inicjuje nową sagę. Następnie odbierane są FirstNameMessage oraz LastNameMessage. Widzimy wyraźnie, że stan jest przechowywany pomiędzy kolejnymi wiadomościami.

Spróbujmy teraz wysłać FirstNameMessage oraz LastNameMessage z identyfikatorem sagi, która nie istnieje tzn.:

bus.Send(new AddPerson() {PersonId = Guid.NewGuid()});            
Thread.Sleep(5000);
bus.Send(new FirstNameMessage() { PersonId = Guid.NewGuid(),FirstName = "Piotr"});
Thread.Sleep(5000);
bus.Send(new LastNameMessage() { PersonId = Guid.NewGuid(), LastName = "Zielinski" });

Za każdym razem generujemy nowy identyfikator. Z tego względu tylko AddPerson zostanie obsłużone bo ta wiadomość rozpoczyna nową sagę. Następne wiadomości zakończą się błędem ponieważ posiadają ID nieistniejącej sagi:

Zróbmy jeszcze jeden eksperyment. Co się stanie jeśli wyślemy trzy razy tą samą wiadomość?

Guid personId = Guid.NewGuid();

bus.Send(new AddPerson() { PersonId = personId });
bus.Send(new FirstNameMessage() { PersonId = personId, FirstName = "Piotr1" });
bus.Send(new FirstNameMessage() { PersonId = personId, FirstName = "Piotr2" });
bus.Send(new FirstNameMessage() { PersonId = personId, FirstName = "Piotr3" });

Wynik:

Jak widzimy, wszystkie wiadomości zostały prawidłowo obsłużone. Wynika to z tego, że nie zamknęliśmy sagi. Można tego dokonać za pomocą metody MarkAsCompleted. Wysłanie kilkukrotne FirstNameMessage poskutkuje teraz:

Jak widzimy, tylko raz wiadomość zostanie obsłużona. Kolejne wywołania prowadzą do błędu, ponieważ po pierwszym odebraniu FirstNameMessage, saga jest zakończona. W moich postach korzystam z RavenDB i zaglądając do bazy, przekonamy się, że stan sagi faktycznie jest tam przechowywany:

Dla przypomnienia, bazę konfigurowaliśmy (patrz poprzednie wpisy) w EndPointConfiguration:

public class EndpointConfig : IConfigureThisEndpoint, AsA_Server
{
   public void Customize(BusConfiguration configuration)
   {
      configuration.UsePersistence<RavenDBPersistence>();
   }
}

Zaglądając do bazowej klasy ContainSagaData zobaczymy:

public abstract class ContainSagaData : IContainSagaData
  {
    /// <summary>
    /// The saga id
    /// 
    /// </summary>
    public virtual Guid Id { get; set; }
    /// <summary>
    /// The address io the endpoint that started the saga
    /// 
    /// </summary>
    public virtual string Originator { get; set; }
    /// <summary>
    /// The id of the message that started the saga
    /// 
    /// </summary>
    public virtual string OriginalMessageId { get; set; }
  }

Widzimy, że dzięki nServiceBus wiemy kto rozpoczął sagę za pomocą Originator oraz OriginalMessageId.  W naszym przypadku jest to klient, ale zwykle jest to jakiś endpoint. Z tego względu, możemy odpowiedzieć bezpośrednio mu za pomocą:

 ReplyToOriginator(new AnyMessage { FirstName = Data.CustomField2 });

Mamy również do dyspozycji timeout. Jeśli wiemy, że obsługa sagi nie ma sensu, gdy przetwarzanie zajęło dłużej niż określony czas, wtedy możemy zaimplementować interfejs IHandleTimeouts:

 class SagaExample : Saga<MySagaData>, IAmStartedByMessages<AddPerson>, 
        IHandleMessages<FirstNameMessage>, 
        IHandleMessages<LastNameMessage>,
        IHandleTimeouts<MyCustomTimeout>
    {
    // ...
    }

Kolejnym krokiem jest wywołanie RequestTimeout z określonym czasem:

public void Handle(AddPerson message)
{
  Data.SagaId = message.PersonId;
  RequestTimeout<MyCustomTimeout>(TimeSpan.FromHours(1));
}

W tym przypadku jest to 1h. Po tym czasie, zostanie wywołana metoda Timeout (część interfejsu IHandleTimeouts):

public void Timeout(MyCustomTimeout state)
{
    ReplyToOriginator(new ErrorMessage(){PersonId=Data.SagaId});
}

W powyższym przykładzie,  wywołuję ReplyToOrignator co ma sens, ponieważ chcemy poinformować handler, który rozpoczął sagę o timeout.

MyCustomTimeout to zwykła struktura danych. Zaglądając do sygnatur RequestTimeout przekonamy się, jak jest ona wypełniana:

protected void RequestTimeout<TTimeoutMessageType>(TimeSpan within, Action<TTimeoutMessageType> messageConstructor) where TTimeoutMessageType : new();

W dzisiejszym wpisie pokażę jak nServiceBus radzi sobie z błędami. W środowisku rozproszonych, opartym o kolejki, możliwe jest, że będziemy mieli do czynienia z tymczasowymi błędami. Może to być np. timeout związany z próbą połączenia się z usługą lub deadlock w bazie danych.  Z tego względu, nServiceBus posiada kilkuwarstwowy mechanizm wznawiania wiadomości.

Jeśli przetwarzanie zakończy się błędem, wiadomość nie jest od razu umieszczana w kolejce errors. Najpierw handler jest powtarzany przez określoną liczbę razy (domyślnie 5). Dzięki temu, tymczasowe błędy takie jak deadlock czy timeout mogą zostać zniwelowane. Oczywiście, czasami natychmiastowa ponowna próba jest zbyt szybka. Z tego względu, istnieje druga warstwa mechanizmu recovery. Jeśli pierwsze 5 prób zakończy się niepowodzeniem, wtedy nServiceBus będzie robił przerwy przed kolejnymi próbami i stanowi to drugą warstwę recovery.

Stwórzmy handler, który będzie wyrzucał wyjątek:

class AddPersonHandler:IHandleMessages<AddPerson>
{
   public void Handle(AddPerson message)
   {
       Console.WriteLine(DateTime.Now.ToLongTimeString());
       
       throw new Exception();
   }
}

Na początku, nServiceBus będzie próbował zastosować mechanizm z pierwszej warstwy co poskutkuje pięciokrotnym wykonaniem Handle, jeden po drugim. Jeśli chcemy zmienić wartość domyślną powtórzeń, możemy to w pliku konfiguracyjnym określić:

<MsmqTransportConfig MaxRetries="5" />

Następnie zostanie zastosowana druga warstwa, która ma na celu naprawienie błędów tymczasowych, które jednak mogą potrwać kilka sekund. Domyślne zachowanie można skonfigurować za pomocą:

<SecondLevelRetriesConfig Enabled="true" TimeIncrease="00:00:10" NumberOfRetries="3" />

Widzimy, że będziemy mieli trzy próby, w odstępach 10, 20 oraz 30 sekund (TimeIncrease). Innymi słowy, każda kolejna próba wiążę się z coraz dłuższą przerwą.

Jeśli błąd jest permanentny, wtedy naturalnie żaden mechanizm retry nie zadziała i wiadomość zostanie przekazana do kolejki błędów (errors). Można również skonfigurować, która kolejka będzie zawierała  wiadomości, które nie można było przetworzyć za pomocą:

<MessageForwardingInCaseOfFaultConfig ErrorQueue="error"/>

Możemy zajrzeć do Computer Management i przekonać się, że faktycznie nasza wiadomość została umieszczona w kolejce errors:

Należy uważać, na konfiguracje liczby ponowień, ponieważ zbyt wysoka liczba może spowodować spore problemy wydajnościowe. Nie chcemy w końcu w nieskończoność ponawiać wykonania jeśli wiemy, że błędy są permanentne. Metoda w końcu może zawierać skomplikowaną logikę i inne czasochłonne operacje. Każde ponowne wykonanie takiej metody, powoduje zużycie zasobów komputera.