Add CompositeSubscription

- also a utility method for creating a Subscription around a Future

Author: benjchristensen

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

@@ -1,3 +1,18 @@
+/**
+ * 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 java.util.concurrent.atomic.AtomicBoolean;

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

@@ -0,0 +1,80 @@
+/**
+ * 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 java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import rx.Subscription;
+import rx.util.functions.Functions;
+
+/**
+ * Subscription that represents a group of Subscriptions that are unsubscribed together.
+ * 
+ * @see Rx.Net equivalent CompositeDisposable at http://msdn.microsoft.com/en-us/library/system.reactive.disposables.compositedisposable(v=vs.103).aspx
+ */
+public class CompositeSubscription implements Subscription {
+
+    private static final Logger logger = LoggerFactory.getLogger(Functions.class);
+
+    /*
+     * The reason 'synchronized' is used on 'add' and 'unsubscribe' is because AtomicBoolean/ConcurrentLinkedQueue are both being modified so it needs to be done atomically.
+     * 
+     * TODO evaluate whether use of synchronized is a performance issue here and if it's worth using an atomic state machine or other non-locking approach
+     */
+    private AtomicBoolean unsubscribed = new AtomicBoolean(false);
+    private final ConcurrentLinkedQueue<Subscription> subscriptions = new ConcurrentLinkedQueue<Subscription>();
+
+    public CompositeSubscription(List<Subscription> subscriptions) {
+        this.subscriptions.addAll(subscriptions);
+    }
+
+    public CompositeSubscription(Subscription... subscriptions) {
+        for (Subscription s : subscriptions) {
+            this.subscriptions.add(s);
+        }
+    }
+
+    public boolean isUnsubscribed() {
+        return unsubscribed.get();
+    }
+
+    public synchronized void add(Subscription s) {
+        if (unsubscribed.get()) {
+            s.unsubscribe();
+        } else {
+            subscriptions.add(s);
+        }
+    }
+
+    @Override
+    public synchronized void unsubscribe() {
+        if (unsubscribed.compareAndSet(false, true)) {
+            for (Subscription s : subscriptions) {
+                try {
+                    s.unsubscribe();
+                } catch (Exception e) {
+                    logger.error("Failed to unsubscribe.", e);
+                }
+            }
+        }
+    }
+
+}

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

@@ -1,5 +1,22 @@
+/**
+ * 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 java.util.concurrent.Future;
+
 import rx.Subscription;
 import rx.util.functions.Action0;
 import rx.util.functions.FuncN;
@@ -31,6 +48,37 @@ public void unsubscribe() {
         };
     }
 
+    /**
+     * A {@link Subscription} that wraps a {@link Future} and cancels it when unsubscribed.
+     * 
+     * 
+     * @param f
+     *            {@link Future}
+     * @return {@link Subscription}
+     */
+    public static Subscription create(final Future<?> f) {
+        return new Subscription() {
+
+            @Override
+            public void unsubscribe() {
+                f.cancel(true);
+            }
+
+        };
+    }
+
+    /**
+     * A {@link Subscription} that groups multiple Subscriptions together and unsubscribes from all of them together.
+     * 
+     * @param subscriptions
+     *            Subscriptions to group together
+     * @return {@link Subscription}
+     */
+
+    public static CompositeSubscription create(Subscription... subscriptions) {
+        return new CompositeSubscription(subscriptions);
+    }
+
     /**
      * A {@link Subscription} implemented via an anonymous function (such as closures from other languages).
      *