/**
 * Copyright 2013 Netflix, Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package rx.subscriptions;

import static rx.subscriptions.Subscriptions.empty;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import rx.Subscription;

/**
 * Represents a subscription whose underlying subscription can be swapped for another subscription
 * which causes the previous underlying subscription to be unsubscribed.
 * 
 * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.disposables.serialdisposable(v=vs.103).aspx">Rx.Net equivalent SerialDisposable</a>
 */
public class SerialSubscription implements Subscription {
	private final AtomicBoolean unsubscribed = new AtomicBoolean();
	private final AtomicReference<Subscription> reference = new AtomicReference<Subscription>(empty());

	@Override
	public void unsubscribe() {
		if (unsubscribed.compareAndSet(false, true)) {
			reference.getAndSet(empty()).unsubscribe();
		}
	}

	public void setSubscription(final Subscription subscription) {
		if (unsubscribed.get()) {
			subscription.unsubscribe();
		} else {
			reference.getAndSet(subscription == null ? empty() : subscription).unsubscribe();
		}
	}

	public Subscription getSubscription() {
		return reference.get();
	}
}

@@ -771,16 +771,163 @@
 ference;
+%0Aimport java.util.concurrent.locks.Lock;%0Aimport java.util.concurrent.locks.ReadWriteLock;%0Aimport java.util.concurrent.locks.ReentrantReadWriteLock;
 %0A%0Aimport
@@ -1344,17 +1344,20 @@
 ption %7B%0A
-%09
+    
 private 
@@ -1412,17 +1412,20 @@
 lean();%0A
-%09
+    
 private 
@@ -1516,18 +1516,252 @@
 pty());%0A
-%0A%09
+    private final Lock read;%0A    private final Lock write;%0A%0A    public SerialSubscription() %7B%0A        final ReadWriteLock lock = new ReentrantReadWriteLock();%0A        read = lock.readLock();%0A        write = lock.writeLock();%0A    %7D%0A%0A    
 @Overrid
@@ -1762,17 +1762,20 @@
 verride%0A
-%09
+    
 public v
@@ -1790,26 +1790,72 @@
 bscribe() %7B%0A
-%09%09
+        write.lock();%0A        try %7B%0A            
 if (unsubscr
@@ -1889,19 +1889,32 @@
 rue)) %7B%0A
-%09%09%09
+                
 referenc
@@ -1945,33 +1945,107 @@
 ubscribe();%0A
-%09%09%7D%0A%09%7D%0A%0A%09
+            %7D%0A        %7D finally %7B%0A            write.unlock();%0A        %7D%0A    %7D%0A%0A    
 public void 
@@ -2095,18 +2095,63 @@
 tion) %7B%0A
-%09%09
+        read.lock();%0A        try %7B%0A            
 if (unsu
@@ -2168,19 +2168,32 @@
 et()) %7B%0A
-%09%09%09
+                
 subscrip
@@ -2216,113 +2216,176 @@
 ();%0A
-%09%09%7D else %7B%0A%09%09%09reference.getAndSet(subscription == null ? empty() : subscription).unsubscribe();%0A%09%09%7D%0A%09%7D%0A%0A%09
+            %7D else %7B%0A                reference.getAndSet(subscription).unsubscribe();%0A            %7D%0A        %7D finally %7B%0A            read.unlock();%0A        %7D%0A    %7D%0A%0A    
 publ
@@ -2424,10 +2424,16 @@
 ) %7B%0A
-%09%09
+        
 retu
@@ -2456,9 +2456,12 @@
 ();%0A
-%09
+    
 %7D%0A%7D%0A

Make unsubscribe + reference modification an atomic operation