fix: getSubscription should not expose internal sentinel subscription

Author: jloisel

rxjava-core/src/main/java/rx/subscriptions/SerialSubscription.java

@@ -29,7 +29,8 @@
  */
 public class SerialSubscription implements Subscription {
     private final AtomicReference<Subscription> reference = new AtomicReference<Subscription>(empty());
-
+    private volatile Subscription unsubscribe;
+    
     private static final Subscription UNSUBSCRIBED = new Subscription() {
         @Override
         public void unsubscribe() {
@@ -50,8 +51,16 @@ public void setSubscription(final Subscription subscription) {
             }
             if (reference.compareAndSet(current, subscription)) {
                 current.unsubscribe();
+                if(subscription == UNSUBSCRIBED) {
+                	unsubscribe = current;
+                }
                 break;
             }
         } while (true);
     }
+    
+    public Subscription getSubscription() {
+    	Subscription subscription = reference.get();
+		return subscription == UNSUBSCRIBED ? unsubscribe : subscription;
+    }
 }

rxjava-core/src/test/java/rx/subscriptions/SerialSubscriptionTests.java

@@ -15,6 +15,7 @@
  */
 package rx.subscriptions;
 
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -46,6 +47,25 @@ public void unsubscribingWithoutUnderlyingDoesNothing() {
         serialSubscription.unsubscribe();
     }
 
+    @Test
+    public void getSubscriptionShouldReturnSubscriptionAfterUnsubscribe() {
+    	 final Subscription underlying = mock(Subscription.class);
+         serialSubscription.setSubscription(underlying);
+         serialSubscription.unsubscribe();
+         assertSame(underlying, serialSubscription.getSubscription());
+    }
+    
+    @Test
+    public void getSubscriptionShouldReturnSetSubscription() {
+    	 final Subscription underlying = mock(Subscription.class);
+         serialSubscription.setSubscription(underlying);
+         assertSame(underlying, serialSubscription.getSubscription());
+         
+         final Subscription another = mock(Subscription.class);
+         serialSubscription.setSubscription(another);
+         assertSame(another, serialSubscription.getSubscription());
+    }
+    
     @Test
     public void unsubscribingTwiceDoesUnsubscribeOnce() {
     	 Subscription underlying = mock(Subscription.class);