rebus-org/rebus.mongodb

:bus: MongoDB persistence implementations for Rebus

Rebus.MongoDb

Provides MongoDB persistence implementations for Rebus for

• sagas
• subscriptions
• timeouts

Provides transport MongoDb implementations for Rebus

Unit Tests

To run unit test please provide a mongo instance to run test and set the connection string ino REBUS_MONGODB environment variable.

Local nuget pack

To manually create a nuget package specifying manually the version you can use this commandline.

dotnet pack Rebus.MongoDb\Rebus.MongoDb.csproj -o c:\target_directory -c release /p:PackageVersion=6.0.1011 /p:AssemblyVersion=6.0.0 /p:FileVersion=6.0.0 /p:InformationalVersion=6.0.0-__SHA_OF_COMMIT__

//then you need to push
dotnet nuget push .\Rebus.MongoDb.6.0.1011.nupkg -s https://pkgs.dev.azure.com/xxxxx/_packaging/__packageName__/nuget/v3/index.json --api-key az

---

rebus.mongodb's Issues

Properties of ISagaData documents are locked to pascal casing

Due to using strings instead of properties in a BsonDocument filter, compatibility with the CamelCaseElementNameConvention is lost.

Since I rely on camel casing for all other MongoDb actions (conventions applied globally), the only way I'm able to get around this currently is applying a custom class map for each SagaData implementation, which is obviously not ideal :(

Example:
https://github.com/rebus-org/Rebus.MongoDb/blob/master/Rebus.MongoDb/MongoDb/Sagas/MongoDbSagaStorage.cs#L103

var criteria = Builders<BsonDocument>.Filter.And(
    Builders<BsonDocument>.Filter.Eq("_id", sagaData.Id),
    Builders<BsonDocument>.Filter.Eq(nameof(ISagaData.Revision), sagaData.Revision)
);

Using Filter.Where instead of Filter.Eq would get around this, but I think you'd need to be strongly typed instead of using BsonDocument in order to do that.

This would also be an issue in the BSON validation:

https://github.com/rebus-org/Rebus.MongoDb/blob/master/Rebus.MongoDb/MongoDb/Sagas/MongoDbSagaStorage.cs#L247-L250

if (!target.Contains("Revision"))
{
    throw GetException("could not find 'Revision' in the serialied BSON document");
}

^ this one might be fixed just by doing something like this instead.

if (!target.Any(x => x.Name.Equals("Revision", StringComparison.InvariantCultureIgnoreCase)))

Issue with index creation

In my company I don't have permissiton to create an index. Can we make it optional?

Please provide some example of how to use this.

Seems like a great project, but the learning curve is steep.

I have a difficult time figuring out what this does and how to use it. Documentation is none-existing..

MongoDb Transport

Does Rebus.MongoDb support MongoDB for Transport? Or only storage Subscriptions and Sagas?

