m-wilczynski / resonance Goto Github PK

It would be useful to have some sort of IOutboxLogger for implementing integrations with logging libraries for Resonance.Outbox internals (ie. TransactionalOutbox and OutboxForwader functionalities).

Targeted libraries support would be:

• log4net
• Serilog
• NLog

Currently only wrapping around execution of whole mechanism (and IMessageForwarder implementations) is possible.

It seems that .NET Framework based applications would not resolve particular version on System.Data.SqlClient that is required by Dapper in Resonance.Outbox.Storage.SqlServer.

Needs:

• reproduction in .NET Framework 4.7.x
• enforcing dependency on System.Data.SqlClient for all platforms (now Dapper pulls it for netstandard2.0 only)
• some kind of test for both .NET Framework 4.7.x and .NET Core 2.x/3.x?

Add documentation on how to use Outbox, ie:

• send events to ITransactionalOutbox
• push events out of outbox with IOutboxForwarder with scheduling, using for ex. Hangfire for scheduling and MassTransit for messaging

GetMessagesAsMarkedSent for SQL Server:

works only because IX for SuccessfulForwardDateUtc is in place::

Fix subquery (aliased as msg above), so that it selects batch with TOP X rows only from resultset where SuccessfulForwardDateUtc IS NULL.

Currently Resonance.Outbox.Storage.SqlServer implements only ReadThenMarkAsRead strategy.
There is another ReadMode - ReadThenDelete that is currently not implemented for SQL Server storage.

Idea is to DELETE records per batch (or all, depending on ForwardingOptions) and return them to OutboxForwarder with OUTPUT clause.

Currently there are only structures in DB and empty methods for logging succesful or failed ForwardUnsentMessagesFromOutbox performed by OutboxForwarder.
Introduce contract for logging attempted forwards in IMessageRepository and implement it in current SQL Server integration.

Only SQL Server integration is available now via Resonance.Outbox.Storage.SqlServer.
At least one more RDBMS should be available for users to choose, especially if targetting platforms other than Windows.
PostgreSQL looks like a good direction to go.

Point of interest: Npgsql

If chosen to use ReadMode.ReadThenMarkAsRead strategy, messages table can get out of control pretty quickly.
Some sort of retention policy should be introduced + ways of cleaning sent messages older than X days (hours?).
Should probably be analyzed together with #5.

When #3 is completed, it would be useful to have ability to automatically clean up forwarding log with entries older than X days (hours?).
Provided we have forward from outbox per 3 seconds, it makes (60/3)x60x24 = 28,800/day produced by OutboxForwarder.
If not introduced, log would get quite fat pretty quickly.

Problem

Currently both Inbound and Outbound implementation assumes there is:

• IDbConnection
• either explicit IDbTransaction or ambient TransactionScope

With such constraints there is no way to introduce other backing storage with NoSQL databases such as MongoDB, Neo4j, Redis etc..

Possible solution

Assuming we want to keep the flow of current Resonance.Outbox API, there should still be a way to:

using (var anyConnection = OpenConnection())
{
    using (var anyTransaction = BeginTransaction())
    {
        anyConnection.DoUserCode();
        outobox.Send(myEvent);
        anyTransaction.Commit();
    }
}

There should be:

• generic type param for transaction, ie. ITransactionalOutbox<TTransaction> where TTransaction would be any implementation of database transaction, such as:
  • IDbTransaction for SQL Databases
  • AsyncTransaction for Neo4j
  • IClientSession for MongoDb
  • ITransaction for Redis
  • etc.
• common contract to start connection and transaction and to commit/rollback transaction in IMessageRepository

Such change would change current implementation of these APIs:

ITransactionalOutbox:

public interface ITransactionalOutbox<TTransactionSource>
{
    Task Send<TMessage>(TMessage message, TTransactionSource transaction = null, DateTime? sendTime = null);
}

TransactionalOutbox:

public class SqlTransactionalOutbox : ITransactionalOutbox<IDbTransaction>

IMessageRepository:

public interface IMessageRepository<TTransactionSource>
{
    Task SaveMessage(SerializedMessage serializedMessage, TTransactionSource transaction = null);
    Task<ICollection<SerializedMessage>> GetMessagesAsMarkedSent(TTransactionSource transaction, uint? howManyMessages = null);
    Task<ICollection<SerializedMessage>> GetMessagesAsRemoved(TTransactionSource transaction, uint? howManyMessages = null);
}

SqlServerMessageRepository:

public class SqlServerMessageRepository : IMessageRepository<IDbTransaction>
{
    //(...)
}

and would also introduce extension of contract for IMessageRepository for managing transaction for Outbound part:

public interface IReadMessageRepository<TTransactionSource> : IMessageRepository<TTransactionSource>
{
    Task<TTransactionSource> BeginTransaction();
    Task CommitTransaction();
    Task RollbackTransaction();
}

Definition of done

1. Implementation of example NoSQL integration; example: MongoDB, Neo4j.
2. Previous SQL solution still works.

Points of further improvement

Perhaps there also should be a way to introduce transaction options as second generic type param, ie:

public interface IReadMessageRepository<TTransactionSource, TTransactionOptions> : IMessageRepository<TTransactionSource>
{
    Task<TTransactionSource> BeginTransaction(TTransactionOptions transactionOptions);
    //(...)
}