Merge pull request #32 from naxxster/fix-switch-subscription

Fix switch operator subscription to unsubscribe from inner subscription

rx.lua

@@ -1459,7 +1459,7 @@ end
 -- @returns {Observable}
 function Observable:switch()
   return Observable.create(function(observer)
-    local subscription
+    local innerSubscription
 
     local function onNext(...)
       return observer:onNext(...)
@@ -1474,14 +1474,23 @@ function Observable:switch()
     end
 
     local function switch(source)
-      if subscription then
-        subscription:unsubscribe()
+      if innerSubscription then
+        innerSubscription:unsubscribe()
       end
 
-      subscription = source:subscribe(onNext, onError, nil)
+      innerSubscription = source:subscribe(onNext, onError, nil)
     end
 
-    return self:subscribe(switch, onError, onCompleted)
+    local subscription = self:subscribe(switch, onError, onCompleted)
+    return Subscription.create(function()
+      if innerSubscription then
+        innerSubscription:unsubscribe()
+      end
+
+      if subscription then
+        subscription:unsubscribe()
+      end
+    end)
   end)
 end
 

src/operators/switch.lua

@@ -5,7 +5,7 @@ local Observable = require 'observable'
 -- @returns {Observable}
 function Observable:switch()
   return Observable.create(function(observer)
-    local subscription
+    local innerSubscription
 
     local function onNext(...)
       return observer:onNext(...)
@@ -20,13 +20,22 @@ function Observable:switch()
     end
 
     local function switch(source)
-      if subscription then
-        subscription:unsubscribe()
+      if innerSubscription then
+        innerSubscription:unsubscribe()
       end
 
-      subscription = source:subscribe(onNext, onError, nil)
+      innerSubscription = source:subscribe(onNext, onError, nil)
     end
 
-    return self:subscribe(switch, onError, onCompleted)
+    local subscription = self:subscribe(switch, onError, onCompleted)
+    return Subscription.create(function()
+      if innerSubscription then
+        innerSubscription:unsubscribe()
+      end
+
+      if subscription then
+        subscription:unsubscribe()
+      end
+    end)
   end)
 end

tests/switch.lua

@@ -36,4 +36,17 @@ describe('switch', function()
     expect(#onError).to.equal(0)
     expect(#onCompleted).to.equal(1)
   end)
+
+  it('should unsubscribe from inner subscription too', function()
+    local unsubscribeA = spy()
+    local observableA = Rx.Observable.create(function(observer)
+      return Rx.Subscription.create(unsubscribeA)
+    end)
+
+    local subject = Rx.Subject.create()
+    local subscription = subject:switch():subscribe()
+    subject:onNext(observableA)
+    subscription:unsubscribe()
+    expect(#unsubscribeA).to.equal(1)
+  end)
 end)