4.x: Fix ActivePlan crashing with self-joins (#514)

Rx.NET/Source/src/System.Reactive/Joins/ActivePlan.cs

@@ -14,7 +14,10 @@ namespace System.Reactive.Joins
 
         protected void AddJoinObserver(IJoinObserver joinObserver)
         {
-            joinObservers.Add(joinObserver, joinObserver);
+            if (!joinObservers.ContainsKey(joinObserver))
+            {
+                joinObservers.Add(joinObserver, joinObserver);
+            }
         }
 
         protected void Dequeue()
@@ -1432,4 +1435,4 @@ namespace System.Reactive.Joins
             }
         }
     }
-}
+}

Rx.NET/Source/tests/Tests.System.Reactive/Tests/Linq/Observable/WhenTest.cs

@@ -223,5 +223,15 @@ namespace ReactiveTests.Tests
             yield break;
         }
 
+        [Fact]
+        public void SameSource()
+        {
+            var source = Observable.Range(1, 5);
+
+            var list = Observable.When(source.And(source).Then((a, b) => a + b))
+                .ToList().First();
+
+            Assert.Equal(new List<int>() { 2, 4, 6, 8, 10 }, list);
+        }
     }
 }