EditArticle

Edit Article XML Content

<document> <header> <issuecode /> <articlecode /> <zone /> <title>Implementing the Outbox Pattern with Kafka and C#</title> <authors /> <copyright>CODE Magazine</copyright> <owner>CODE Magazine</owner> </header> <body> The outbox pattern is a proven and scalable software design pattern often used in a distributed environment to publish events reliably and enforce data consistency. This article presents an overview of the outbox pattern and examines how it can be implemented using Kafka and C# in ASP.NET Core applications. If you’re to work with the code examples discussed in this article, you need the following installed in your system: <list type="bulleted"> <bulletedlist>Visual Studio 2022 </bulletedlist> <bulletedlist>.NET 8.0</bulletedlist> <bulletedlist>ASP.NET Core 8.0 Runtime</bulletedlist> </list> If you don’t already have Visual Studio 2022 installed on your computer, you can download it from here: https://visualstudio.microsoft.com/downloads/. In this article, I'll examine the following points: <list type="bulleted"> <bulletedlist>An overview of the outbox pattern and its benefits and downsides</bulletedlist> <bulletedlist>An introduction to Kafka as a message broker</bulletedlist> <bulletedlist>How the transaction outbox pattern works</bulletedlist> <bulletedlist>How to implement the outbox pattern in ASP.NET Core, Kafka, and C#</bulletedlist> <bulletedlist>How to implement the consumer application to consume messages stored in Kafka</bulletedlist> </list> <h2>A Real-World Use Case</h2> Typically, microservices-based applications maintain their own data store to persist and retrieve data. This strategy helps improve the application’s scalability and promotes agility in the deployment process. Additionally, these applications take advantage of messaging to facilitate asynchronous communication between the services. There are several messaging solutions for you to choose from such as Kafka, Azure Service Bus, RabbitMQ, etc. <h3>Understanding the Problem</h3> Consider the flow of a typical Order Processing application at a glance: <list type="numbered"> <numberedlist>The client application sends a request to the API to persist the data to the underlying database.</numberedlist> <numberedlist>The API validates the incoming request, executes the necessary business logic, and persists the data in the database, as shown in Figure 1.</numberedlist> </list> <figure id="1" src="image1.png"> Figure 1 : The flow of a typical Order Processing application </figure> <list type="numbered"> <numberedlist>Once the data is stored in the database, the API calls a message broker to publish the event so that other services can consume it if needed, as shown in Figure 2.</numberedlist> </list> <figure id="2" src="image2.png"> Figure 2 : The API calls the message broker to publish the event. </figure> What’s wrong with this? Although the flow may appear straightforward, complications can arise. For instance, if the message broker is unavailable, there’s an error in publishing the messages or a runtime exception occurs in the application. As shown in Figure 3, the ordering service and the data can become inconsistent. <figure id="3" src="image3.png"> Figure 3 : An e rror in publishing messages</figure> Addressing this issue can be challenging, as writing code to solve it may lead to a violation of the Single Responsibility Principle, potentially requiring additional responsibilities for the OrderService. Here’s where the transactional outbox pattern comes to the rescue. <h2>An Overview of the Outbox Pattern</h2> The outbox pattern is a technique used in microservices architectures to guarantee reliable communication between services. For this purpose, a temporary storage area is created in the database of a microservice, often referred to as an outbox table, in which the outgoing data changes are recorded. The data is serialized before being sent from the outbox table to other external systems or services. In distributed systems, duplicate writes occur when an operation must update a local database and notify other services. The outbox design solves this problem by combining the publication of messages/events with the same database transaction that updates the local state—making both actions atomic and, therefore, complete or incomplete. In the outbox pattern, you won’t be publishing an event to the message broker directly. Instead, a record will be inserted into a special database named Outbox. Incidentally, the Outbox database table is stored in the same database where the data of the application is being stored. Then an asynchronous process executes in the background and periodically checks for incoming messages. When it finds one, it publishes the message to a message broker (i.e., Apache Kafka, RabbitMQ, etc.), as shown in Figure 4. <figure id="4" src="image4.png"> Figure 4 : Demonstrating the Outbox Pattern</figure> <h2>Components of the Outbox Pattern</h2> The outbox pattern is comprised of the following key components: <list type="bulleted"> <bulletedlist> Events: At runtime, an application generates events that are stored in the Outbox table.</bulletedlist> <bulletedlist> Outbox table: This data store is used to store the messages until they are processed by the publisher.</bulletedlist> <bulletedlist> Publisher: This component is used to process messages from the outbox table and then dispatch them to the appropriate destinations</bulletedlist> </list> <h2>Benefits and Downsides</h2> The outbox pattern is a proven technique widely used for enhancing data consistency and reliability in microservices architectures. However, like any architectural pattern, it has its benefits and downsides. <h3>Benefits</h3> The following are the benefits of the outbox pattern: <list type="bulleted"> <bulletedlist> Transactional c onsistency: The outbox pattern ensures that any changes to the database residing locally and the publishing of events/messages are part of the same database transaction. This prevents data inconsistencies that might occur due to operation failures after the database changes are committed but before the events are published. By sending the messages as part of the same transaction, the outbox pattern maintains atomicity and consistency. Moreover, the outbox pattern enables events to be retried until they have been successfully processed.</bulletedlist> <bulletedlist> Reliable m essage p ublishing: The outbox pattern ensures reliable message sending by taking advantage of a persistent storage mechanism where the messages or events are stored in the database, thereby preventing message loss due to failures in the event publishing mechanism or network issues. This persistent storage of messages coupled with asynchronous message sending helps decouple the message dispatch process from the application's response, thereby minimizing the chances of message loss and improving resiliency.</bulletedlist> <bulletedlist> Scalability: The outbox pattern enhances scalability of an application by separating the concerns of event creation from event distribution. It enables an application to focus on processing incoming requests without interruption by offloading message transmission to a different process or service.</bulletedlist> <bulletedlist> Performance: The outbox pattern improves performance by sending messages asynchronously, thereby processing the messages without any delays to enhance responsiveness and throughput. Storing the events in the same database transaction as the business operations minimizes the number of total transactions executed, which can enhance performance.</bulletedlist> </list> <h3>Downsides</h3> Despite the benefits of the outbox pattern stated earlier, there are certain downsides as well: <list type="bulleted"> <bulletedlist> Increased c omplexity: The outbox pattern can introduce additional complexity to your architecture, application design, and development. By implementing this pattern, you need to manage the outbox processor and verify whether the messages are recorded correctly, thereby making your application difficult to manage, maintain, and debug.</bulletedlist> <bulletedlist> Latency: When using the outbox pattern, there might be a delay from the time an event occurs and when it’s published to other services. It usually publishes events in batches at a regular time interval set by the outbox processor. However, this delay isn’t recommended in situations where an immediate or real-time communication is needed.</bulletedlist> <bulletedlist> Database c oupling: Because the outbox pattern depends on the database to store events before they’re published, there's an inherent coupling with the database, thereby making future changes more challenging.</bulletedlist> <bulletedlist> Scalability: Although the outbox pattern ensures that microservices are not tightly coupled and can be independently scaled, scaling your application may still be challenging. If an outbox processor isn’t working as expected or there’s too much traffic, you might have to consider scaling your infrastructure for processing outboxes.</bulletedlist> <bulletedlist> Resiliency: Although the outbox pattern ensures that events aren’t lost, there’s no built-in mechanism to handle failures while processing the events. You need to write your code to handle duplicate events, idempotency, retries, and failure recovery.</bulletedlist> <bulletedlist> O perational overhead: Monitoring the outbox repository regularly and adopting the strategy to transmit messages and potential retries can be an additional overhead. Keeping an eye on the outbox storage, ensuring that the messages are delivered, and dealing with errors or retries may be challenging.</bulletedlist> <bulletedlist> Integration challenges: You must take the necessary steps to ensure reliable message delivery. This may involve implementing additional configurations and dependencies when connecting to external systems like RabbitMQ and Apache Kafka.</bulletedlist> </list> <h2>How Does the Outbox Pattern Work?</h2> Assume an order processing application in which an order service is used to handle orders placed by the customer. As soon as a new order is placed, the application must update the order database table and also notify the inventory service that the stock should be updated. The complete flow is illustrated in the sequence of steps of a typical order processing use case given below: <list type="numbered"> <numberedlist> Create O rder: A customer uses the application to place a new order. Next, the Order Service picks up this order and processes it.</numberedlist> <numberedlist> Update O rder d atabase t able: The Order Service stores the new order in the Order database table with information that includes the order details and the customer details together with the Order status.</numberedlist> <numberedlist> Create Stock Update m essage: Besides updating the Order database table, the Order Service creates a message to indicate that the Stock database table needs to be updated. This message is persisted in the Outbox database table within the context of the same transaction.</numberedlist> <numberedlist> Commit t ransaction: Next, the transaction that comprises updating the Order database table and insertion of the message into the Outbox database table is committed.</numberedlist> <numberedlist> Check Outbox d atabase t able p eriodically: A process named Message Relay executes in the background to check the Outbox database table for new messages.</numberedlist> <numberedlist> Dispatch m essage to Stock s ervice: When the Message Relay detects a new message in the Outbox database table, it forwards the message to the Stock service for processing via the Message queue.</numberedlist> <numberedlist> Stock u pdate: Upon receipt of a new message, the Stock service updates the Stock database table accordingly.</numberedlist> </list> In the outbox pattern, when an event occurs, instead of a message being sent to a message broker such as Kafka, a new record that contains the details of the event is stored in the Outbox database table. It should be noted that typically the Outbox table is part of the same database and this all happens as part of the same transaction. Figure 5 illustrates how this all works together. <figure id="5" src="image5.png"> Figure 5 : The Outbox Pattern in action </figure> <h2>Key Terminologies</h2> Here are the key terminologies you often come across when exploring the Outbox pattern: <list type="bulleted"> <bulletedlist> Event l og or the O utbox: Each microservice maintains an event log or an Outbox where the events that have to be published are first recorded to ensure that the events are never lost. Essentially, an Outbox is a database table to store events or messages before the events are published. </bulletedlist> <bulletedlist> Transactional o peration: For the purpose of ensuring that only successful business operations are captured in the event log, events are added to the log as part of the same transaction as the business operations that generated them. </bulletedlist> <bulletedlist> Outbox p rocessor: An outbox processor runs as a separate background task and scans the event log at regular intervals of time. It collects events without processing them and then publishes them to the event bus or message broker. </bulletedlist> <bulletedlist> Message b roker or e vent b us: A message broker or an event bus receives those events that have been published. The message broker or event bus provides durability and reliability and distributes these events to the subscribers or the other microservices.</bulletedlist> </list> <h2>Introduction to Kafka as a Message Broker</h2> Apache Kafka is a distributed, high-performance, open source, scalable, and versatile stream-processing software. It’s written in Java and Scala, and has become popular in recent times for building systems that are adept at handling massive volumes of data. Kafka is a messaging platform based on the publish/subscribe model. It comes with built-in features for replication, partitioning, fault tolerance, and improved throughput, compared to applications without message brokers. Kafka is a good option for applications that need vast data processing capabilities. The main use of Kafka is in creating real-time streaming data pipelines. For this reason, Kafka includes stream processing and fault-tolerant storage functionalities, which enable storing and analyzing historical and real-time data. Figure 6 shows how Kafka fits in a typical implementation of the outbox pattern. <figure id="6" src="image6.png"> Figure 6 : Using Kafka as the message broker in a typical Outbox pattern implementation</figure> <h2>Getting Started with Apache Kafka</h2> To get started with Apache Kafka, follow the steps outlined in the next few sections. <h3>Download and Extract Kafka</h3> You can download Apache Kafka from here: https://kafka.apache.org/downloads. Alternatively, you can run the following command at the terminal to download Kafka in your computer: <codesnippet>wget»https://archive.apache.org/</codesnippet> <codesnippet>dist/kafka/3.7.0/</codesnippet> <codesnippet>kafka_2.13-3.7.0.tgz</codesnippet> Once Kafka has been downloaded, you should extract the zip archive by running the following command at the terminal window: <codesnippet>tar»-xzf»kafka_2.13-3.7.0.tgz</codesnippet> <h3>Start ZooKeeper</h3> You need ZooKeeper to run Kafka on your computer. Once you've downloaded and extracted Apache Kafka to your computer, start ZooKeeper by running the following command at the terminal window: <codesnippet>cd»kafka_2.13-3.7.0</codesnippet> <codesnippet>bin/windows/</codesnippet> <codesnippet>zookeeper-server-start.bat»</codesnippet> <codesnippet>config/zookeeper.properties</codesnippet> <h3>Create a Topic</h3> Now that ZooKeeper and Kafka are up and running in your computer, you can start creating topics. You can create a new topic by running the following command at the terminal window. <codesnippet>.\bin\windows\</codesnippet> <codesnippet>kafka-topics.bat»--create»</codesnippet> <codesnippet>--topic»mynewtopic»</codesnippet> <codesnippet>--bootstrap-server»</codesnippet> <codesnippet>localhost:9092</codesnippet> <h3>Display All Kafka Messages in a Topic</h3> To display all messages in a particular topic, use the following command: <codesnippet>».\bin\windows\</codesnippet> <codesnippet>kafka-console-consumer.bat»</codesnippet> <codesnippet>--bootstrap-server»</codesnippet> <codesnippet>localhost:9092»--topic»</codesnippet> <codesnippet>mynewtopic»--from-beginning</codesnippet> <h3>Display all Topics</h3> To list all topics, you can use the following command: <codesnippet>.\bin\windows\kafka-topics.bat»</codesnippet> <codesnippet>--list»--bootstrap-server»</codesnippet> <codesnippet>localhost:9092</codesnippet> <h3>Start the Producer</h3> You can run the following command at the terminal window to start a producer. <codesnippet>bin/windows/</codesnippet> <codesnippet>kafka-console-producer.bat»</codesnippet> <codesnippet>--topic»mynewtopic»</codesnippet> <codesnippet>--bootstrap-server»</codesnippet> <codesnippet>localhost:9092</codesnippet> <h3>Start the Consumer</h3> In Kafka, you need a consumer to read the messages produced by a producer. You can run the following command at the terminal window to read all messages sent by the producer as shown in Figure 7. <codesnippet>bin/windows/</codesnippet> <codesnippet>kafka-console-consumer.bat»</codesnippet> <codesnippet>--topic»mynewtopic»</codesnippet> <codesnippet>--from-beginning»</codesnippet> <codesnippet>--bootstrap-server»</codesnippet> <codesnippet>localhost:9092</codesnippet> <figure id="7" src="image7.png"> Figure 7 : The Kafka consumer in action</figure> <h3>Shut Down Zookeeper and Kafka</h3> To shut down Zookeeper, use the zookeeper-server-stop.bat script, as shown below: <codesnippet>bin\windows\</codesnippet> <codesnippet>zookeeper-server-stop.bat</codesnippet> To shut down Kafka, you use the kafka-server-stop.bat script as shown below: <codesnippet> bin\windows\ </codesnippet> <codesnippet> kafka-server-stop.bat </codesnippet> <h2>Implementing the Outbox Pattern in ASP.NET Core, Kafka, and C#</h2> In this section, I’ll examine how to implement the outbox pattern in an ASP.NET Core application using Kafka. In this example, you’ll use the following interfaces and classes. <list type="bulleted"> <bulletedlist> OutboxMessage : This is the model class</bulletedlist> <bulletedlist> IKafkaProducer : This is the interface for the KafkaProducer class.</bulletedlist> <bulletedlist> IKafkaConsumer : This is the interface for the KafkaConsumer class.</bulletedlist> <bulletedlist> IOrderService: This interface contains the declaration of methods supported by the OrderService.</bulletedlist> <bulletedlist> IOutboxMessageRepository : This represents the interface for the OutboxMessageRepository class.</bulletedlist> <bulletedlist> KafkaProducer : This class is used to send messages to Kafka. </bulletedlist> <bulletedlist> KafkaConsumer : This class is used to consume messages in Kafka.</bulletedlist> <bulletedlist> OrderService : The OrderService class implements the IOrderService interface and encapsulates the logic for invoking the appropriate methods of the OrderMessageRepository.</bulletedlist> <bulletedlist> KafkaMessageProcessor : This is a background service that calls KafkaConsumer at regular intervals of time to consume messages residing in Kafka.</bulletedlist> <bulletedlist> OutboxMessageProcessor : This is yet another background service used to invoke KafkaProducer to transmit messages to Kafka.</bulletedlist> <bulletedlist> OutboxMessageRepository: This class contains methods to retrieve messages from the Outbox database table and also update a message as needed.</bulletedlist> <bulletedlist> ApplicationDbContext : This class represents the data context that acts as a gateway to connect to the underlying database being used by the application. </bulletedlist> <bulletedlist> OrderController : This represents the Order API that can be consumed by clients to retrieve existing orders or create new orders.</bulletedlist> </list> <h3>Create a New ASP.NET Core 8 Project in Visual Studio 2022</h3> You can create a project in Visual Studio 2022 in several ways such as, from the Visual Studio 2022 Developer Command Prompt or by launching the Visual Studio 2022 IDE. When you launch Visual Studio 2022, you'll see the Start window. You can choose "Continue without code" to launch the main screen of the Visual Studio 2022 IDE. Now that you know the basics, let’s start setting up the project. To create a new ASP.NET Core 8 Project in Visual Studio 2022: <list type="numbered"> <numberedlist>Start the Visual Studio 2022 IDE.</numberedlist> <numberedlist>In the "Create a new project" window, select "ASP.NET Core Web API" and click Next to move on.</numberedlist> <numberedlist>Specify the project name as Outbox_Pattern_Demo and the path where it should be created in the "Configure your new project" window.</numberedlist> <numberedlist>If you want the solution file and project to be created in the same directory, you can optionally check the "Place solution and project in the same directory" checkbox. Click Next to move on.</numberedlist> <numberedlist>In the next screen, specify the target framework and authentication type as well. Ensure that the "Configure for HTTPS," "Enable Docker Support," "Do not use top-level statements", and the "Enable OpenAPI support" checkboxes are unchecked because you won’t use any of these in this example.</numberedlist> <numberedlist>Remember to leave the Use controllers checkbox checked because you won’t use minimal API in this example.</numberedlist> <numberedlist>Click Create to complete the process.</numberedlist> </list> A new ASP.NET Core Web API project is created. You’ll use this project to implement the outbox pattern using Kafka in ASP.NET Core and C#. <h3>Create the Outbox Database Table</h3> First off, create the Outbox database table to store messages. To do this, create a new database called Outbox_Pattern_Demo using the database script given below: <codesnippet>CREATE»DATABASE»Outbox_Pattern_Demo;</codesnippet> Next, create a new database table called OutboxMessage using the following database script. <codesnippet> CREATE»TABLE»Outbox_Message</codesnippet> <codesnippet>(</codesnippet> <codesnippet>»»»Event_Id»BIGINT»IDENTITY»PRIMARY»KEY,</codesnippet> <codesnippet>»»»Event_Payload»NVARCHAR(MAX)»NOT»NULL,</codesnippet> <codesnippet>»»»Event_Date»DATETIME»NOT»NULL»»»</codesnippet> <codesnippet>»»»DEFAULT»CURRENT_TIMESTAMP,</codesnippet> <codesnippet>»»»IsMessageDispatched»BIT»NOT»NULL</codesnippet> <codesnippet>);</codesnippet> Create another database table in the same database named Order using the following script. <codesnippet>CREATE»TABLE»[Order]</codesnippet> <codesnippet>(</codesnippet> <codesnippet>»»»»Order_Id»BIGINT»IDENTITY»PRIMARY»KEY,</codesnippet> <codesnippet>»»»»Customer_Id»INT»NOT»NULL,</codesnippet> <codesnippet>»»»»Order_Date»DATETIME,</codesnippet> <codesnippet>»»»»Order_Amount»MONEY»NOT»NULL</codesnippet> <codesnippet>);</codesnippet> <h3>Install NuGet Package(s)</h3> In this example, you’ll take advantage of Entity Framework to interact with the database and perform CRUD (an acronym for Create Read Update Delete) operations. You’ll also need a Kafka client package to produce and consume messages. To work with Apache Kafka, you’ll use the Confluent.Kafka NuGet package. To install the required package into your project, right-click on the solution and then select Manage NuGet Packages for Solution…. Now search for the Microsoft.EntityFramework.SqlServer and Confluent.Kafka packages in the search box and install it. Alternatively, you can type the commands shown below at the NuGet Package Manager Command Prompt: <codesnippet>PM>»Install-Package</codesnippet> <codesnippet>Microsoft.EntityFrameworkCore.SqlServer</codesnippet> <codesnippet>Install-Package»Confluent.Kafka</codesnippet> You can also install this package by executing the following commands at the Windows Shell: <codesnippet>dotnet»add»package</codesnippet> <codesnippet>Microsoft.EntityFrameworkCore.SqlServer</codesnippet> <codesnippet>dotnet»add»package»Confluent.Kafka</codesnippet> <h3>Create the OutboxMessage Class</h3> <codesnippet> public»class»OutboxMessage</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»public»int»Event_Id»{»get;»set;»}</codesnippet> <codesnippet>»»public»string»Event_Payload»{»get;»set;»}</codesnippet> <codesnippet>»»public»DateTime»Event_Date»{»get;»set;»}</codesnippet> <codesnippet>»»public»bool»IsMessageDispatched»{»get;»set;»}</codesnippet> <codesnippet>}</codesnippet> <h3>Create the Order Model Class</h3> <codesnippet>public»class»Order</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»public»long»Order_Id»{»get;»set;»}</codesnippet> <codesnippet>»»»public»int»Customer_Id»{»get;»set;»}</codesnippet> <codesnippet>»»»public»DateTime»Order_Date»{»get;»set;»}</codesnippet> <codesnippet>»»»public»decimal»Amount»{»get;»set;»}</codesnippet> <codesnippet>}</codesnippet> <codesnippet> public»class»ApplicationDbContext»:»</codesnippet> <codesnippet>»»»»»»»»»»»»»DbContext</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»public»ApplicationDbContext</codesnippet> <codesnippet>»»»(DbContextOptions</codesnippet> <codesnippet>»»»<ApplicationDbContext>»</codesnippet> <codesnippet>»»»»options)»:»base(options)</codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»}</codesnippet> <codesnippet>»»»»protected»override»void»</codesnippet> <codesnippet>»»»»OnModelCreating</codesnippet> <codesnippet>»»»»(ModelBuilder»modelBuilder)</codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»»»»»base.OnModelCreating(modelBuilder);</codesnippet> <codesnippet>»»»»}</codesnippet> <codesnippet>»»»»public»DbSet<OutboxMessage>»</codesnippet> <codesnippet>»»»»OutboxEvents»{»get;»set;»}</codesnippet> <codesnippet>»»»»public»DbSet<Order>»</codesnippet> <codesnippet>»»»»Orders»{»get;»set;»}</codesnippet> <codesnippet>}</codesnippet> <codesnippet>protected»override»void»</codesnippet> <codesnippet> OnModelCreating </codesnippet> <codesnippet>(ModelBuilder»modelBuilder)</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»modelBuilder.Entity<Order>(entity»=></codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»»»»»entity.ToTable("Order");</codesnippet> <codesnippet>»»»»»»»»entity.HasKey(e»=>»e.Order_Id);</codesnippet> <codesnippet> »»»»}); </codesnippet> <codesnippet> »»»»modelBuilder.Entity< OutboxMessage > </codesnippet> <codesnippet> »»»»(entity»=> </codesnippet> <codesnippet> »»»»{ </codesnippet> <codesnippet> »»»»»»»»entity. ToTable ( "OutboxMessage" ); </codesnippet> <codesnippet> »»»»»»»»entity. HasKey (e»=>»e.Event_Id); </codesnippet> <codesnippet> »»»»}); </codesnippet> <codesnippet> »»»»base.OnModelCreating(modelBuilder);</codesnippet> <codesnippet>}</codesnippet> <codesnippet>public»interface»IOrderService</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»public»Task<List<Order>>»</codesnippet> <codesnippet>»»»GetAllOrdersAsync();</codesnippet> <codesnippet>»»»public»Task<Order>»</codesnippet> <codesnippet>»»»GetOrderAsync(int»Id);</codesnippet> <codesnippet>»»»public»Task»</codesnippet> <codesnippet>»»»CreateOrderAsync(Order»order);</codesnippet> <codesnippet>}</codesnippet> <codesnippet>public»async»Task»</codesnippet> <codesnippet> CreateOrderAsync(Order»order)</codesnippet> <codesnippet>{</codesnippet> <codesnippet>using»var»transaction»=</codesnippet> <codesnippet>»_context.Database.BeginTransaction();</codesnippet> <codesnippet>»»»»try</codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»»»»»_context.Orders.Add(order);</codesnippet> <codesnippet>»»»»»»»»await»_context.SaveChangesAsync();</codesnippet> <codesnippet>»»»»»»»»var»outboxMessage»=»new»OutboxMessage</codesnippet> <codesnippet>»»»»»»»»{</codesnippet> <codesnippet>»»»»»»»»»»»»Event_Payload»=</codesnippet> <codesnippet>»»»»»»»»»»»»JsonSerializer.Serialize(order),</codesnippet> <codesnippet>»»»»»»»»»»»»Event_Date»=»DateTime.Now,</codesnippet> <codesnippet>»»»»»»»»»»»»IsMessageDispatched»=»false</codesnippet> <codesnippet>»»»»»»»»};</codesnippet> <codesnippet> »»»»»»»»_context.OutboxMessages. </codesnippet> <codesnippet> »»»»»»»»Add(outboxMessage); </codesnippet> <codesnippet> »»»»»»»»await»_context.SaveChangesAsync(); </codesnippet> <codesnippet> »»»»»»»»await»transaction.CommitAsync(); </codesnippet> <codesnippet> »»»»} </codesnippet> <codesnippet> »»»»catch </codesnippet> <codesnippet> »»»»{ </codesnippet> <codesnippet> »»»»»»»»await»transaction.RollbackAsync(); </codesnippet> <codesnippet> »»»»»»»»throw;</codesnippet> <codesnippet>»»»»}</codesnippet> <codesnippet>}</codesnippet> <h3>Create the OutboxMessageProcessor Background Service</h3> <codesnippet>public»interface»IHostedService</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»Task»StartAsync</codesnippet> <codesnippet>»»»»(CancellationToken»cancellationToken);</codesnippet> <codesnippet>»»»»Task»StopAsync</codesnippet> <codesnippet>»»»»(CancellationToken»cancellationToken);</codesnippet> <codesnippet>}</codesnippet> <codesnippet>public»abstract»class»</codesnippet> <codesnippet> BackgroundService»:»»</codesnippet> <codesnippet> IHostedService,»</codesnippet> <codesnippet> IDisposable </codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»public»virtual»void»Dispose();</codesnippet> <codesnippet>»»»»public»virtual»Task»</codesnippet> <codesnippet>»»»»StartAsync(CancellationToken»</codesnippet> <codesnippet>»»»»cancellationToken);</codesnippet> <codesnippet>»»»»public»virtual»Task»</codesnippet> <codesnippet>»»»»StopAsync(CancellationToken»</codesnippet> <codesnippet>»»»»cancellationToken);</codesnippet> <codesnippet>»»»»protected»abstract»Task»</codesnippet> <codesnippet>»»»»ExecuteAsync(CancellationToken»</codesnippet> <codesnippet>»»»»stoppingToken);</codesnippet> <codesnippet>}</codesnippet> <codesnippet> public»class»OutboxMessageProcessor»:»</codesnippet> <codesnippet>BackgroundService</codesnippet> <codesnippet>{»»»»»»»»</codesnippet> <codesnippet>»»»protected»override»async»Task»</codesnippet> <codesnippet>»»»ExecuteAsync(CancellationToken»</codesnippet> <codesnippet>»»»cancellationToken)</codesnippet> <codesnippet>»»»{</codesnippet> <codesnippet>»»»»»»»»»»»»</codesnippet> <codesnippet>»»»}»»»»»»»»</codesnippet> <codesnippet>}</codesnippet> <codesnippet>private»async»Task»</codesnippet> <codesnippet> PublishOutboxMessagesAsync </codesnippet> <codesnippet>(CancellationToken»cancellationToken)</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»try</codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»»»»»using»var»scope»=»</codesnippet> <codesnippet>»»»»»»»_scopeFactory.CreateScope();</codesnippet> <codesnippet>»»»»»»»»await»using»var»_dbContext»=»</codesnippet> <codesnippet>»»»»»»»»scope.ServiceProvider.</codesnippet> <codesnippet>»»»»»»»»GetRequiredService</codesnippet> <codesnippet>»»»»»»»»<ApplicationDbContext>();</codesnippet> <codesnippet>»»»»»»»»List<OutboxMessage>»messages»=»</codesnippet> <codesnippet>»»»»»»»»_dbContext.OutboxMessages.Where(om»=>»</codesnippet> <codesnippet>»»»»»»»»om.IsMessageDispatched»</codesnippet> <codesnippet>»»»»»»»»!=»false).ToList();</codesnippet> <codesnippet>»»»»»»»»foreach»(OutboxMessage»outboxMessage»</codesnippet> <codesnippet>»»»»»»»»in»messages)</codesnippet> <codesnippet>»»»»»»»»{</codesnippet> <codesnippet>»»»»»»»»»»»»try</codesnippet> <codesnippet>»»»»»»»»»»»»{</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»await»_producer.</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»SendMessageToKafkaAsync</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»(outboxMessage);</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»outboxMessage.</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»IsMessageDispatched»=»true;</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»outboxMessage.Event_Date»=»</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»DateTime.UtcNow;</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»_dbContext.OutboxMessages.</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»Update(outboxMessage);</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»await»_dbContext.</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»SaveChangesAsync();</codesnippet> <codesnippet>»»»»»»»»»»»»}</codesnippet> <codesnippet>»»»»»»»»»»»»catch»(Exception»e)</codesnippet> <codesnippet>»»»»»»»»»»»»{</codesnippet> <codesnippet>»»»»»»»»»»»»»»»»Console.WriteLine(e);</codesnippet> <codesnippet>»»»»»»»»»»»»}</codesnippet> <codesnippet>»»»»»»»»}</codesnippet> <codesnippet>»»»»}</codesnippet> <codesnippet>»»»»catch</codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»»»»»throw;</codesnippet> <codesnippet>»»»»}</codesnippet> <codesnippet>»»»»await»Task.Delay</codesnippet> <codesnippet>»»»»(TimeSpan.FromSeconds(5),»</codesnippet> <codesnippet>»»»»cancellationToken);</codesnippet> <codesnippet>}</codesnippet> <h3>Create the OutboxMessageRepository</h3> <codesnippet> namespace»Outbox_Pattern_Demo</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»public»interface»</codesnippet> <codesnippet>»»»»IOutboxMessageRepository</codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»»»»»Task<IreadOnlyCollection</codesnippet> <codesnippet>»»»»»»»»<OutboxMessage>>»</codesnippet> <codesnippet>»»»»»»»»GetUnsentMessagesAsync();</codesnippet> <codesnippet>»»»»»»»»Task<IreadOnlyCollection</codesnippet> <codesnippet>»»»»»»»»<OutboxMessage>></codesnippet> <codesnippet>»»»»»»»»GetMessagesByIdsAsync</codesnippet> <codesnippet>»»»»»»»»(IEnumerable<int>»ids);</codesnippet> <codesnippet>»»»»»»»»Task»UpdateAsync</codesnippet> <codesnippet>»»»»»»»»(OutboxMessage»message,»</codesnippet> <codesnippet>»»»»»»»»bool»status);</codesnippet> <codesnippet>»»»»};</codesnippet> <codesnippet>}</codesnippet> The complete source code of the OutboxMessageRepository class is given in Listing 4. listi <codesnippet> [HttpPost("CreateOrder")]</codesnippet> <codesnippet>public»async»Task<IActionResult>»</codesnippet> <codesnippet> CreateOrder([FromBody]»Order»order)</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»await»_orderService.</codesnippet> <codesnippet>»»»»CreateOrderAsync(order);</codesnippet> <codesnippet>»»»»return»Ok();</codesnippet> <codesnippet>}</codesnippet> <h3>Send Messages to Apache Kafka</h3> Create an interface named IKafkaProducer in a file named IKafkaProducerr.cs and write the following code in there: <codesnippet> public»interface»IKafkaProducer </codesnippet> <codesnippet> { </codesnippet> <codesnippet> »»»Task»SendMessageToKafkaAsync </codesnippet> <codesnippet> »»»(OutboxMessage»message);» </codesnippet> <codesnippet> }; </codesnippet> Next, create a new class named KafkaProducer in a file having the same name with a .cs extension and write the code given in Listing 6 in there. In addition, you should register the instances of type ImessageRepository and IKafkaProducer with the service collection, as shown below: <codesnippet>builder.Services.AddScoped</codesnippet> <codesnippet><IOutboxMessageRepository,»</codesnippet> <codesnippet>OutboxMessageRepository>();</codesnippet> <codesnippet>builder.Services.AddScoped</codesnippet> <codesnippet><IKafkaProducer,»KafkaProducer>();</codesnippet> You should also register the hosted service in the Program.cs file, as shown in the code snippet given below: <codesnippet>services.AddSingleton</codesnippet> <codesnippet><IHostedService,»</codesnippet> <codesnippet>OutboxMessageProcessor>();</codesnippet> <h3>Sequence Diagram of the Complete Flow</h3> The sequence diagram of the complete flow is shown in Figure 8. <figure id="8" src="image8.png"> Figure 8 : A sequence diagram of the complete flow</figure> <h2>Create the Message Consumer Application</h2> Let’s create a message consumer to consume the messages stored in Kafka. To do this, create another ASP.NET Core application by following the steps mentioned earlier in this article. Create a new interface named IKafkaConsumer and write the following code in there: <codesnippet> namespace»Outbox_Pattern_Demo</codesnippet> <codesnippet>{</codesnippet> <codesnippet>»»»»public»interface»IKafkaConsumer</codesnippet> <codesnippet>»»»»{</codesnippet> <codesnippet>»»»»»»»»public»Task»ConsumeMessagesAsync</codesnippet> <codesnippet>»»»»»»»»(CancellationToken»cancellationToken);</codesnippet> <codesnippet>»»»»}</codesnippet> <codesnippet>}</codesnippet> Next, create a new class named KafkaConsumer that implements the IKafkaConsumer interface, as shown in Listing 8. You now need a service that runs in the background and consumes the messages residing in Kafka. To do this, create a class named KafkaMessageProcessor that extends the BackgroundService class, as shown in Listing 9. Register the instance of type IKafkaConsumer as a scoped service and the instance of KafkaMessageProcessor as a singleton hosted service. <codesnippet>builder.Services.AddScoped</codesnippet> <codesnippet><IKafkaConsumer,»KafkaConsumer>();</codesnippet> <codesnippet>builder.Services.AddSingleton</codesnippet> <codesnippet><IHostedService,»KafkaMessageProcessor>();</codesnippet> Figure 9 illustrates the complete flow visually. <figure id="9" src="image9.png"> Figure 9 : The complete flow of the consumer application</figure> Listing 10 shows the Program.cs file of the consumer application. <h2>Execute the Application</h2> Run the producer and the consumer applications separately because they’re part of different solutions. You should also set appropriate breakpoints in the source code of both applications so that you can debug them. To run the application, follow the steps outlined below: <list type="numbered"> <numberedlist>Start ZooKeeper in a terminal window.</numberedlist> <numberedlist>Start Kafka in another terminal window.</numberedlist> <numberedlist>Execute the producer application.</numberedlist> <numberedlist>Execute the consumer application.</numberedlist> <numberedlist>Launch the Postman Http Debugger tool.</numberedlist> <numberedlist>Send an HTTP POST request to the producer API using Postman.</numberedlist> </list> Figure 10 shows how you can invoke the API endpoint in Postman. <figure id="10" src="image10.png"> Figure 1 0 : Invoking the API endpoint in Postman</figure> <h2>Real-World Use Cases for the Outbox Pattern</h2> The outbox pattern has several use cases in real-life: real-time notifications and data synchronization. The next two sections discuss these. <h3>Real-Time Notifications</h3> You might often want real-time notifications in your application to send instant alerts, messages, etc. You can use the outbox pattern to implement notifications in your application in real-time. These events can then be processed asynchronously, thereby triggering real-time notifications to the logged-in users of the application. <h3>Data Synchronization</h3> Real-time data synchronization is yet another use case where the outbox pattern fits in. For example, in an e-commerce application when a customer places an order, both the order and the stock database tables are updated simultaneously in the same transaction. Additionally, the order, product, and cart data must be synchronized across all devices in use using asynchronous processing. <h2>Alternatives to the Transactional Outbox Pattern</h2> Although the outbox pattern is a popular strategy for handling distributed transactions efficiently and ensuring reliable and consistent communication between microservices, there are certain downsides to using it as well. I’ve discussed them earlier in this article. The Two-Phase Commit and the Saga Pattern are two popular alternatives to this strategy. <h3>Two-Phase Commit</h3> This approach is used for performing an atomic transaction across multiple resources. Keep in mind that there are two phases in this strategy. While the transaction coordinator notifies all other resources about its desire to execute a transaction in the first phase, it instructs all resources to execute this operation in the next phase. The major drawback of such an approach is that if any resource fails or there are network issues, the whole transaction stops and locks are applied on resources. <h3>Saga Pattern</h3> This approach splits the transactions into multiple steps and completes each step as a separate transaction. This particular design is best suited for long-running transactions or those that take place between microservices. However, when you adopt this approach, it can be difficult to manage which transactions must be rolled back in the event of an error in your application. This is because the transaction timing and sequence for such transactions can be difficult to comprehend. <h2>Where Do We Go from Here?</h2> The outbox pattern, a reliable pattern for consistent event publishing in microservices-based applications, is a great way to make your microservices apps more stable and scalable. It helps separate services through which the events are successfully delivered and isolates business operations from publishing events. However, it also has certain downsides, which explains why you should understand these constraints before deciding on your application's design. I’ll discuss more on the Outbox Pattern in a future article. </body> <sidebars> <sidebar title="Sponsored Sidebar " /> <sidebar title="Ready to Modernize a Legacy App?"> Need advice on migrating yesterday’s legacy applications to today’s modern platforms? Take advantage of CODE Consulting’s years of experience and contact us today to schedule a FREE consulting call to discuss your options. No strings. No commitment. For more information , see www.codemag.com/consulting or email us at info@codemag.com. </sidebar> </sidebars> <tables /> <codelistings> <codelisting id="1" header="Listing 1: The ApplicationDbContext class"> <code> using»Microsoft.EntityFrameworkCore;</code> <code /> <code> namespace»Outbox_Pattern_Demo</code> <code>{</code> <code>»»»»public»class»ApplicationDbContext»:»DbContext</code> <code>»»»»{</code> <code>»»»»»»»»public»ApplicationDbContext(DbContextOptions</code> <code>»»»»»»»»<ApplicationDbContext>»options)»:»base(options)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»protected»override»void»OnModelCreating</code> <code>»»»»»»»»(ModelBuilder»modelBuilder)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»modelBuilder.Entity<Order>(entity»=></code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»entity.ToTable("Order");</code> <code>»»»»»»»»»»»»»»»»entity.HasKey(e»=>»e.Order_Id);</code> <code>»»»»»»»»»»»»});</code> <code /> <code>»»»»»»»»»»»»modelBuilder.Entity</code> <code>»»»»»»»»»»»»<OutboxMessage>(entity»=></code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»entity.ToTable("OutboxMessage");</code> <code>»»»»»»»»»»»»»»»»entity.HasKey(e»=>»e.Event_Id);</code> <code>»»»»»»»»»»»»});</code> <code /> <code>»»»»»»»»»»»»base.OnModelCreating(modelBuilder);</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»DbSet<OutboxMessage>»</code> <code>»»»»»»»»OutboxMessages»{»get;»set;»}</code> <code>»»»»»»»»public»DbSet<Order>»</code> <code>»»»»»»»»Orders»{»get;»set;»}</code> <code>»»»»}</code> <code>}</code> </codelisting> <codelisting id="2" header="Listing 2: The OrderService Class"> <code> namespace»Outbox_Pattern_Demo</code> <code>{</code> <code>»»»using»System.Text.Json;</code> <code /> <code>»»»»public»class»OrderService»:»IOrderService</code> <code>»»»»{</code> <code>»»»»»»»»private»readonly»ApplicationDbContext»_context;</code> <code /> <code>»»»»»»»»public»OrderService(ApplicationDbContext»context)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»_context»=»context;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»async»Task<List<Order>>»GetAllOrdersAsync()</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»return»await»Task.FromResult</code> <code>»»»»»»»»»»»»(_context.Orders.</code> <code>»»»»»»»»»»»»ToList<Order>());</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»async»Task<Order>»</code> <code>»»»»»»»»GetOrderAsync(int»Id)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»return»await»Task.FromResult</code> <code>»»»»»»»»»»»»(_context.Orders.FirstOrDefault</code> <code>»»»»»»»»»»»»(x»=>»x.Order_Id»==»Id));</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»async»Task»CreateOrderAsync(Order»order)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»using»var»transaction»=»</code> <code>»»»»»»»»»»»»_context.Database.BeginTransaction();</code> <code /> <code>»»»»»»»»»»»»try</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»_context.Orders.Add(order);</code> <code>»»»»»»»»»»»»»»»»await»_context.SaveChangesAsync();</code> <code /> <code>»»»»»»»»»»»»»»»»var»outboxMessage»=»new»OutboxMessage</code> <code>»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»Event_Payload»=»</code> <code>»»»»»»»»»»»»»»»»»»»»JsonSerializer.Serialize(order),</code> <code>»»»»»»»»»»»»»»»»»»»»Event_Date»=»DateTime.Now,</code> <code>»»»»»»»»»»»»»»»»»»»»IsMessageDispatched»=»false</code> <code>»»»»»»»»»»»»»»»»};</code> <code /> <code> »»»»»»»»»»»»»»»»_context.OutboxMessages. </code> <code> »»»»»»»»»»»»»»»»Add(outboxMessage); </code> <code /> <code> »»»»»»»»»»»»»»»» await »_context.SaveChangesAsync(); </code> <code> »»»»»»»»»»»»»»»» await »transaction.CommitAsync(); </code> <code> »»»»»»»»»»»»} </code> <code> »»»»»»»»»»»» catch </code> <code> »»»»»»»»»»»»{ </code> <code> »»»»»»»»»»»»»»»» await »transaction.RollbackAsync(); </code> <code> »»»»»»»»»»»»»»»» throw;</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»}</code> <code>»»»»}</code> <code>}</code> </codelisting> <codelisting id="3" header="Listing 3: The OutboxMessageProcessor class"> <code> namespace»Outbox_Pattern_Demo</code> <code>{</code> <code>»»»»public»class»OutboxMessageProcessor»:»BackgroundService</code> <code>»»»»{</code> <code>»»»»»»»»private»readonly»IServiceScopeFactory»_scopeFactory;</code> <code>»»»»»»»»private»readonly»IKafkaProducer»_producer;</code> <code /> <code>»»»»»»»»public»OutboxMessageProcessor</code> <code>»»»»»»»»(IServiceScopeFactory»scopeFactory,»</code> <code>»»»»»»»»IKafkaProducer»producer)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»_scopeFactory»=»scopeFactory;</code> <code>»»»»»»»»»»»»_producer»=»producer;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»protected»override»async»Task»ExecuteAsync</code> <code>»»»»»»»»(CancellationToken»cancellationToken)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»while»(!cancellationToken.IsCancellationRequested)</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»await»PublishOutboxMessagesAsync</code> <code>»»»»»»»»»»»»»»»»(cancellationToken);</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»private»async»Task»PublishOutboxMessagesAsync</code> <code>»»»»»»»»(CancellationToken»cancellationToken)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»try</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»using»var»scope»=»_scopeFactory.CreateScope();</code> <code>»»»»»»»»»»»»»»»»await»using»var»_dbContext»=»</code> <code>»»»»»»»»»»»»»»»»scope.ServiceProvider.GetRequiredService</code> <code>»»»»»»»»»»»»»»»»<ApplicationDbContext>();</code> <code /> <code>»»»»»»»»»»»»»»»»List<OutboxMessage>»messages»=»</code> <code>»»»»»»»»»»»»»»»»»_dbContext.OutboxMessages.Where</code> <code>»»»»»»»»»»»»»»»»(om»=>»om.IsMessageDispatched»!=»false).ToList();</code> <code /> <code>»»»»»»»»»»»»»»»»foreach»(OutboxMessage»outboxMessage»in»messages)</code> <code>»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»try</code> <code>»»»»»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»await»_producer.</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»SendMessageToKafkaAsync(outboxMessage);</code> <code /> <code>»»»»»»»»»»»»»»»»»»»»»»»»outboxMessage.IsMessageDispatched»=»true;</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»outboxMessage.Event_Date»=»DateTime.UtcNow;</code> <code /> <code>»»»»»»»»»»»»»»»»»»»»»»»»_dbContext.OutboxMessages.</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»Update(outboxMessage);</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»await»_dbContext.SaveChangesAsync();</code> <code>»»»»»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»»»»»»»»»catch»(Exception»e)</code> <code>»»»»»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»Console.WriteLine(e);</code> <code>»»»»»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»catch</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»throw;</code> <code>»»»»»»»»»»»»}</code> <code /> <code>»»»»»»»»»»»»await»Task.Delay</code> <code>»»»»»»»»»»»»(TimeSpan.FromSeconds(5),»</code> <code>»»»»»»»»»»»»cancellationToken);</code> <code>»»»»»»»»}</code> <code>»»»»}</code> <code>}</code> </codelisting> <codelisting id="4" header="Listing 4: The OutboxMessageRepository class"> <code> using»System.Collections.ObjectModel;</code> <code> using»System.Data;</code> <code /> <code> namespace»Outbox_Pattern_Demo</code> <code>{</code> <code>»»»»public»class»OutboxMessageRepository»:»</code> <code>»»»»IOutboxMessageRepository</code> <code>»»»»{</code> <code>»»»»»»»»private»readonly»ApplicationDbContext»_context;</code> <code>»»»»»»»»private»IreadOnlyCollection</code> <code>»»»»»»»»<OutboxMessage>»_outboxMessages;</code> <code /> <code>»»»»»»»»public»OutboxMessageRepository</code> <code>»»»»»»»»(ApplicationDbContext»context)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»_context»=»context;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»IreadOnlyCollection</code> <code>»»»»»»»»<OutboxMessage>»OutboxMessages</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»get</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»return»_outboxMessages»??»</code> <code>»»»»»»»»»»»»»»»»(_outboxMessages»=»new»</code> <code>»»»»»»»»»»»»»»»»ReadOnlyCollection<OutboxMessage></code> <code>»»»»»»»»»»»»»»»»(_context.OutboxMessages.ToList()));</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»async»Task<IreadOnlyCollection</code> <code>»»»»»»»»<OutboxMessage>>»GetUnsentMessagesAsync()</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»List<OutboxMessage>?»unsentMessages»=</code> <code>»»»»»»»»»»»»_context.OutboxMessages.Where</code> <code>»»»»»»»»»»»»(e»=>»e.IsMessageDispatched»!=»true).ToList();</code> <code>»»»»»»»»»»»»ReadOnlyCollection<OutboxMessage>?»</code> <code>»»»»»»»»»»»»result»=»new»ReadOnlyCollection</code> <code>»»»»»»»»»»»»<OutboxMessage>(unsentMessages);</code> <code>»»»»»»»»»»»»return»result;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»async»Task<IreadOnlyCollection</code> <code>»»»»»»»»<OutboxMessage>>»GetMessagesByIdsAsync</code> <code>»»»»»»»»(IEnumerable<int>»ids)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»List<OutboxMessage>?»</code> <code>»»»»»»»»»»»»orders»=»_context.</code> <code>»»»»»»»»»»»»OutboxMessages.ToList();</code> <code>»»»»»»»»»»»»var»readOnlyOrders»=»</code> <code>»»»»»»»»»»»»new»ReadOnlyCollection</code> <code>»»»»»»»»»»»»<OutboxMessage>(orders);</code> <code>»»»»»»»»»»»»return»readOnlyOrders;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»async»Task»UpdateAsync</code> <code>»»»»»»»»(OutboxMessage»message,»bool»status)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»var»entity»=»_context.OutboxMessages.</code> <code>»»»»»»»»»»»»FirstOrDefault(o»=>»o.Event_Id»==»message.Event_Id);</code> <code /> <code>»»»»»»»»»»»»if»(entity»!=»null)</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»entity.Event_Id»=»message.Event_Id;</code> <code>»»»»»»»»»»»»»»»»entity.Event_Date»=»message.Event_Date;</code> <code> »»»»»»»»»»»»»»»»entity.Event_Payload»=»message.Event_Payload; </code> <code> »»»»»»»»»»»»»»»»entity.IsMessageDispatched»= </code> <code> »»»»»»»»»»»»»»»»message.IsMessageDispatched; </code> <code> »»»»»»»»»»»»»»»» await »_context.SaveChangesAsync(); </code> <code> »»»»»»»»»»»»} </code> <code> »»»»»»»»} </code> <code> »»»»};</code> <code>}</code> </codelisting> <codelisting id="5" header="Listing 5: The OrderController Class"> <code> using»Microsoft.AspNetCore.Mvc;</code> <code /> <code> namespace»Outbox_Pattern_Demo.Controllers</code> <code>{</code> <code>»»»»[Route("api/[controller]")]</code> <code>»»»»[ApiController]</code> <code>»»»»public»class»OrderController»:»ControllerBase</code> <code>»»»»{</code> <code>»»»»»»»»private»IOrderService»_orderService;</code> <code>»»»»»»»»public»OrderController</code> <code>»»»»»»»»(IOrderService»orderService)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»_orderService»=»orderService;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»[HttpGet("GetOrders")]</code> <code>»»»»»»»»public»async»Task<List<Order>>»GetOrders()</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»return»await»_orderService.</code> <code>»»»»»»»»»»»»GetAllOrdersAsync();</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»[HttpGet("{id}")]</code> <code>»»»»»»»»public»async»Task<Order>»GetOrder(int»id)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»return»await»_orderService.</code> <code>»»»»»»»»»»»»GetOrderAsync(id);</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»[HttpPost("CreateOrder")]</code> <code>»»»»»»»»public»async»Task<IActionResult>»</code> <code>»»»»»»»»CreateOrder([FromBody]»Order»order)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»await»_orderService.</code> <code>»»»»»»»»»»»»CreateOrderAsync(order);</code> <code>»»»»»»»»»»»»return»Ok();</code> <code>»»»»»»»»}</code> <code>»»»»}</code> <code>}</code> </codelisting> <codelisting id="6" header="Listing 6: The KafkaProducer class"> <code> using»Confluent.Kafka;</code> <code> using»System.Net;</code> <code /> <code> namespace»Outbox_Pattern_Demo</code> <code>{</code> <code>»»»»public»class»KafkaProducer»:»IKafkaProducer</code> <code>»»»»{</code> <code>»»»»»»»»private»readonly»ProducerConfig»_producerConfig;</code> <code>»»»»»»»»private»readonly»IOutboxRepository»_outboxRepository;</code> <code>»»»»»»»»private»readonly»string»topic»=»"test";</code> <code /> <code>»»»»»»»»public»KafkaProducer(IOutboxRepository»outboxRepository)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»_outboxRepository»=»outboxRepository;</code> <code /> <code>»»»»»»»»»»»»_producerConfig»=»new»ProducerConfig</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»BootstrapServers»=»"localhost:9092",</code> <code>»»»»»»»»»»»»»»»»ClientId»=»Dns.GetHostName()</code> <code>»»»»»»»»»»»»};</code> <code /> <code>»»»»»»»»»»»»_outboxRepository»=»outboxRepository;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»public»async»Task»SendMessageToKafkaAsync</code> <code>»»»»»»»»(OutboxMessage»message)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»if»(message»==»null)</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»throw»new»ArgumentNullException</code> <code>»»»»»»»»»»»»»»»»(nameof(message));</code> <code>»»»»»»»»»»»»}</code> <code /> <code>»»»»»»»»»»»»using»var»producer»=»</code> <code>»»»»»»»»»»»»new»ProducerBuilder<Null,»</code> <code>»»»»»»»»»»»»string>(_producerConfig).Build();</code> <code /> <code>»»»»»»»»»»»»try</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»var»result»=»await»producer.ProduceAsync</code> <code>»»»»»»»»»»»»»»»»»»»»(topic,»new»Message<Null,»string></code> <code>»»»»»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»Value»=»message.EventPayload</code> <code>»»»»»»»»»»»»»»»»»»»»});</code> <code /> <code>»»»»»»»»»»»»»»»»if»(result.Status»==»PersistenceStatus.Persisted)</code> <code>»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»await»_outboxRepository.UpdateAsync</code> <code>»»»»»»»»»»»»»»»»»»»»(message,»OutboxMessageStatus.Sent);</code> <code>»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»catch»(Exception)</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»await»_outboxRepository.UpdateAsync</code> <code>»»»»»»»»»»»»»»»»(message,»OutboxMessageStatus.New);</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»}</code> <code>»»»»};</code> <code>}</code> </codelisting> <codelisting id="7" header="Listing 7: The Program.cs file (Producer Application)"> <code> using»Microsoft.EntityFrameworkCore;</code> <code> using»Outbox_Pattern_Demo;</code> <code /> <code> var»builder»=»WebApplication.CreateBuilder(args);</code> <code /> <code> //»Add»services»to»the»container. </code> <code /> <code>builder.Services.AddControllers();</code> <code>builder.Services.AddScoped</code> <code><IOrderService,»OrderService>();</code> <code>builder.Services.AddScoped</code> <code><IOutboxMessageRepository,»</code> <code>OutboxMessageRepository>();</code> <code>builder.Services.AddScoped</code> <code><IKafkaProducer,»KafkaProducer>();</code> <code /> <code>builder.Services.AddSingleton</code> <code><IHostedService,»OutboxMessageProcessor>();</code> <code /> <code>builder.Services.AddDbContext</code> <code><ApplicationDbContext>(options»=></code> <code>options.UseSqlServer</code> <code>(@"Data»Source=JOYDIP;</code> <code> Initial»Catalog=Outbox_Pattern_Demo; </code> <code> Trusted_Connection=True;» </code> <code> TrustServerCertificate=True; </code> <code> Integrated»Security=True;"));</code> <code /> <code> var»app»=»builder.Build();</code> <code /> <code> //»Configure»the»HTTP»request»pipeline. </code> <code /> <code>app.UseAuthorization();</code> <code>app.MapControllers();</code> <code>app.Run();</code> </codelisting> <codelisting id="8" header="Listing 8: The KafkaConsumer class"> <code> using»Confluent.Kafka;</code> <code> using»System.Text.Json;</code> <code /> <code> namespace»Outbox_Pattern_Demo</code> <code>{</code> <code>»»»»public»class»KafkaConsumer»:»IKafkaConsumer</code> <code>»»»»{</code> <code>»»»»»»»»private»readonly»string»topic»=»"test";</code> <code>»»»»»»»»private»readonly»string»groupId»=»"test_group";</code> <code>»»»»»»»»private»readonly»string»bootstrapServers»=»</code> <code>»»»»»»»»"localhost:9092";</code> <code /> <code>»»»»»»»»public»async»Task»ConsumeMessagesAsync</code> <code>»»»»»»»(CancellationToken»cancellationToken)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»var»config»=»new»ConsumerConfig</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»GroupId»=»groupId,</code> <code>»»»»»»»»»»»»»»»»BootstrapServers»=»bootstrapServers,</code> <code>»»»»»»»»»»»»»»»»AutoOffsetReset»=»AutoOffsetReset.Earliest</code> <code>»»»»»»»»»»»»};</code> <code /> <code>»»»»»»»»»»»»try</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»using»(var»consumerBuilder»=»new»ConsumerBuilder</code> <code>»»»»»»»»»»»»»»»»<Ignore,»string>(config).Build())</code> <code>»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»consumerBuilder.Subscribe(topic);</code> <code>»»»»»»»»»»»»»»»»»»»»var»cancelToken»=»new»CancellationTokenSource();</code> <code /> <code>»»»»»»»»»»»»»»»»»»»»try</code> <code>»»»»»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»while»(true)</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»»»»»var»consumer»=»consumerBuilder.Consume</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»»»»»»»»(cancelToken.Token);</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»»»»»var»order»=»JsonSerializer.Deserialize</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»»»»»»»»»<Order>(consumer.Message.Value);</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»»»»»»»»»catch»(OperationCanceledException)</code> <code>»»»»»»»»»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»»»»»»»»»consumerBuilder.Close();</code> <code>»»»»»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»»»»»catch</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»throw;</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»}</code> <code>»»»»}</code> <code>}</code> </codelisting> <codelisting id="9" header="Listing 9: The KafkaMessageProducer class"> <code> namespace»Outbox_Pattern_Demo</code> <code>{</code> <code>»»»»public»class»KafkaMessageProcessor»:»BackgroundService</code> <code>»»»»{</code> <code>»»»»»»»»private»readonly»IServiceScopeFactory»_scopeFactory;</code> <code>»»»»»»»»private»readonly»IKafkaConsumer»_consumer;</code> <code /> <code>»»»»»»»»public»KafkaMessageProcessor</code> <code>»»»»»»»»(IServiceScopeFactory»scopeFactory,»</code> <code>»»»»»»»»IKafkaConsumer»consumer)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»_scopeFactory»=»scopeFactory;</code> <code>»»»»»»»»»»»»_consumer»=»consumer;</code> <code>»»»»»»»»}</code> <code /> <code>»»»»»»»»protected»override»async»Task»ExecuteAsync</code> <code>»»»»»»»»(CancellationToken»cancellationToken)</code> <code>»»»»»»»»{</code> <code>»»»»»»»»»»»»while»(!cancellationToken.IsCancellationRequested)</code> <code>»»»»»»»»»»»»{</code> <code>»»»»»»»»»»»»»»»»await»_consumer.ConsumeMessagesAsync</code> <code>»»»»»»»»»»»»»»»»(cancellationToken);</code> <code>»»»»»»»»»»»»}</code> <code>»»»»»»»»}</code> <code>»»»»}</code> <code>}</code> </codelisting> <codelisting id="10" header="Listing 10: The Program.cs file (Consumer Application)"> <code> using»Microsoft.EntityFrameworkCore;</code> <code> using»Outbox_Pattern_Client_Demo;</code> <code /> <code> var»builder»=»WebApplication.CreateBuilder(args);</code> <code /> <code> //»Add»services»to»the»container. </code> <code /> <code>builder.Services.AddControllers();</code> <code>builder.Services.AddScoped</code> <code><IKafkaConsumer,»KafkaConsumer>();</code> <code>builder.Services.AddSingleton</code> <code><IHostedService,»</code> <code>KafkaMessageProcessor>();</code> <code /> <code> var»app»=»builder.Build();</code> <code /> <code> //»Configure»the»HTTP»request»pipeline. </code> <code /> <code>app.UseAuthorization();</code> <code>app.MapControllers();</code> <code>app.Run();</code> </codelisting> </codelistings> </document>

Add New Technology/Category
Category Name: *	(Example: .NET)
Category Group: (optional)	(Example: .NET, Visual Studio, Web Development)