fix bug of RefCount

sandbox/ConsoleApp1/Program.cs

@@ -1,18 +1,31 @@
 using R3;
+//using System.Reactive.Linq;
 using System.Diagnostics;
 using System.Threading.Channels;
 using System.Xml.Serialization;
 
 
 
-string[] array = { "a", "b", "c" };
-array
- .ToObservable()
- .SubscribeAwait(static async (element, token) =>
- {
- Console.WriteLine(element);
- await Task.Yield();
- }, AwaitOperation.Sequential);
+var doScan = Observable.FromAsync(async (token) =>
+{
+ Console.WriteLine("scan start");
+ await Task.Delay(TimeSpan.FromSeconds(3));
+ Console.WriteLine("scan end");
+ return 5;
+});
+
+var doCalc = Observable.FromAsync(async (token) =>
+{
+ Console.WriteLine("calc start");
+ await Task.Delay(TimeSpan.FromSeconds(3), token);
+ Console.WriteLine("calc end");
+ return 10;
+});
+
+var countDown = Observable.Interval(TimeSpan.FromMilliseconds(300)).Index().Select(v => v > 9 ? 9 : v);
+
+var work = doScan.Select(_ => doCalc).Switch().Replay(1).RefCount();
+countDown.TakeUntil(work.LastAsync()).Concat(work.TakeLast(1)).Subscribe(v => Console.WriteLine($"progress: {v}"));
 
 
 Console.ReadLine();

src/R3/Operators/RefCount.cs

@@ -18,8 +18,12 @@ protected override IDisposable SubscribeCore(Observer<T> observer)
  {
  lock (gate)
  {
- var subcription = source.Subscribe(new _RefCount(this, observer));
- if (++refCount == 1)
+ // incr refCount before Subscribe(completed source decrement refCount in Subscribe)
+ ++refCount;
+ bool needConnect = refCount == 1;
+ var coObserver = new _RefCount(this, observer);
+ var subcription = source.Subscribe(coObserver);
+ if (needConnect && !coObserver.IsDisposed)
  {
  connection = source.Connect();
  }