1.x: compact MultipleAssignment- and Serial-Subscriptions

src/main/java/rx/Scheduler.java

@@ -18,11 +18,10 @@
 import java.util.concurrent.TimeUnit;
 
 import rx.annotations.Experimental;
-import rx.functions.Action0;
-import rx.functions.Func1;
+import rx.functions.*;
 import rx.internal.schedulers.SchedulerWhen;
+import rx.internal.subscriptions.SequentialSubscription;
 import rx.schedulers.Schedulers;
-import rx.subscriptions.MultipleAssignmentSubscription;
 
 /**
  * A {@code Scheduler} is an object that schedules units of work. You can find common implementations of this
@@ -126,7 +125,9 @@ public Subscription schedulePeriodically(final Action0 action, long initialDelay
             final long firstNowNanos = TimeUnit.MILLISECONDS.toNanos(now());
             final long firstStartInNanos = firstNowNanos + unit.toNanos(initialDelay);
 
-            final MultipleAssignmentSubscription mas = new MultipleAssignmentSubscription();
+            final SequentialSubscription first = new SequentialSubscription();
+            final SequentialSubscription mas = new SequentialSubscription(first);
+
             final Action0 recursiveAction = new Action0() {
                 long count;
                 long lastNowNanos = firstNowNanos;
@@ -155,14 +156,11 @@ public void call() {
                         lastNowNanos = nowNanos;
 
                         long delay = nextTick - nowNanos;
-                        mas.set(schedule(this, delay, TimeUnit.NANOSECONDS));
+                        mas.replace(schedule(this, delay, TimeUnit.NANOSECONDS));
                     }
                 }
             };
-            MultipleAssignmentSubscription s = new MultipleAssignmentSubscription();
-            // Should call `mas.set` before `schedule`, or the new Subscription may replace the old one.
-            mas.set(s);
-            s.set(schedule(recursiveAction, initialDelay, unit));
+            first.replace(schedule(recursiveAction, initialDelay, unit));
             return mas;
         }
 

src/main/java/rx/internal/subscriptions/SequentialSubscription.java

@@ -0,0 +1,189 @@
+/**
+ * Copyright 2014 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.internal.subscriptions;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import rx.Subscription;
+import rx.subscriptions.Subscriptions;
+
+/**
+ * A container of a Subscription that supports operations of SerialSubscription 
+ * and MultipleAssignmentSubscription via methods (update, replace) and extends
+ * AtomicReference to reduce allocation count (beware the API leak of AtomicReference!).
+ * @since 1.1.9
+ */
+public final class SequentialSubscription extends AtomicReference<Subscription> implements Subscription {
+
+    /** */
+    private static final long serialVersionUID = 995205034283130269L;
+
+    /**
+     * Create an empty SequentialSubscription.
+     */
+    public SequentialSubscription() {
+
+    }
+
+    /**
+     * Create a SequentialSubscription with the given initial Subscription.
+     * @param initial the initial Subscription, may be null
+     */
+    public SequentialSubscription(Subscription initial) {
+        lazySet(initial);
+    }
+
+    /**
+     * Returns the current contained Subscription (may be null).
+     * <p>(Remark: named as such because get() is final).
+     * @return the current contained Subscription (may be null)
+     */
+    public Subscription current() {
+        Subscription current = super.get();
+        if (current == Unsubscribed.INSTANCE) {
+            return Subscriptions.unsubscribed();
+        }
+        return current;
+    }
+
+    /**
+     * Atomically sets the contained Subscription to the provided next value and unsubscribes
+     * the previous value or unsubscribes the next value if this container is unsubscribed.
+     * <p>(Remark: named as such because set() is final).
+     * @param next the next Subscription to contain, may be null
+     * @return true if the update succeded, false if the container was unsubscribed
+     */
+    public boolean update(Subscription next) {
+        for (;;) {
+            Subscription current = get();
+
+            if (current == Unsubscribed.INSTANCE) {
+                if (next != null) {
+                    next.unsubscribe();
+                }
+                return false;
+            }
+
+            if (compareAndSet(current, next)) {
+                if (current != null) {
+                    current.unsubscribe();
+                }
+                return true;
+            }
+        }
+    }
+
+    /**
+     * Atomically replaces the contained Subscription to the provided next value but
+     * does not unsubscribe the previous value or unsubscribes the next value if this 
+     * container is unsubscribed.
+     * @param next the next Subscription to contain, may be null
+     * @return true if the update succeded, false if the container was unsubscribed
+     */
+    public boolean replace(Subscription next) {
+        for (;;) {
+            Subscription current = get();
+
+            if (current == Unsubscribed.INSTANCE) {
+                if (next != null) {
+                    next.unsubscribe();
+                }
+                return false;
+            }
+
+            if (compareAndSet(current, next)) {
+                return true;
+            }
+        }
+    }
+
+    /**
+     * Atomically tries to set the contained Subscription to the provided next value and unsubscribes
+     * the previous value or unsubscribes the next value if this container is unsubscribed.
+     * <p>
+     * Unlike {@link #update(Subscription)}, this doesn't retry if the replace failed
+     * because a concurrent operation changed the underlying contained object.
+     * @param next the next Subscription to contain, may be null
+     * @return true if the update succeded, false if the container was unsubscribed
+     */
+    public boolean updateWeak(Subscription next) {
+        Subscription current = get();
+        if (current == Unsubscribed.INSTANCE) {
+            if (next != null) {
+                next.unsubscribe();
+            }
+            return false;
+        }
+        if (compareAndSet(current, next)) {
+            return true;
+        }
+
+        current = get();
+
+        if (next != null) {
+            next.unsubscribe();
+        }
+        return current == Unsubscribed.INSTANCE;
+    }
+
+    /**
+     * Atomically tries to replace the contained Subscription to the provided next value but
+     * does not unsubscribe the previous value or unsubscribes the next value if this container 
+     * is unsubscribed.
+     * <p>
+     * Unlike {@link #replace(Subscription)}, this doesn't retry if the replace failed
+     * because a concurrent operation changed the underlying contained object.
+     * @param next the next Subscription to contain, may be null
+     * @return true if the update succeded, false if the container was unsubscribed
+     */
+    public boolean replaceWeak(Subscription next) {
+        Subscription current = get();
+        if (current == Unsubscribed.INSTANCE) {
+            if (next != null) {
+                next.unsubscribe();
+            }
+            return false;
+        }
+        if (compareAndSet(current, next)) {
+            return true;
+        }
+
+        current = get();
+        if (current == Unsubscribed.INSTANCE) {
+            if (next != null) {
+                next.unsubscribe();
+            }
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public void unsubscribe() {
+        Subscription current = get();
+        if (current != Unsubscribed.INSTANCE) {
+            current = getAndSet(Unsubscribed.INSTANCE);
+            if (current != null && current != Unsubscribed.INSTANCE) {
+                current.unsubscribe();
+            }
+        }
+    }
+
+    @Override
+    public boolean isUnsubscribed() {
+        return get() == Unsubscribed.INSTANCE;
+    }
+}

src/main/java/rx/internal/subscriptions/Unsubscribed.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright 2016 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.internal.subscriptions;
+
+import rx.Subscription;
+
+/**
+ * Represents an unsubscribed Subscription via a singleton; don't leak it!
+ */
+public enum Unsubscribed implements Subscription {
+    INSTANCE;
+
+    @Override
+    public boolean isUnsubscribed() {
+        return true;
+    }
+
+    @Override
+    public void unsubscribe() {
+        // deliberately ignored
+    }
+}

src/main/java/rx/subscriptions/MultipleAssignmentSubscription.java

@@ -15,56 +15,25 @@
  */
 package rx.subscriptions;
 
-import java.util.concurrent.atomic.AtomicReference;
-
-import rx.Observable;
-import rx.Subscription;
+import rx.*;
+import rx.internal.subscriptions.SequentialSubscription;
 
 /**
  * Subscription that can be checked for status such as in a loop inside an {@link Observable} to exit the loop
  * if unsubscribed.
  */
 public final class MultipleAssignmentSubscription implements Subscription {
 
-    final AtomicReference<State> state = new AtomicReference<State>(new State(false, Subscriptions.empty()));
+    final SequentialSubscription state = new SequentialSubscription();
 
-    static final class State {
-        final boolean isUnsubscribed;
-        final Subscription subscription;
-
-        State(boolean u, Subscription s) {
-            this.isUnsubscribed = u;
-            this.subscription = s;
-        }
-
-        State unsubscribe() {
-            return new State(true, subscription);
-        }
-
-        State set(Subscription s) {
-            return new State(isUnsubscribed, s);
-        }
-
-    }
     @Override
     public boolean isUnsubscribed() {
-        return state.get().isUnsubscribed;
+        return state.isUnsubscribed();
     }
 
     @Override
     public void unsubscribe() {
-        State oldState;
-        State newState;
-        final AtomicReference<State> localState = this.state;
-        do {
-            oldState = localState.get();
-            if (oldState.isUnsubscribed) {
-                return;
-            } else {
-                newState = oldState.unsubscribe();
-            }
-        } while (!localState.compareAndSet(oldState, newState));
-        oldState.subscription.unsubscribe();
+        state.unsubscribe();
     }
 
     /**
@@ -78,18 +47,7 @@ public void set(Subscription s) {
         if (s == null) {
             throw new IllegalArgumentException("Subscription can not be null");
         }
-        State oldState;
-        State newState;
-        final AtomicReference<State> localState = this.state;
-        do {
-            oldState = localState.get();
-            if (oldState.isUnsubscribed) {
-                s.unsubscribe();
-                return;
-            } else {
-                newState = oldState.set(s);
-            }
-        } while (!localState.compareAndSet(oldState, newState));
+        state.replace(s);
     }
 
     /**
@@ -98,7 +56,6 @@ public void set(Subscription s) {
      * @return the {@link Subscription} that underlies the {@code MultipleAssignmentSubscription}
      */
     public Subscription get() {
-        return state.get().subscription;
+        return state.current();
     }
-
 }