CODEWITHSUNDEEP
c#.netsystem.reactive
Kevin
Kevin

Reputation: 4747

Can I use reactive extensions to queue calls to a windows service?

I have a window service that is calling a non-thread safe api (ghostscript) to convert pdf's to images so I need to ensure the method call is called one at a time but queue all the calls which can be coming from various app instances.

app -> windows service -> ghostscript

Upvotes: 2

Views: 1119

Answers (2)

Asti
Asti

Reputation: 12667

Rx is a bit of overkill for the job. Consider TPL if you have complex queuing/wait/batching needs.

If your requirements are simple, you can go for the simple Task.Start(...), Task.ContinueWith(...) to create a thread-safe processing queue of sorts.

It could be as simple as:

public class TaskQueue
{
    protected Task Pending;
    public bool Ready { get { return Pending == null || Pending.IsCompleted || Pending.IsCanceled || Pending.IsFaulted; } }

    public Task Enqueue(Action work)
    {
        lock (this)
            return Pending = Ready ? Task.Factory.StartNew(work) : Pending.ContinueWith(_ => work());
    }
}

The easiest way to test this is to use TPL, so: 

var tasks = new TaskQueue();
Func<int, Action> queue = i => () => tasks.Enqueue(
() => { Thread.Sleep(1000); Console.WriteLine("{0}: {1}", DateTime.Now, i); });

Parallel.Invoke
(
    queue(1),
    queue(2),
    queue(3),
    queue(4)
);

Results:

3/10/2012 5:59:02 PM: 1
3/10/2012 5:59:03 PM: 2
3/10/2012 5:59:04 PM: 3
3/10/2012 5:59:05 PM: 4

Upvotes: 1

Ana Betts
Ana Betts

Reputation: 74654

Here you go:

var eventLoop = new EventLoopScheduler();

IObservable<Unit> QueueAnItem(string input) 
{
    return Observable.Start(() => CallGhostScriptAndWaitForItToFinish(input), eventLoop);
}

QueueAnItem("Foo").Subscribe(
    x => Console.WriteLine("It Finished!"), 
    ex => Console.WriteLine("Something Bad Happened: {0}", ex.ToString());

Upvotes: 3

Related Questions