If no, should I implement my own (rebus-org/Rebus#840) or are there very good reasons not to?

Pub/Sub

Hello, I am using mongodb as the Rebus transport. When I use pub/sub, two consumers subscribe to the same topic, and then when I use the producer to send an event, it will prompt id conflict,It generates two identical messages, including the same id.

System.AggregateException: Messages sent were not unique (Cannot send message with ID 4ab31b8c-da85-4a6f-ac24-fb482c954f34 because a message already exists with that ID)
 ---> Rebus.MongoDb.Transport.Exceptions.UniqueMessageIdViolationException: Cannot send message with ID 4ab31b8c-da85-4a6f-ac24-fb482c954f34 because a message already exists with that ID
   --- End of inner exception stack trace ---

this is my producer code:

// builder
builder.Services.AddRebus(configure => configure
     .Logging(l => l.ColoredConsole(Rebus.Logging.LogLevel.Info)).Transport(t => t.UseMongoDb(new MongoDbTransportOptions(mongoUrl), "producer"))
    .Subscriptions(s => s.StoreInMongoDb(mongoDatebase, "testMessageEvent", true))
    .Serialization(s => s
        .UseCustomMessageTypeNames()
        .AddWithShortName(typeof(TestMessageEvent)))
    .Routing(r => r
        .TypeBased()
        .MapAssemblyOf<OrderCreated>("testMessage")
        .Map<TestMessageEvent>("testMessage"))
);

// app
app.Services.UseRebus(r => r.Advanced.Topics.Subscribe("producerTest"));


// send message
 await _bus.Advanced.Topics.Publish("producerTest", new TestMessageEvent(Guid.NewGuid().ToString(), "test"));

this is my consumer code:

// builder
builder.Services.AddRebus(configure => configure
     .Logging(l => l.ColoredConsole(Rebus.Logging.LogLevel.Info)).Transport(t => t.UseMongoDb(new MongoDbTransportOptions(mongoUrl), "consumer"))
    .Subscriptions(s => s.StoreInMongoDb(mongoDatebase, "testMessageEvent", true))
    .Serialization(s => s
        .UseCustomMessageTypeNames()
        .AddWithShortName(typeof(TestMessageEvent)))
    .Routing(r => r
        .TypeBased()
        .MapAssemblyOf<OrderCreated>("testMessage")
        .Map<TestMessageEvent>("testMessage"))
);

// app
app.Services.UseRebus(r => r.Advanced.Topics.Subscribe("producerTest"));

and this my mongodb data from testMessageEvent:

{
    "_id": "producerTest",
    "addresses": [
        "producer",
        "consumer"
    ]
}

in Rebus.MongoDb.Transport.MongoDbMessageProducer

I don't know Rebus well, so I don't know why it's doing this here, is there any reason?

Any plans to upgrade to MongoDb.Driver 2.7.3?

Just wondering! 😄
https://github.com/mongodb/mongo-csharp-driver

Implement data bus storage

use GridFS to provide an implementation of IDataBusStorage

Improper blocking of async functions

I can say from a number of very painful experiences that Microsoft's TAP async/await model handles synchronous blocking code extremely poorly. Annoyingly, it tends to show up by deadlocking, or at least extremely poor performance, in completely unrelated code.

This code uses a spinlock, which will block the current TAP thread.

Rebus.MongoDb/Rebus.MongoDb/MongoDb/Transport/MongoDbTransport.cs

Line 214 in 15ab99a

SpinWait.SpinUntil(() => _concurrentSend < 10);

The correct fix would be to use a SemaphoreSlim, which allows you to await access without a spinlock. As SemaphoreSlim already enables a maximum concurrent access count, the code would likely be both more correct, and simpler to read.

Exception trying to use Mongo Timeout Manager

When trying to use MongoDB for timeouts, Rebus is throwing an exception resolving the bus with the inner exceptions:
ResolutionException: Could not resolve Rebus.Timeouts.ITimeoutManager with decorator depth 0 - registrations: Rebus.Injection.Injectionist+Handler
and
ArgumentException: Value cannot be empty. (Parameter 'name')

New MongoDB transport

Port code from

https://github.com/rebus-org/MongolianBarbecue

and re-implement the transport.

Remember to derive the transport from AbstractRebusTransport and then pass the contract tests.

Figure out correct saga data field names

Because MongoDB's BSON serializer is global, there's a chance (risk!) that any conventions applied at a global level will apply to how sagas are serialized. Also, this might actually be what the user wants 🙂

BUT it poses a problem e.g. when using camel-cased field names in BSON, because Rebus will e.g. be querying for a field named CorrelationId, which is the name in the C# class, but in BSON it's represented as correlationId.

By doing an initial dummy-serialization of an empty instance of the saga data type, it might be possible to deduce the actual resulting field names corresponding to the saga's fields (at least if it's only the casing that's changed).

Collection name is not respected

The collection "messages" is created and used. Not the one I passed.

MarkAsComplete throws an Error if GUID uses BsonType.String

In our application uses a GuidSerializer with String representation.

BsonSerializer.RegisterSerializer(typeof(Guid), new GuidSerializer(BsonType.String));

In this case the filter does not work correctly and no data records are found and an exception is thrown.

After a littlebit of debugging the MongoDbSagaStorage, I found a working solution.

Instead of

var result = await collection.DeleteManyAsync(new BsonDocument("_id", sagaData.Id)).ConfigureAwait(false);

I used

var criteria = Builders<BsonDocument>.Filter.Eq("_id", Data.Id);
var result = await collection.DeleteManyAsync(criteria).ConfigureAwait(false);

and the saga was correctly deleted.

Will mongodb transport be released?

Any release date for mongodb transport?

Additional Outbox providers

Hey @mookid8000 just sussing out if you're open to PRs for Postgres and MongoDB outbox implementations? Anything already in the works in these areas?

Index creation on saga correlation properties breaks my app

I've already a code that create an index on correlation properties of a saga. Can you add a check that skip creation if an index with same properties already exists? Now i get this error because my index are created before your and/or already exists.
Command createIndexes failed: Index: { v: 2, unique: true, key: { WorkflowId: 1 }, name: "WorkflowId_1", ns: "devops-test-core.WorkflowSagaData" } already exists with different options: { v: 2, key: { WorkflowId: 1 }, name: "A-WorkflowId", background: true, ns: "devops-test-core.WorkflowSagaData" }.

Thanks