Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,20 @@ namespace RabbitMQ.Client.Impl
internal sealed class AsyncConsumerWorkService : ConsumerWorkService
{
private readonly ConcurrentDictionary<IModel, WorkPool> _workPools = new ConcurrentDictionary<IModel, WorkPool>();
private readonly Func<IModel, WorkPool> _startNewWorkPoolFunc = model => StartNewWorkPool(model);

public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
{
_workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);
/*
* rabbitmq/rabbitmq-dotnet-client#841
* https://docs.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentdictionary-2.getoradd
* Note that the value delegate is not atomic but the Schedule method will not be called concurrently.
*/
WorkPool workPool = _workPools.GetOrAdd(model, _startNewWorkPoolFunc);
workPool.Enqueue(work);
}

private WorkPool StartNewWorkPool(IModel model)
private static WorkPool StartNewWorkPool(IModel model)
{
var newWorkPool = new WorkPool(model as ModelBase);
newWorkPool.Start();
Expand Down
13 changes: 10 additions & 3 deletions projects/RabbitMQ.Client/client/impl/ConsumerWorkService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -5,16 +5,23 @@

namespace RabbitMQ.Client.Impl
{
class ConsumerWorkService
internal class ConsumerWorkService
{
private readonly ConcurrentDictionary<IModel, WorkPool> _workPools = new ConcurrentDictionary<IModel, WorkPool>();
private readonly Func<IModel, WorkPool> _startNewWorkPoolFunc = model => StartNewWorkPool(model);

public void AddWork(IModel model, Action fn)
{
_workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(fn);
/*
* rabbitmq/rabbitmq-dotnet-client#841
* https://docs.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentdictionary-2.getoradd
* Note that the value delegate is not atomic but the AddWork method will not be called concurrently.
*/
WorkPool workPool = _workPools.GetOrAdd(model, _startNewWorkPoolFunc);
workPool.Enqueue(fn);
}

private WorkPool StartNewWorkPool(IModel model)
private static WorkPool StartNewWorkPool(IModel model)
{
var newWorkPool = new WorkPool();
newWorkPool.Start();
Expand Down