The .NET development community at Raiffeisenbank continues with a brief overview of the content from ViennaNET. You can read about how and why we got to this point .
In this article, we will go through the libraries for working with distributed transactions, queues, and databases that have not yet been covered, which can be found in our GitHub repository (), and .
ViennaNET.Sagas
When a project transitions to DDD and microservices architecture, distributing business logic across various services raises a challenge concerning the need to implement a mechanism for distributed transactions, as many scenarios often impact multiple domains. You can learn more about such mechanisms in .
In our projects, we implemented a simple yet useful mechanism: a saga, more precisely, an orchestration-based saga. Its essence is as follows: there is a business scenario where operations need to be sequentially performed in different services, and if any issues arise at any step, it is necessary to invoke a rollback for all previous steps where it is planned. Thus, at the end of the saga execution, regardless of success, we achieve consistent data across all domains.
Our implementation is currently basic and not tied to any specific methods of interaction with other services. It is easy to apply: simply create a subclass from the abstract base class SagaBase, where T is your context class that can store the original data required for the saga's operation, as well as some intermediate results. An instance of the context will be passed to all steps during execution. The saga itself is a stateless class, so the instance can be registered in DI as a Singleton to obtain the necessary dependencies.
Example declaration:
public class ExampleSaga : SagaBase
{
public ExampleSaga()
{
Step("Step 1")
.WithAction(c => ...)
.WithCompensation(c => ...);
AsyncStep("Step 2")
.WithAction(async c => ...);
}
}
Example call:
var saga = new ExampleSaga();
var context = new ExampleContext();
await saga.Execute(context);
You can see full examples of different implementations and in the assembly with .
ViennaNET.Orm.*
A set of libraries for working with various databases through NHibernate. We use a DB-First approach with Liquibase, so only the functionality for working with data in the finished database is present here.
ViennaNET.Orm.Seedwork and ViennaNET.Orm are the main assemblies containing basic interfaces and their implementations, respectively. Let's take a closer look at their contents.
The PerformanceResourceTiming IEntityFactoryService and its implementation EntityFactoryService are the main entry points for working with the database, as here the Unit of Work is created, repositories for working with specific entities, as well as command executors and direct SQL queries. Sometimes it is convenient to limit the capabilities of the class for database operations, for example, to allow only data reading. For such cases, there is a parent interface IEntityFactoryService IEntityRepositoryFactory , which declares only the method for creating repositories.To directly access the database, the provider mechanism is used. For each DBMS used in our teams, there is its own implementation:
ViennaNET.Orm.MSSQL, ViennaNET.Orm.Oracle, ViennaNET.Orm.SQLite, ViennaNET.Orm.PostgreSql At the same time, multiple providers can be registered in one application simultaneously, which allows for step-by-step migration from one DBMS to another within a single service without any infrastructure modification costs. The mechanism for selecting the required connection and, consequently, the provider for a specific entity class (for which the mapping to the database tables is written) is implemented through entity registration in the BoundedContext class (which contains a method for registering domain entities) or its subclass ApplicationContext (which contains methods for registering application entities, direct queries, and commands), where the connection identifier from the configuration is taken as an argument:.
"db": [ { "nick": "mssql_connection", "dbServerType": "MSSQL", "ConnectionString": "...", "useCallContext": true }, { "nick": "oracle_connection", "dbServerType": "Oracle", "ConnectionString": "..." } ],
Example ApplicationContext:
internal sealed class DbContext : ApplicationContext { public DbContext() { AddEntity("mssql_connection"); AddEntity("oracle_connection"); AddEntity("oracle_connection"); } }
If the connection identifier is not specified, the connection named "default" will be used.
If the connection identifier is not specified, the connection named "default" will be used.
Entity mapping to database tables is implemented using standard NHibernate features. You can use descriptions through both XML files and classes. For convenient creation of stub repositories in unit tests, there is a library ViennaNET.TestUtils.Orm.
Full usage examples of ViennaNET.Orm.* can be found .
ViennaNET.Messaging.*
A set of libraries for working with queues.
The approach for working with queues is similar to that used with various DBMS, namely a maximally unified approach from the perspective of working with the library, regardless of the queue manager used. The library ViennaNET.Messaging is responsible for this unification, and ViennaNET.Messaging.MQSeriesQueue, ViennaNET.Messaging.RabbitMQQueue, and ViennaNET.Messaging.KafkaQueue contain implementations of adapters for IBM MQ, RabbitMQ, and Kafka, respectively.
In working with queues, there are two processes: receiving a message and sending one.
Let's consider the receiving process. There are 2 options: for constant listening and for receiving a single message. For continuous listening to the queue, you must first describe a processor class that inherits from IMessageProcessor, which will be responsible for processing incoming messages. Next, it needs to be ‘bound’ to a specific queue, which is done through registration in IQueueReactorFactory with the queue identifier from the configuration:
"messaging": {
"ApplicationName": "MyApplication"
},
"rabbitmq": {
"queues": [
{
"id": "myQueue",
"queuename": "lalala",
...
}
]
},
Example of starting listening:
_queueReactorFactory.Register("myQueue");
var queueReactor = queueReactorFactory.CreateQueueReactor("myQueue");
queueReactor.StartProcessing();
Then, upon starting the service and calling the method to begin listening, all messages from the specified queue will go to the corresponding processor.
To receive a single message, in the factory interface IMessagingComponentFactory there is a method CreateMessageReceiver, which will create a receiver waiting for messages from the specified queue:
using (var receiver = _messagingComponentFactory.CreateMessageReceiver("myQueue"))
{
var message = receiver.Receive();
}
To send a message you need to use the same IMessagingComponentFactory and create a message sender:
using (var sender = _messagingComponentFactory.CreateMessageSender("myQueue"))
{
sender.SendMessage(new MyMessage { Value = ...});
}
For serialization and deserialization of messages, there are three ready-made options: plain text, XML, and JSON, but if necessary, custom implementations of the interfaces can be created. IMessageSerializer and IMessageDeserializer.
We have tried to retain the unique capabilities of each queue manager, for example, ViennaNET.Messaging.MQSeriesQueue allows not only sending text messages but also byte messages, and ViennaNET.Messaging.RabbitMQQueue supports routing and creating queues 'on the fly'. In our RabbitMQ adapter wrapper, we have also implemented something resembling RPC: we send a message and await a response from a special temporary queue that is created solely for a single response message.
Here .
ViennaNET.CallContext
We use queues not only for integration between different systems but also for communication between microservices of a single application, for instance, within a saga. This has led to the need to pass auxiliary data along with the message, such as the user's login, the request identifier for end-to-end logging, the source IP address, and authentication details. To implement the passing of this data, we developed a library ViennaNET.CallContext, which allows storing data from the incoming request to the service. The method by which the request was made, whether via a queue or through HTTP, does not matter. Then, before sending the outgoing request or message, the data is retrieved from the context and placed in the headers. This way, the next service receives the auxiliary data and can manage it correspondingly.
Thank you for your attention, we look forward to your comments and pull requests!
Source: habr.com
