![\"\"](\"\/wp-content\/uploads\/2020\/02\/scalability.jpg\")
<\/a>Bildi\u011fimiz gibi uygulamalar\u0131m\u0131za bu kabiliyetleri kazand\u0131rabilmek i\u00e7in, bir \u00e7ok teknoloji ve mimarilerden yararlan\u0131yoruz. Microservice yakla\u015f\u0131m\u0131 ile scalable<\/strong>, resilient<\/strong> ve lightweight<\/strong> uygulamalar geli\u015ftiriyor, response s\u00fcrelerini minimize edebilmek i\u00e7in de farkl\u0131 caching teknolojilerinden faydalanmaya \u00e7al\u0131\u015f\u0131yoruz.<\/p>\n Peki, bu kadar mimari boyutta uygulamalar\u0131m\u0131za bir \u015feyler kazand\u0131rabilmek i\u00e7in u\u011fra\u015f\u0131rken, baz\u0131 durumlarda neden kod’lar\u0131m\u0131z\u0131, thread’lerimizi blokluyoruz?<\/p>\n Unutmamal\u0131y\u0131z ki mimari boyutta uygulamalar\u0131m\u0131za bir \u015feyler kazand\u0131rmaya \u00e7al\u0131\u015ft\u0131\u011f\u0131m\u0131z kadar, kod boyutunda yapt\u0131klar\u0131m\u0131z da bi o kadar \u00f6nemlidir.<\/p>\n Baz\u0131 durumlarda procedural programlama paradigmas\u0131 mant\u0131\u011f\u0131ndan farkl\u0131 olarak, yani kod sat\u0131rlar\u0131ndaki ak\u0131\u015flar\u0131n s\u0131ras\u0131yla takip edilmesi yerine, event-based programlama mant\u0131\u011f\u0131nda hareket etmeli ve uygulamalar\u0131m\u0131z\u0131 reactive<\/strong> bir hale getirmeliyiz.<\/p>\n Rx’i, k\u0131saca hat\u0131rlayal\u0131m.<\/p>\n Rx i\u00e7in k\u0131saca, observable stream’leri<\/strong>\u00a0kullanarak push-based, asynchronous ve daha responsive<\/strong> uygulamalar geli\u015ftirebilmemize olanak sa\u011flayan g\u00fc\u00e7l\u00fc bir library’dir diyebiliriz.<\/p>\n Rx, complex business logic’lerini daha basit bir halde ve ayr\u0131ca asynchronous olarak kolay bir \u015fekilde handle edebilmemizi sa\u011flamaktad\u0131r. Ayr\u0131ca Rx, bir \u00e7o\u011fumuzun bildi\u011fi gibi yeni bir konsept olmamakla beraber, temelinde Observer<\/em> design pattern konsept’i bulunmaktad\u0131r.<\/p>\n Kullan\u0131m senaryolar\u0131n\u0131 ise, genel olarak a\u015fa\u011f\u0131daki gibi s\u0131ralayabiliriz.<\/p>\n Bu makale kapsam\u0131nda ise kod’lar\u0131m\u0131z\u0131 ve thread’lerimizi blok’lamadan, asynchronous<\/strong> ve event-based<\/strong> interaction’lar\u0131 uygulamalar\u0131m\u0131z i\u00e7erisinde en basit haliyle nas\u0131l handle edebiliriz senaryosunu cover etmeye \u00e7al\u0131\u015faca\u011f\u0131m.<\/p>\n \u00d6rnek bir uygulamaya ge\u00e7meden \u00f6nce, Rx’in temel ta\u015flar\u0131n\u0131 olu\u015fturan “IObservable<T><\/em>” ve “IObserver<T><\/em>” interface’lerinden bahsetmek istiyorum.<\/p>\n Bu harika ikili, t\u0131pk\u0131 “IEnumerable<T><\/em>” and “IEnumerator<T><\/em>” interface’lerine benzemektedir. Farkl\u0131 olarak pull-based<\/strong> bir yakla\u015f\u0131m yerine, push-based<\/strong> bir yakla\u015f\u0131m ile \u00e7al\u0131\u015fmaktad\u0131r. B\u00f6ylece s\u00fcrekli bir kaynakta data var m\u0131 diye sormak yerine, ilgili kayna\u011fa subscribe olarak event’lere kar\u015f\u0131 daha responsive uygulamalar geli\u015ftirebilmekteyiz.<\/p>\n “IObservable<T><\/em>” interface’ini, g\u00f6zlemlemek istedi\u011fimiz bir kaynak olarak d\u00fc\u015f\u00fcnebiliriz. \u0130\u00e7erisinde ise “Subscribe(IObserver<T> observer)<\/em>” method’unu bulundurmaktad\u0131r. Ad\u0131ndan da anlayabilece\u011fimiz \u00fczere, “IObserver<T><\/em>” ise g\u00f6zlemcimiz.<\/p>\n Hemen basit bir \u00f6rnek ger\u00e7ekle\u015ftirelim.<\/p>\n Yukar\u0131daki console uygulamas\u0131na bakarsak, “IObservable<long><\/em>” tipinde her saniye tetiklenecek bir kaynak olu\u015fturduk. Ard\u0131ndan bu kayna\u011fa subscribe olarak, data var oldu\u011fu s\u00fcrece console ekran\u0131na yazd\u0131r\u0131lmas\u0131n\u0131 sa\u011flad\u0131k.<\/p>\n Console \u00e7\u0131kt\u0131s\u0131 ise, a\u015fa\u011f\u0131daki gibi sonsuz bir \u015fekilde olacakt\u0131r.<\/p>\n Bir g\u00fcn bizden anl\u0131k bir chat uygulamas\u0131 geli\u015ftirmemiz istedi\u011fini varsayal\u0131m. Ayr\u0131ca chat mesajlar\u0131n\u0131n persistent olmas\u0131 ve bu i\u015flemin son kullan\u0131c\u0131 response s\u00fcrelerini etkilememesi de istenmektedir.<\/p>\n Bu kapsamda, anl\u0131k chat i\u015flemleri i\u00e7in SignalR<\/em> library’sinden yararlanal\u0131m.<\/p>\n Peki, \u00f6ncelikle a\u015fa\u011f\u0131daki gibi bir ASP.NET Core Web API<\/em> projesi olu\u015ftural\u0131m. Ard\u0131ndan “SignalR<\/em>” ve “System.Reactive<\/em>” library’lerini NuGet \u00fczerinden projeye dahil edelim.<\/p>\n \u015eimdi “Models<\/em>” isminde bir klas\u00f6r olu\u015ftural\u0131m ve i\u00e7erisinde “ChatMessageReceivedEvent<\/em>” ad\u0131nda bir event model’i tan\u0131mlayal\u0131m.<\/p>\n Bu event’i, socket \u00fczerinden bir chat mesaj\u0131 ald\u0131\u011f\u0131m\u0131zda publish edece\u011fiz. B\u00f6ylece non-blocking, event-based bir chat uygulamas\u0131 geli\u015ftirmeye \u00e7al\u0131\u015faca\u011f\u0131z.<\/p>\n \u015eimdi ise “Handlers<\/em>” isminde bir klas\u00f6r olu\u015ftural\u0131m ve i\u00e7erisinde Rx’i implemente edece\u011fimiz interface’i a\u015fa\u011f\u0131daki gibi tan\u0131mlayal\u0131m.<\/p>\n Tan\u0131mlam\u0131\u015f oldu\u011fumuz “Publish<\/em>” method’u ile, Rx stream’ine bir event g\u00f6nderece\u011fiz. Ard\u0131ndan “Subscribe<\/em>” method’u ile de, Rx stream’ine diledi\u011fimiz bir action’\u0131 ekleyece\u011fiz.<\/p>\n \u015eimdi a\u015fa\u011f\u0131daki gibi “Handlers\/Implementations<\/em>” klas\u00f6r\u00fc alt\u0131nda implementation i\u015flemini ger\u00e7ekle\u015ftirelim.<\/p>\n Dikkat edersek burada, “IObservable<T><\/em>” yerine “Subject<\/em>” class’\u0131n\u0131 kulland\u0131k. \u00c7\u00fcnk\u00fc bu harika class, hem “IObservable<T><\/em>” hem de “IObserver<T><\/em>” \u0131 implemente etmektedir ve bir proxy gibi davranmaktad\u0131r.<\/p>\n “Publish<\/em>” method’unda ise, stream i\u00e7erisinde yeni bir event ger\u00e7ekle\u015fti\u011finde “Subject<\/em>” class’\u0131n\u0131n t\u00fcm subscriber’lar\u0131 bilgilendirece\u011fi bir observable contract’\u0131 olan “OnNext<\/em>” method’unu implemente ettik.<\/p>\n Bu ak\u0131\u015f\u0131 g\u00f6rselle\u015ftirdi\u011fimizde ise, a\u015fa\u011f\u0131daki gibi g\u00f6r\u00fcnecektir.<\/p>\n “Subscribe<\/em>” method’u i\u00e7erisinde ise, Rx stream’ine subscribe olma i\u015flemini ger\u00e7ekle\u015ftiriyoruz. Rx’in en sevdi\u011fim taraf\u0131 ise, Linq operasyonlar\u0131n\u0131 da destekliyor olmas\u0131.<\/p>\n Unsubscribe i\u015flemini ger\u00e7ekle\u015ftirebilmemiz i\u00e7in ise, subscribe olma i\u015flemi geriye bir “IDisposable” <\/em>d\u00f6nmektedir.<\/em> “Dispose<\/em>” method’u i\u00e7erisinde ise, dictionary’e eklemi\u015f oldu\u011fumuz subscriber’lar\u0131n unsubscribe i\u015flemlerini ger\u00e7ekle\u015ftiriyoruz. Bu konu hakk\u0131ndaki detayl\u0131 bilgiye ise, buradan<\/a><\/em> ula\u015fabilirsiniz.<\/p>\n \u015eimdi chat k\u0131sm\u0131na ge\u00e7ebiliriz.<\/p>\n Bunun i\u00e7in, “Hubs<\/em>” isminde bir klas\u00f6r olu\u015fturarak i\u00e7erisinde “ChatHub<\/em>” isminde bir class tan\u0131mlayal\u0131m ve a\u015fa\u011f\u0131daki gibi implemente edelim.<\/p>\n Burada basit olarak “SignalR<\/em>” library’sinin “Hub<\/em>” class’\u0131n\u0131 inherit alarak, chat i\u00e7in socket \u00fczerinden kullanaca\u011f\u0131m\u0131z “SendMessage<\/em>” method’unu tan\u0131mlad\u0131k.<\/p>\n Ayr\u0131ca son kullan\u0131c\u0131n\u0131n g\u00f6nderdi\u011fi mesajlar\u0131, son kullan\u0131c\u0131n\u0131n response s\u00fcresini etkilemeden persistent bir hale getirebilmek i\u00e7in, “IChatEventHandler<\/em>” arac\u0131l\u0131\u011f\u0131yla bir observable stream haline \u00e7eviriyoruz.<\/p>\n Her bir mesaj publish edildi\u011finde ise, Rx stream’i i\u00e7erisindeki “OnNext<\/em>” method’u \u00e7a\u011f\u0131r\u0131lacak ve ilgili t\u00fcm subscriber’lar bu event’ten haberdar edilecektir.<\/p>\n \u015eimdi chat mesajlar\u0131n\u0131 persist edebilmek i\u00e7in, Rx stream’ine “ChatHistoryConsumer<\/em>” ad\u0131nda “Handlers\/Implementations<\/em>” path’i alt\u0131na a\u015fa\u011f\u0131daki gibi bir observer subscribe edelim.<\/p>\n Burada ise “ChatHistoryConsumer<\/em>” class’\u0131n\u0131n bir background task’\u0131 olarak \u00e7al\u0131\u015fabilmesi i\u00e7in, “BackgroundService<\/em>” class’\u0131n\u0131 inherit ederek “ExecuteAsync<\/em>” method’unu implemente ettik.<\/p>\n Ard\u0131ndan “ExecuteAsync<\/em>” method’u i\u00e7erisinde ise, “ChatMessageReceivedEvent<\/em>” i\u00e7in subscription i\u015flemini ger\u00e7ekle\u015ftirdik.<\/p>\n \u015eimdi “Startup<\/em>” class’\u0131n\u0131 ise, a\u015fa\u011f\u0131daki gibi g\u00fcncelleyelim.<\/p>\n Burada ise chat uygulamalas\u0131n\u0131 test edebilmek i\u00e7in, Razor Pages<\/em>‘i ve SignalR<\/em>‘\u0131 etkinle\u015ftirdik. Ard\u0131ndan gerekli injection i\u015flemlerini ger\u00e7ekle\u015ftirdik.<\/p>\n \u015eimdi test i\u015flemini yapabilmek i\u00e7in, basit bir UI<\/em> haz\u0131rlayal\u0131m.<\/p>\n \u00d6ncelikle gerekli SignalR<\/em> javascript paketlerini projeye dahil edebilmek i\u00e7in, buradaki<\/a><\/em> ilk ad\u0131m\u0131 takip edelim.<\/p>\n Ard\u0131ndan “Pages<\/em>” isminde bir klas\u00f6r olu\u015fturarak, i\u00e7erisinde “Chat<\/em>” ad\u0131nda a\u015fa\u011f\u0131daki gibi bir Razor Page<\/em> olu\u015ftural\u0131m.<\/p>\n SignalR<\/em> javascript client’\u0131n\u0131 ise, “chat.js<\/em>” ad\u0131yla “wwwroot\/lib\/signalr<\/em>” klas\u00f6r path’i alt\u0131nda a\u015fa\u011f\u0131daki gibi olu\u015ftural\u0131m.<\/p>\n Burada ise olu\u015fturmu\u015f oldu\u011fumuz “ChatHub<\/em>” a socket \u00fczerinden ba\u011flanacak ve mesaj g\u00f6nderilebilecek basit bir UI<\/em> haz\u0131rlad\u0131k.<\/p>\n \u015eimdi test i\u015flemine ge\u00e7ebiliriz.<\/p>\n Test edebilmek i\u00e7in uygulamay\u0131 “dotnet run<\/em>” komutu ile \u00e7al\u0131\u015ft\u0131ral\u0131m ve iki farkl\u0131 browser \u00fczerinden a\u015fa\u011f\u0131daki gibi chat i\u015flemini ger\u00e7ekle\u015ftirelim.<\/p>\nEn b\u00fcy\u00fck yard\u0131mc\u0131m\u0131z, Reactive Extensions (Rx)!<\/h2>\n
<\/a>\n
Observers<\/h2>\n
using System;\r\nusing System.Reactive.Linq;\r\n\r\nnamespace ReactiveNumbers\r\n{\r\n class Program\r\n {\r\n static void Main(string[] args)\r\n {\r\n IObservable<long> numbers = Observable.Interval(TimeSpan.FromSeconds(1));\r\n\r\n numbers.Subscribe(num => \r\n {\r\n Console.WriteLine(num);\r\n });\r\n\r\n Console.ReadKey();\r\n }\r\n }\r\n}<\/pre>\n% dotnet run\r\n0\r\n1\r\n2\r\n3\r\n4\r\n5\r\n6\r\n7\r\n8\r\n9\r\n...\r\n\r\n<\/pre>\n
Subject<\/h2>\n
dotnet new webapi -n MyChat<\/pre>\n
dotnet add package SignalR\r\ndotnet add package System.Reactive<\/pre>\n
namespace MyChat.Models\r\n{\r\n public class ChatMessageReceivedEvent\r\n {\r\n public string Message { get; set; }\r\n }\r\n}<\/pre>\nusing System;\r\nusing MyChat.Models;\r\n\r\nnamespace MyChat.Handlers\r\n{\r\n public interface IChatEventHandler\r\n {\r\n void Publish(ChatMessageReceivedEvent eventMessage);\r\n void Subscribe(string subscriberName, Action<ChatMessageReceivedEvent> action);\r\n void Subscribe(string subscriberName, Func<ChatMessageReceivedEvent, bool> predicate, Action<ChatMessageReceivedEvent> action);\r\n }\r\n}<\/pre>\nusing System;\r\nusing System.Collections.Generic;\r\nusing System.Reactive.Linq;\r\nusing System.Reactive.Subjects;\r\nusing MyChat.Models;\r\n\r\nnamespace MyChat.Handlers.Implementations\r\n{\r\n public class ChatEventHandler : IChatEventHandler, IDisposable\r\n {\r\n private readonly Subject<ChatMessageReceivedEvent> _subject;\r\n private readonly Dictionary<string, IDisposable> _subscribers;\r\n\r\n public ChatEventHandler()\r\n {\r\n _subject = new Subject<ChatMessageReceivedEvent>();\r\n _subscribers = new Dictionary<string, IDisposable>();\r\n }\r\n\r\n public void Publish(ChatMessageReceivedEvent eventMessage)\r\n {\r\n _subject.OnNext(eventMessage);\r\n }\r\n\r\n public void Subscribe(string subscriberName, Action<ChatMessageReceivedEvent> action)\r\n {\r\n if (!_subscribers.ContainsKey(subscriberName))\r\n {\r\n _subscribers.Add(subscriberName, _subject.Subscribe(action));\r\n }\r\n }\r\n\r\n public void Subscribe(string subscriberName, Func<ChatMessageReceivedEvent, bool> predicate, Action<ChatMessageReceivedEvent> action)\r\n {\r\n if (!_subscribers.ContainsKey(subscriberName))\r\n {\r\n _subscribers.Add(subscriberName, _subject.Where(predicate).Subscribe(action));\r\n }\r\n }\r\n\r\n public void Dispose()\r\n {\r\n if (_subject != null)\r\n {\r\n _subject.Dispose();\r\n }\r\n\r\n foreach (var subscriber in _subscribers)\r\n {\r\n subscriber.Value.Dispose();\r\n }\r\n }\r\n }\r\n}<\/pre>\n<\/a><\/p>\nusing System.Threading.Tasks;\r\nusing Microsoft.AspNetCore.SignalR;\r\nusing MyChat.Handlers;\r\nusing MyChat.Models;\r\n\r\nnamespace MyChat.Hubs\r\n{\r\n public class ChatHub : Hub\r\n {\r\n private readonly IChatEventHandler _chatEventHandler;\r\n\r\n public ChatHub(IChatEventHandler chatEventHandler)\r\n {\r\n _chatEventHandler = chatEventHandler;\r\n }\r\n\r\n public async Task SendMessage(string sender, string message)\r\n {\r\n await Clients.All.SendAsync(\"chat\", sender, message);\r\n\r\n _chatEventHandler.Publish(new ChatMessageReceivedEvent\r\n {\r\n Message = message\r\n });\r\n }\r\n }\r\n}<\/pre>\nusing System.Threading;\r\nusing System.Threading.Tasks;\r\nusing Microsoft.Extensions.Hosting;\r\nusing MyChat.Models;\r\n\r\nnamespace MyChat.Handlers.Implementations\r\n{\r\n public class ChatHistoryConsumer : BackgroundService\r\n {\r\n private readonly IChatEventHandler _eventHandler;\r\n\r\n public ChatHistoryConsumer(IChatEventHandler eventHandler)\r\n {\r\n _eventHandler = eventHandler;\r\n }\r\n\r\n protected override Task ExecuteAsync(CancellationToken stoppingToken)\r\n {\r\n _eventHandler.Subscribe(subscriberName: typeof(ChatHistoryConsumer).Name,\r\n action: async (e) =>\r\n {\r\n if (e is ChatMessageReceivedEvent)\r\n {\r\n await PersistChatMessagesToDBAsync((ChatMessageReceivedEvent)e);\r\n }\r\n });\r\n\r\n return Task.CompletedTask;\r\n }\r\n\r\n private async Task PersistChatMessagesToDBAsync(ChatMessageReceivedEvent e)\r\n {\r\n await System.Console.Out.WriteLineAsync($\"Chat message received and persisted: {e.Message}\");\r\n }\r\n }\r\n}<\/pre>\nusing Microsoft.AspNetCore.Builder;\r\nusing Microsoft.AspNetCore.Hosting;\r\nusing Microsoft.Extensions.Configuration;\r\nusing Microsoft.Extensions.DependencyInjection;\r\nusing Microsoft.Extensions.Hosting;\r\nusing MyChat.Handlers;\r\nusing MyChat.Handlers.Implementations;\r\nusing MyChat.Hubs;\r\n\r\nnamespace MyChat\r\n{\r\n public class Startup\r\n {\r\n public Startup(IConfiguration configuration)\r\n {\r\n Configuration = configuration;\r\n }\r\n\r\n public IConfiguration Configuration { get; }\r\n\r\n \/\/ This method gets called by the runtime. Use this method to add services to the container.\r\n public void ConfigureServices(IServiceCollection services)\r\n {\r\n services.AddRazorPages();\r\n services.AddControllers();\r\n services.AddSignalR();\r\n services.AddSingleton<IChatEventHandler, ChatEventHandler>();\r\n services.AddHostedService<ChatHistoryConsumer>();\r\n }\r\n\r\n \/\/ This method gets called by the runtime. Use this method to configure the HTTP request pipeline.\r\n public void Configure(IApplicationBuilder app, IWebHostEnvironment env)\r\n {\r\n if (env.IsDevelopment())\r\n {\r\n app.UseDeveloperExceptionPage();\r\n }\r\n\r\n app.UseHttpsRedirection();\r\n app.UseStaticFiles();\r\n\r\n app.UseRouting();\r\n\r\n app.UseAuthorization();\r\n\r\n app.UseEndpoints(endpoints =>\r\n {\r\n endpoints.MapRazorPages();\r\n endpoints.MapHub<ChatHub>(\"\/hubs\/chat\");\r\n });\r\n }\r\n }\r\n}<\/pre>\n@page\r\n\r\n<style>\r\ninput[type=text] {\r\n width: 100%;\r\n padding: 12px 20px;\r\n margin: 8px 0;\r\n box-sizing: border-box;\r\n}\r\n<\/style>\r\n\r\n<div>\r\n <h2>Chat Test<\/h2>\r\n <div>\r\n <ul id=\"messageList\"><\/ul>\r\n <\/div>\r\n<\/div>\r\n<div>\r\n <div>Sender: <input type=\"text\" id=\"sender\"\/><\/div>\r\n <div>Message: <input type=\"text\" id=\"message\"\"><\/div>\r\n <div><input type=\"button\" id=\"sendMessage\" value=\"Send\" \/><\/div>\r\n<\/div>\r\n\r\n<script src=\"\/lib\/signalr\/signalr.js\"><\/script>\r\n<script src=\"\/lib\/signalr\/chat.js\"><\/script><\/pre>\nconst connection = new signalR.HubConnectionBuilder()\r\n .withUrl(\"https:\/\/localhost:5001\/hubs\/chat\")\r\n .build();\r\n\r\ndocument.getElementById(\"sendMessage\").addEventListener(\"click\", event => {\r\n const message = document.getElementById(\"message\").value;\r\n const sender = document.getElementById(\"sender\").value;\r\n\r\n connection.invoke(\"SendMessage\", sender, message).catch(err => console.error(err.toString()));\r\n event.preventDefault();\r\n});\r\n\r\nconnection.on(\"chat\", (sender, message) => {\r\n const recMessage = sender + \": \" + message;\r\n const li = document.createElement(\"li\");\r\n\r\n li.textContent = recMessage;\r\n document.getElementById(\"messageList\").appendChild(li);\r\n});\r\n\r\nconnection.start().catch(err => console.error(err.toString()));<\/pre>\n