ASP.NET Core ile Event Sourcing - 02 Messaging

1. Giriş

Bu yazı aşağıdaki yazının devamı niteliğinde olduğu için aşağıdaki yazıyı okuyup uygulamanızı öneririm. Aşağıdaki yazıdaki örnek proje üzerinden devam edeceğiz.

Bir önceki yazıda Kanban Board örneği yapmıştık. RESTful API oluşturarak create, assign, move ve complete endpoint'lerini yazmıştık. Bu endpoint'lere gelen istekleri birer event olarak Event Store'a kaydetmiştik. Yani Event Store'un store kısmı ile ilgilenmiştik. Bu yazıda ise messaging kısmı ile ilgileneceğiz.

RESTful API endpoint'lerimize aşağıdaki endpoint'i de dahil edeceğiz.

[GET] api/tasks/{id}

2. Couchbase'in Hazırlanması

Event'ler sorgulanabilir yapıda olmadıkları için event'lerin bir bütünü temsil edecek şekilde nosql dökümanı veya sql satırı olarak saklanması gerekiyor. Örneğin; Event Store'dan durumu "Done" olan task'leri alamayız. Bu sebeple event'leri sorgulanabilir bir veritabanında bir bütün halde tutmamız gerekiyor. Bu örneğimizde task'leri couchbase'de tutacağız.

Aşağıdaki komut satırı ile Docker’da Couchbase ayağa kaldırıyoruz.

    docker run -d --name couchbase -p 8091-8094:8091-8094 -p 11210:11210 couchbase

Couchbase ayağa kalktığında aşağıdaki adresten panele girip bir cluster oluşturmalısınız.

http://localhost:8091/

Aşağıdaki curl komut satırları ile Couchbase'de "checkpoints" ve "tasks" isimli iki adet bucket oluşturalım. -u parametresini cluster oluştururken vermiş olduğunuz bilgiler ile değiştirmeyi unutmayın.

    curl -X POST -u [admin]:[password] http://localhost:8091/pools/default/buckets
  -d name=checkpoints -d ramQuotaMB=100 -d authType=none -d bucketType=couchbase

    curl -X POST -u [admin]:[password] http://localhost:8091/pools/default/buckets
  -d name=tasks -d ramQuotaMB=100 -d authType=none -d bucketType=couchbase

3. Couchbase'in API'ye Uyarlanması

Aşağıdaki komut satırları ile "CouchbaseNetClient" ve "Couchbase.Extensions.DependencyInjection" paketlerini API projesine yüklüyoruz.

    dotnet add package CouchbaseNetClient -v 2.7.16
dotnet add package Couchbase.Extensions.DependencyInjection -v 2.0.2

appsettings.json'a Couchbase bağlantı bilgilerini aşağıdaki gibi ekliyoruz.

Startup.cs dosyasında ise Couchbase'i konfigüre ediyoruz.

Event Store'da her event'in bir position değeri vardır. Event Store'dan okunan event'leri Couchbase'e başarılı şekilde kaydettikten sonra son okunan event'in position değerini saklamalıyız. Bu position değerini CheckpointDocument tipinde saklayacağız.

Infrastructure klasörünün içerisine "CheckpointDocument.cs" isminde bir class ekleyelim. Class'ın içerisine aşağıdaki kodu yapıştıralım.

Couchbase'e yazdığımız son position değerini okumak veya position değerini kaydetmek için repository class'ı yazıyoruz.

Infrastructure klasörünün içerisine "CheckpointRepository.cs" isminde bir class ekleyelim. Class'ın içerisine aşağıdaki kodu yapıştıralım.

Task bilgilerini TaskDocument tipinde saklayacağız. Infrastructure klasörünün içerisine "TaskDocument.cs" isminde bir class ekleyelim. Class'ın içerisine aşağıdaki kodu yapıştıralım.

Event Store'dan okunan event'leri Couchbase'e yazmak ve task'leri Couchbase'den sorgulamak için repository class'ı yazıyoruz.

Infrastructure klasörünün içerisine "TaskRepository.cs" isminde bir class ekleyelim. Class'ın içerisine aşağıdaki kodu yapıştıralım.

Save metotunda Event Store'a kaydedilen event'in tipine göre ilgili task'i ekleyip, task'in ilgili bilgilerini güncelliyoruz.
Get metotunda ise Couchbase'den id bilgisine göre task'leri sorguluyoruz.

Startup.cs dosyasında CheckpointRepository ve TaskRepository class'larını DI Container’e aşağıdaki gibi ekleyelim.

4. Consumer'ın Hazırlanması

Event Store'dan sürekli olarak event'leri dinleyeceğimiz consumer'a ihtiyacımız var. Bu sebeple ASP.NET Core'daki Hosted Service'i kullanacağız.

Projeye "HostedServices" isminde klasör ekleyelim ve içerisine "TaskHostedService.cs" isminde bir class ekleyelim. Class'ın içerisine aşağıdaki kodu yapıştıralım.

33. satırda Couchbase'e kaydedilen son event'in position değerini sorguluyoruz.
42. satırda ise Event Store'a abone olup event'leri dinlemeye başlıyoruz. lastCheckpoint değeri verilmez ise projeyi her ayağı kaldırdığımızda tüm event'ler tekrar tekrar okunur. lastCheckpoint ile birlikte son okuduğumuz position değerinden sonraki event'leri okuyacağımızı belirtiyoruz.
47. satırda Event Store'un kendine ait event'leri ile ilgili işlem yapmamak için pas geçiyoruz.
52. satırda okunan event'in tipini buluyoruz. 53. satırda ise event'i tipine göre deserialize ediyoruz.
55. satırda bu event'lerin dışındaki event'ler için işlem yapmamak için pas geçiyoruz.
58. satırda task'i Couchbase'e kaydediyoruz.
60. satırda ise Event Store'dan okunan event'in position değerini Couchbase'e kaydediyoruz.

5. API Endpoint'inin Hazırlanması

TasksController'a task'i sorgulayacağımız endpoint'i aşağıdaki gibi ekliyoruz.

Artık her şey hazır. API projesini çalıştırabiliriz. Bir önceki yazının sonunda eklediğimiz task'i aşağıdaki curl komut satırı ile sorgulayabiliriz.

    curl https://localhost:44361/api/tasks/3a7daba9-872c-4f4d-8d6f-e9700d78c4f5

Projenin son haline Github'dan erişebilirsiniz.

Kolay gelsin.