Your observable isn't hot. It's a cold observable with a shared source and it only makes the subsequent observers behave as if they got a hot observable. It's probably best described as a warm observable.
Let's look at an example:
var query = Observable.Range(0, 3).ObserveOn(Scheduler.Default).Publish().RefCount();
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("A"); });
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("B"); });
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("C"); });
Thread.Sleep(10000);
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("D"); });
Observable
.Range(0, 3)
.ObserveOn(Scheduler.Default)
.Publish()
.RefCount()
.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("E"); });
When I run this I get:
A
A
B
C
A
B
C
E
E
E
The "B" & "C" observers miss the first value of the sequence.
And, after the "A", "B", and "C" observers are done the sequence is finished, so "D" never gets a value. I've had to create a brand new observable to get the values "E" to display.
So, in your code you have a problem, if the first observer finishes one or more values before the second and third subscribe then those observers miss values. Is that what you want?
Nevertheless, your question asks about how to deal with disposable values returned from an observable. It's simple if you use Observable.Using.
Here's a similar situation to your code:
public static IObservable<IDisposable> ImagesInFolder(IScheduler scheduler)
{
return
Observable
.Range(0, 3)
.ObserveOn(Scheduler.Default)
.SelectMany(x =>
Observable
.Using(
() => Disposable.Create(() => Console.WriteLine("Disposed!")),
y => Observable.Return(y)))
.Publish()
.RefCount();
}
Now if I run this code:
var query = ImagesInFolder(Scheduler.Default);
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("A"); });
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("B"); });
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("C"); });
Thread.Sleep(10000);
query.Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("D"); });
I get this output:
A
B
C
Disposed!
A
B
C
Disposed!
A
B
C
Disposed!
Again "D" never produces any values - and it's possible for "B" & "C" to miss values, but this does show how to return an observable value that automatically gets disposed with the observer/s is/are finished.
Your code would look like this:
public static IObservable<System.Drawing.Bitmap> ImagesInFolder(string path, IScheduler scheduler)
{
return
Directory
.GetFiles(path, "*.bmp")
.ToObservable(scheduler)
.SelectMany(x =>
Observable
.Using(
() => new System.Drawing.Bitmap(x),
bm => Observable.Return(bm)))
.Publish()
.RefCount();
}
However, you're still in the land of possibly missing values.
Therefore you need to really do this:
public static IConnectableObservable<System.Drawing.Bitmap> ImagesInFolder(string path, IScheduler scheduler)
{
return
Directory
.GetFiles(path, "*.bmp")
.ToObservable(scheduler)
.SelectMany(x =>
Observable
.Using(
() => new System.Drawing.Bitmap(x),
bm => Observable.Return(bm)))
.Publish();
}
Then you call it like this:
public void Main()
{
var images = ImagesInFolder("c:\Users\VASIYA\Desktop\Sample Images", TaskPoolScheduler.Instance);
var process1 = images.Subscribe(SaveBwImages);
var process2 = images.Subscribe(SaveScaledImages);
var process3 = images.Select(Cats).Subscribe(SaveCatsImages);
images.Connect();
}
The other option is to drop the whole .Publish().RefCount() code and make sure you do it properly yourself when you subscribe.
Try this code:
void Main()
{
ImagesInFolder(Scheduler.Default)
.Publish(iif =>
Observable
.Merge(
iif.Select(x => { Thread.Sleep(1000); Console.WriteLine("A"); return "A"; }),
iif.Select(x => { Thread.Sleep(3000); Console.WriteLine("B"); return "B"; }),
iif.Select(x => { Thread.Sleep(2000); Console.WriteLine("C"); return "C"; })))
.Subscribe();
}
public static IObservable<IDisposable> ImagesInFolder(IScheduler scheduler)
{
return
Observable
.Range(0, 3)
.ObserveOn(Scheduler.Default)
.SelectMany(x =>
Observable
.Using(
() => Disposable.Create(() => Console.WriteLine("Disposed!")),
y => Observable.Return(y)));
}
I get this out:
A
B
C
Disposed!
A
B
C
Disposed!
A
B
C
Disposed!
Again, one Disposed! after each observer has run, but the problem now is that I changed the delay in the processing of each observer, but the code still output the is the order that the observers were added. The issue is that Rx runs each observer in sequence and each value produced is in sequence.
I expect that you thought you might get parallel processing using .Publish(). You don't.
The way to get this to run in parallel is to drop the .Publish() entirely.
Just do this kind of thing:
void Main()
{
ImagesInFolder(Scheduler.Default).Subscribe(x => { Thread.Sleep(1000); Console.WriteLine("A"); });
ImagesInFolder(Scheduler.Default).Subscribe(x => { Thread.Sleep(3000); Console.WriteLine("B"); });
ImagesInFolder(Scheduler.Default).Subscribe(x => { Thread.Sleep(2000); Console.WriteLine("C"); });
}
public static IObservable<IDisposable> ImagesInFolder(IScheduler scheduler)
{
return
Observable
.Range(0, 3)
.ObserveOn(Scheduler.Default)
.SelectMany(x =>
Observable
.Using(
() => Disposable.Create(() => Console.WriteLine("Disposed!")),
y => Observable.Return(y)));
}
I now get this:
A
Disposed!
C
Disposed!
A
Disposed!
B
Disposed!
A
Disposed!
C
Disposed!
C
Disposed!
B
Disposed!
B
Disposed!
The code now runs in parallel and finishes as fast as possible - and correctly disposes of the IDisposable when the subscription finishes. You just don't get the joy of sharing a single disposable resource with each observer, but you don't get all of the behavioural issues either.
