CODEWITHSUNDEEP
.netsystem.reactive
Jeff
Jeff

Reputation: 36573

Async Queue Processing With Reactive Extensions

There are a couple of articles on this, and I have this working...but I want to know how to set a max number of Task threads for my Observable subscriptions at once.

I have the following to parallelize async saving of log entries:

private BlockingCollection<ILogEntry> logEntryQueue;

and

 logEntryQueue = new BlockingCollection<ILogEntry>();
 logEntryQueue.GetConsumingEnumerable().ToObservable(Scheduler.TaskPool).Subscribe(SaveLogEntry);

To schedule my saving...but how do I specify the max threads for the scheduler to use at once?

Upvotes: 7

Views: 4931

Answers (2)

Anderson Imes
Anderson Imes

Reputation: 25650

This is not a function of the Observable, but a function of the Scheduler. The Observable defines what and the scheduler defines where.

You'd need to pass in a custom scheduler. A simple way to do this would be to subclass TaskScheduler and override the "MaximumConcurrencyLevel" property.

http://msdn.microsoft.com/en-us/library/system.threading.tasks.taskscheduler.maximumconcurrencylevel.aspx

I actually found a sample of this on MSDN:

http://msdn.microsoft.com/en-us/library/ee789351.aspx

Edit: You asked about how to go from TaskScheduler to IScheduler. Another developer just gave me that little bit of info:

var ischedulerForRx = new TaskPoolScheduler
(
    new TaskFactory
    (
        //This is your custom scheduler
        new LimitedConcurrencyLevelTaskScheduler(1)
    )
);

Upvotes: 9

Richard Szalay
Richard Szalay

Reputation: 84734

If you create your "work" as IObservable<T> with deferred execution (ie. they want do anything until subscribed to), you can use the Merge overload that accepts a number of maximum concurrent subscriptions:

ISubject<QueueItem> synchronizedQueue = new Subject<QueueItem>().Synchronize();

queue
    .Select(item => StartWork(item))
    .Merge(maxConcurrent: 5) // C# 4 syntax for illustrative purposes
    .Subscribe();

// To enqueue:
synchronizedQueue.OnNext(new QueueItem());

Upvotes: 2

Related Questions