diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index 56a4664654a..07c89c0b195 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -26,6 +26,8 @@
 import java.util.concurrent.TimeUnit;
 
 import rx.concurrency.Schedulers;
+import rx.joins.Pattern2;
+import rx.joins.Plan0;
 import rx.observables.BlockingObservable;
 import rx.observables.ConnectableObservable;
 import rx.observables.GroupedObservable;
@@ -51,6 +53,7 @@
 import rx.operators.OperationFirstOrDefault;
 import rx.operators.OperationGroupBy;
 import rx.operators.OperationInterval;
+import rx.operators.OperationJoinPatterns;
 import rx.operators.OperationLast;
 import rx.operators.OperationMap;
 import rx.operators.OperationMaterialize;
@@ -5689,5 +5692,173 @@ private boolean isInternalImplementation(Object o) {
             return isInternal;
         }
     }
-
+    /**
+     * Creates a pattern that matches when both observable sequences have an available element.
+     * @param right Observable sequence to match with the left sequence.
+     * @return Pattern object that matches when both observable sequences have an available element.
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229153.aspx'>MSDN: Observable.And</a>
+     * @throws NullPointerException if right is null
+     */
+    public <T2> Pattern2<T, T2> and(Observable<T2> right) {
+        return OperationJoinPatterns.and(this, right);
+    }
+    /**
+     * Matches when the observable sequence has an available element and projects the element by invoking the selector function.
+     * @param selector Selector that will be invoked for elements in the source sequence.
+     * @return Plan that produces the projected results, to be fed (with other plans) to the When operator.
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh211662.aspx'>MSDN: Observable.Then</a>
+     * @throws NullPointerException if selector is null
+     */
+    public <R> Plan0<R> then(Func1<T, R> selector) {
+        return OperationJoinPatterns.then(this, selector);
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param plans A series of plans created by use of the Then operator on patterns.
+     * @return An observable sequence with the results from matching several patterns.
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     * @throws NullPointerException if plans is null
+     */
+    public static <R> Observable<R> when(Plan0<R>... plans) {
+        return create(OperationJoinPatterns.when(plans));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param plans A series of plans created by use of the Then operator on patterns.
+     * @return An observable sequence with the results from matching several patterns.
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229558.aspx'>MSDN: Observable.When</a>
+     * @throws NullPointerException if plans is null
+     */
+    public static <R> Observable<R> when(Iterable<? extends Plan0<R>> plans) {
+        if (plans == null) {
+            throw new NullPointerException("plans");
+        }
+        return create(OperationJoinPatterns.when(plans));
+    }
+    /**
+     * Joins the results from a pattern.
+     * @param p1 the plan to join
+     * @return An observable sequence with the results from matching a pattern
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1) {
+        return create(OperationJoinPatterns.when(p1));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2) {
+        return create(OperationJoinPatterns.when(p1, p2));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @param p3 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3) {
+        return create(OperationJoinPatterns.when(p1, p2, p3));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @param p3 a plan
+     * @param p4 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan0<R> p4) {
+        return create(OperationJoinPatterns.when(p1, p2, p3, p4));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @param p3 a plan
+     * @param p4 a plan
+     * @param p5 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan0<R> p4, Plan0<R> p5) {
+        return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @param p3 a plan
+     * @param p4 a plan
+     * @param p5 a plan
+     * @param p6 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan0<R> p4, Plan0<R> p5, Plan0<R> p6) {
+        return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @param p3 a plan
+     * @param p4 a plan
+     * @param p5 a plan
+     * @param p6 a plan
+     * @param p7 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan0<R> p4, Plan0<R> p5, Plan0<R> p6, Plan0<R> p7) {
+        return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6, p7));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @param p3 a plan
+     * @param p4 a plan
+     * @param p5 a plan
+     * @param p6 a plan
+     * @param p7 a plan
+     * @param p8 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan0<R> p4, Plan0<R> p5, Plan0<R> p6, Plan0<R> p7, Plan0<R> p8) {
+        return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6, p7, p8));
+    }
+    /**
+     * Joins together the results from several patterns.
+     * @param p1 a plan
+     * @param p2 a plan
+     * @param p3 a plan
+     * @param p4 a plan
+     * @param p5 a plan
+     * @param p6 a plan
+     * @param p7 a plan
+     * @param p8 a plan
+     * @param p9 a plan
+     * @return An observable sequence with the results from matching several patterns
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229889.aspx'>MSDN: Observable.When</a>
+     */
+    @SuppressWarnings("unchecked")
+    public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan0<R> p4, Plan0<R> p5, Plan0<R> p6, Plan0<R> p7, Plan0<R> p8, Plan0<R> p9) {
+        return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6, p7, p8, p9));
+    }
 }
diff --git a/rxjava-core/src/main/java/rx/joins/ActivePlan0.java b/rxjava-core/src/main/java/rx/joins/ActivePlan0.java
new file mode 100644
index 00000000000..d8690136b75
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/ActivePlan0.java
@@ -0,0 +1,37 @@
+/**
+ * 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.joins;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Represents an activated plan.
+ */
+public abstract class ActivePlan0 {
+    protected final Map<JoinObserver, JoinObserver> joinObservers = new HashMap<JoinObserver, JoinObserver>();
+    
+    public abstract void match();
+    
+    protected void addJoinObserver(JoinObserver joinObserver) {
+        joinObservers.put(joinObserver, joinObserver);
+    }
+    protected void dequeue() {
+        for (JoinObserver jo : joinObservers.values()) {
+            jo.dequeue();
+        }
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/joins/ActivePlan1.java b/rxjava-core/src/main/java/rx/joins/ActivePlan1.java
new file mode 100644
index 00000000000..65e50e9908e
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/ActivePlan1.java
@@ -0,0 +1,49 @@
+/**
+ * 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.joins;
+
+import rx.Notification;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+
+/**
+ * Represents an active plan.
+ */
+public class ActivePlan1<T1> extends ActivePlan0 {
+    private final Action1<T1> onNext;
+    private final Action0 onCompleted;
+    private final JoinObserver1<T1> first;
+    public ActivePlan1(JoinObserver1<T1> first, Action1<T1> onNext, Action0 onCompleted) {
+        this.onNext = onNext;
+        this.onCompleted = onCompleted;
+        this.first = first;
+        addJoinObserver(first);
+    }
+
+    @Override
+    public void match() {
+        if (!first.queue().isEmpty()) {
+            Notification<T1> n1 = first.queue().peek();
+            if (n1.isOnCompleted()) {
+                onCompleted.call();
+            } else {
+                dequeue();
+                onNext.call(n1.getValue());
+            }
+        }
+    }
+    
+}
diff --git a/rxjava-core/src/main/java/rx/joins/ActivePlan2.java b/rxjava-core/src/main/java/rx/joins/ActivePlan2.java
new file mode 100644
index 00000000000..3ae19714ce6
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/ActivePlan2.java
@@ -0,0 +1,54 @@
+/**
+ * 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.joins;
+
+import rx.Notification;
+import rx.util.functions.Action0;
+import rx.util.functions.Action2;
+
+/**
+ * Represents an active plan.
+ */
+public class ActivePlan2<T1, T2> extends ActivePlan0 {
+    private final Action2<T1, T2> onNext;
+    private final Action0 onCompleted;
+    private final JoinObserver1<T1> first;
+    private final JoinObserver1<T2> second;
+    public ActivePlan2(JoinObserver1<T1> first, JoinObserver1<T2> second, Action2<T1, T2> onNext, Action0 onCompleted) {
+        this.onNext = onNext;
+        this.onCompleted = onCompleted;
+        this.first = first;
+        this.second = second;
+        addJoinObserver(first);
+        addJoinObserver(second);
+    }
+
+    @Override
+    public void match() {
+        if (!first.queue().isEmpty() && !second.queue().isEmpty()) {
+            Notification<T1> n1 = first.queue().peek();
+            Notification<T2> n2 = second.queue().peek();
+            
+            if (n1.isOnCompleted() || n2.isOnCompleted()) {
+                onCompleted.call();
+            } else {
+                dequeue();
+                onNext.call(n1.getValue(), n2.getValue());
+            }
+        }
+    }
+    
+}
diff --git a/rxjava-core/src/main/java/rx/joins/ActivePlan3.java b/rxjava-core/src/main/java/rx/joins/ActivePlan3.java
new file mode 100644
index 00000000000..cff23dd7522
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/ActivePlan3.java
@@ -0,0 +1,64 @@
+/**
+ * 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.joins;
+
+import rx.Notification;
+import rx.util.functions.Action0;
+import rx.util.functions.Action3;
+
+/**
+ * Represents an active plan.
+ */
+public class ActivePlan3<T1, T2, T3> extends ActivePlan0 {
+    private final Action3<T1, T2, T3> onNext;
+    private final Action0 onCompleted;
+    private final JoinObserver1<T1> first;
+    private final JoinObserver1<T2> second;
+    private final JoinObserver1<T3> third;
+    public ActivePlan3(JoinObserver1<T1> first, 
+            JoinObserver1<T2> second, 
+            JoinObserver1<T3> third,
+            Action3<T1, T2, T3> onNext, 
+            Action0 onCompleted) {
+        this.onNext = onNext;
+        this.onCompleted = onCompleted;
+        this.first = first;
+        this.second = second;
+        this.third = third;
+        addJoinObserver(first);
+        addJoinObserver(second);
+        addJoinObserver(third);
+    }
+
+    @Override
+    public void match() {
+        if (!first.queue().isEmpty() 
+                && !second.queue().isEmpty()
+                && !third.queue().isEmpty()) {
+            Notification<T1> n1 = first.queue().peek();
+            Notification<T2> n2 = second.queue().peek();
+            Notification<T3> n3 = third.queue().peek();
+            
+            if (n1.isOnCompleted() || n2.isOnCompleted() || n3.isOnCompleted()) {
+                onCompleted.call();
+            } else {
+                dequeue();
+                onNext.call(n1.getValue(), n2.getValue(), n3.getValue());
+            }
+        }
+    }
+
+}
diff --git a/rxjava-core/src/main/java/rx/joins/JoinObserver.java b/rxjava-core/src/main/java/rx/joins/JoinObserver.java
new file mode 100644
index 00000000000..d191ed89588
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/JoinObserver.java
@@ -0,0 +1,26 @@
+/**
+ * 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.joins;
+
+import rx.Subscription;
+
+/**
+ * Base interface to manage joined observations.
+ */
+public interface JoinObserver extends Subscription {
+    void subscribe(Object gate);
+    void dequeue();
+}
diff --git a/rxjava-core/src/main/java/rx/joins/JoinObserver1.java b/rxjava-core/src/main/java/rx/joins/JoinObserver1.java
new file mode 100644
index 00000000000..873d3d1a7fa
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/JoinObserver1.java
@@ -0,0 +1,107 @@
+/**
+ * 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.joins;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import rx.Notification;
+import rx.Observable;
+import rx.subscriptions.SingleAssignmentSubscription;
+import rx.util.functions.Action1;
+
+/**
+ * Default implementation of a join observer.
+ */
+public final class JoinObserver1<T> extends ObserverBase<Notification<T>> implements JoinObserver {
+    private Object gate;
+    private final Observable<T> source;
+    private final Action1<Throwable> onError;
+    private final List<ActivePlan0> activePlans;
+    private final Queue<Notification<T>> queue;
+    private final SingleAssignmentSubscription subscription;
+    private volatile boolean done;
+    
+    public JoinObserver1(Observable<T> source, Action1<Throwable> onError) {
+        this.source = source;
+        this.onError = onError;
+        queue = new LinkedList<Notification<T>>();
+        subscription = new SingleAssignmentSubscription();
+        activePlans = new ArrayList<ActivePlan0>();
+    }
+    public Queue<Notification<T>> queue() {
+        return queue;
+    }
+    public void addActivePlan(ActivePlan0 activePlan) {
+        activePlans.add(activePlan);
+    }
+    @Override
+    public void subscribe(Object gate) {
+        this.gate = gate;
+        subscription.set(source.materialize().subscribe(this));
+    }
+
+    @Override
+    public void dequeue() {
+        queue.remove();
+    }
+
+    @Override
+    protected void onNextCore(Notification<T> args) {
+        synchronized (gate) {
+            if (!done) {
+                if (args.isOnError()) {
+                    onError.call(args.getThrowable());
+                    return;
+                }
+                queue.add(args);
+                
+                // remark: activePlans might change while iterating
+                for (ActivePlan0 a : new ArrayList<ActivePlan0>(activePlans)) {
+                    a.match();
+                }
+            }
+        }
+    }
+
+    @Override
+    protected void onErrorCore(Throwable e) {
+        // not expected
+    }
+
+    @Override
+    protected void onCompletedCore() {
+        // not expected or ignored
+    }
+    
+    
+    void removeActivePlan(ActivePlan0 activePlan) {
+        activePlans.remove(activePlan);
+        if (activePlans.isEmpty()) {
+            unsubscribe();
+        }
+    }
+
+    @Override
+    public void unsubscribe() {
+        if (!done) {
+            done = true;
+            subscription.unsubscribe();
+        }
+    }
+    
+}
diff --git a/rxjava-core/src/main/java/rx/joins/ObserverBase.java b/rxjava-core/src/main/java/rx/joins/ObserverBase.java
new file mode 100644
index 00000000000..f1144a8ad26
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/ObserverBase.java
@@ -0,0 +1,72 @@
+/**
+ * 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.joins;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import rx.Observer;
+
+/**
+ * Implements an observer that ensures proper event delivery
+ * semantics to its abstract onXxxxCore methods.
+ */
+public abstract class ObserverBase<T> implements Observer<T> {
+    private final AtomicBoolean completed = new AtomicBoolean();
+
+    @Override
+    public void onNext(T args) {
+        if (!completed.get()) {
+            onNextCore(args);
+        }
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        if (completed.compareAndSet(false, true)) {
+            onErrorCore(e);
+        }
+    }
+
+    @Override
+    public void onCompleted() {
+        if (completed.compareAndSet(false, true)) {
+            onCompletedCore();
+        }
+    }
+    /**
+     * Implement this method to react to the receival of a new element in the sequence.
+     */
+    protected abstract void onNextCore(T args);
+    /**
+     * Implement this method to react to the occurrence of an exception.
+     */
+    protected abstract void onErrorCore(Throwable e);
+    /**
+     * Implement this method to react to the end of the sequence.
+     */
+    protected abstract void onCompletedCore();
+    /**
+     * Try to trigger the error state.
+     * @param t 
+     * @return false if already completed
+     */
+    protected boolean fail(Throwable t) {
+        if (completed.compareAndSet(false, true)) {
+            onErrorCore(t);
+            return true;
+        }
+        return false;
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Pattern.java b/rxjava-core/src/main/java/rx/joins/Pattern.java
new file mode 100644
index 00000000000..1dc65ca3b88
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Pattern.java
@@ -0,0 +1,25 @@
+/**
+ * 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.joins;
+
+/**
+ * Base interface for join patterns.
+ * @see <a href='http://msdn.microsoft.com/en-us/library/system.reactive.joins.pattern(v=vs.103).aspx'>MSDN: Pattern</a>
+ */
+public interface Pattern {
+    
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Pattern1.java b/rxjava-core/src/main/java/rx/joins/Pattern1.java
new file mode 100644
index 00000000000..324daf79fa5
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Pattern1.java
@@ -0,0 +1,45 @@
+/**
+ * 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.joins;
+
+import rx.Observable;
+import rx.util.functions.Func1;
+
+/**
+ * Represents a join pattern over one observable sequence.
+ */
+public class Pattern1<T1> implements Pattern {
+    private final Observable<T1> first;
+    public Pattern1(Observable<T1> first) {
+        this.first = first;
+    }
+    public Observable<T1> first() {
+        return first;
+    }
+    /**
+     * Matches when all observable sequences have an available 
+     * element and projects the elements by invoking the selector function.
+     * @param selector the function that will be invoked for elements in the source sequences.
+     * @return 
+     * @throws NullPointerException if selector is null
+     */
+    public <R> Plan0<R> then(Func1<T1, R> selector) {
+        if (selector == null) {
+            throw new NullPointerException();
+        }
+        return new Plan1<T1, R>(this, selector);
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Pattern2.java b/rxjava-core/src/main/java/rx/joins/Pattern2.java
new file mode 100644
index 00000000000..6f82a66d49f
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Pattern2.java
@@ -0,0 +1,54 @@
+/**
+ * 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.joins;
+
+import rx.Observable;
+import rx.util.functions.Func2;
+
+/**
+ * Represents a join pattern over observable sequences.
+ */
+public class Pattern2<T1, T2> implements Pattern {
+    private final Observable<T1> first;
+    private final Observable<T2> second;
+    public Pattern2(Observable<T1> first, Observable<T2> second) {
+        this.first = first;
+        this.second = second;
+    }
+    public Observable<T1> first() {
+        return first;
+    }
+    public Observable<T2> second() {
+        return second;
+    }
+    /**
+     * Creates a pattern that matches when all three observable sequences have an available element.
+     * @param other Observable sequence to match with the two previous sequences.
+     * @return Pattern object that matches when all observable sequences have an available element.
+     */
+    public <T3> Pattern3<T1, T2, T3> and(Observable<T3> other) {
+        if (other == null) {
+            throw new NullPointerException();
+        }
+        return new Pattern3<T1, T2, T3>(first, second, other);
+    }
+    public <R> Plan0<R> then(Func2<T1, T2, R> selector) {
+        if (selector == null) {
+            throw new NullPointerException();
+        }
+        return new Plan2<T1, T2, R>(this, selector);
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Pattern3.java b/rxjava-core/src/main/java/rx/joins/Pattern3.java
new file mode 100644
index 00000000000..c43b9d0b9e7
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Pattern3.java
@@ -0,0 +1,55 @@
+/**
+ * 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.joins;
+
+import rx.Observable;
+import rx.util.functions.Func3;
+
+/**
+ * Represents a join pattern over observable sequences.
+ */
+public class Pattern3<T1, T2, T3> implements Pattern {
+    private final Observable<T1> first;
+    private final Observable<T2> second;
+    private final Observable<T3> third;
+    public Pattern3(Observable<T1> first, Observable<T2> second,
+            Observable<T3> third) {
+        this.first = first;
+        this.second = second;
+        this.third = third;
+    }
+    public Observable<T1> first() {
+        return first;
+    }
+    public Observable<T2> second() {
+        return second;
+    }
+    public Observable<T3> third() {
+        return third;
+    }
+//    public <T4> Pattern4<T1, T2, T3, T4> and(Observable<T4> other) {
+//        if (other == null) {
+//            throw new NullPointerException();
+//        }
+//        return new Pattern4<T1, T2, T3, T4>(first, second, third, other);
+//    }
+    public <R> Plan0<R> then(Func3<T1, T2, T3, R> selector) {
+        if (selector == null) {
+            throw new NullPointerException();
+        }
+        return new Plan3<T1, T2, T3, R>(this, selector);
+    }    
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Plan0.java b/rxjava-core/src/main/java/rx/joins/Plan0.java
new file mode 100644
index 00000000000..2a647aff3de
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Plan0.java
@@ -0,0 +1,46 @@
+/**
+ * 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.joins;
+
+import java.util.Map;
+import rx.Observable;
+import rx.Observer;
+import rx.util.functions.Action1;
+
+/**
+ * Represents an execution plan for join patterns.
+ */
+public abstract class Plan0<R> {
+    public abstract ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions, 
+            Observer<R> observer, Action1<ActivePlan0> deactivate);
+    
+    @SuppressWarnings("unchecked")
+    public static <T> JoinObserver1<T> createObserver(
+            Map<Object, JoinObserver> externalSubscriptions,
+            Observable<T> observable,
+            Action1<Throwable> onError
+    ) {
+        JoinObserver1<T> observer;
+        JoinObserver nonGeneric = externalSubscriptions.get(observable);
+        if (nonGeneric == null) {
+            observer = new JoinObserver1<T>(observable, onError);
+            externalSubscriptions.put(observable, observer);
+        } else {
+            observer = (JoinObserver1<T>)nonGeneric;
+        }
+        return observer;
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Plan1.java b/rxjava-core/src/main/java/rx/joins/Plan1.java
new file mode 100644
index 00000000000..b64742e27bd
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Plan1.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.joins;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import rx.Observer;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+import rx.util.functions.Func1;
+import rx.util.functions.Actions;
+
+/**
+ * Represents an execution plan for join patterns.
+ */
+public class Plan1<T1, R> extends Plan0<R> {
+    protected Pattern1<T1> expression;
+    protected Func1<T1, R> selector;
+    
+    public Plan1(Pattern1<T1> expression, Func1<T1, R> selector) {
+        this.expression = expression;
+        this.selector = selector;
+    }
+    
+    public Pattern1<T1> expression() {
+        return expression;
+    }
+    public Func1<T1, R> selector() {
+        return selector;
+    }
+
+    @Override
+    public ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions, final Observer<R> observer, final Action1<ActivePlan0> deactivate) {
+        Action1<Throwable> onError = Actions.onErrorFrom(observer);
+        
+        final JoinObserver1<T1> firstJoinObserver = createObserver(externalSubscriptions, expression.first(), onError);
+        
+        final AtomicReference<ActivePlan1<T1>> self = new AtomicReference<ActivePlan1<T1>>();
+        
+        ActivePlan1<T1> activePlan = new ActivePlan1<T1>(firstJoinObserver, new Action1<T1>() {
+            @Override
+            public void call(T1 t1) {
+                R result;
+                try {
+                    result = selector.call(t1);
+                } catch (Throwable t) {
+                    observer.onError(t);
+                    return;
+                }
+                observer.onNext(result);
+            }
+        },
+        new Action0() {
+            @Override
+            public void call() {
+                firstJoinObserver.removeActivePlan(self.get());
+                deactivate.call(self.get());
+            }
+        });
+        
+        self.set(activePlan);
+        
+        firstJoinObserver.addActivePlan(activePlan);
+        return activePlan;
+    }
+    
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Plan2.java b/rxjava-core/src/main/java/rx/joins/Plan2.java
new file mode 100644
index 00000000000..71b931ea0de
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Plan2.java
@@ -0,0 +1,79 @@
+/**
+ * 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.joins;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import rx.Observer;
+import static rx.joins.Plan0.createObserver;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+import rx.util.functions.Action2;
+import rx.util.functions.Actions;
+import rx.util.functions.Func2;
+
+/**
+ * Represents an execution plan for join patterns.
+ */
+public class Plan2<T1, T2, R> extends Plan0<R> {
+    protected Pattern2<T1, T2> expression;
+    protected Func2<T1, T2, R> selector;
+    public Plan2(Pattern2<T1, T2> expression, Func2<T1, T2, R> selector) {
+        this.expression = expression;
+        this.selector = selector;
+    }
+
+    @Override
+    public ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions, 
+            final Observer<R> observer, final Action1<ActivePlan0> deactivate) {
+        Action1<Throwable> onError = Actions.onErrorFrom(observer);
+        
+        final JoinObserver1<T1> firstJoinObserver = createObserver(externalSubscriptions, expression.first(), onError);
+        final JoinObserver1<T2> secondJoinObserver = createObserver(externalSubscriptions, expression.second(), onError);
+        
+        final AtomicReference<ActivePlan2<T1, T2>> self = new AtomicReference<ActivePlan2<T1, T2>>();
+        
+        ActivePlan2<T1, T2> activePlan = new ActivePlan2<T1, T2>(firstJoinObserver, secondJoinObserver, new Action2<T1, T2>() {
+            @Override
+            public void call(T1 t1, T2 t2) {
+                R result;
+                try {
+                    result = selector.call(t1, t2);
+                } catch (Throwable t) {
+                    observer.onError(t);
+                    return;
+                }
+                observer.onNext(result);
+            }
+        },
+        new Action0() {
+            @Override
+            public void call() {
+                firstJoinObserver.removeActivePlan(self.get());
+                secondJoinObserver.removeActivePlan(self.get());
+                deactivate.call(self.get());
+            }
+        });
+        
+        self.set(activePlan);
+        
+        firstJoinObserver.addActivePlan(activePlan);
+        secondJoinObserver.addActivePlan(activePlan);
+        
+        return activePlan;
+    }
+    
+}
diff --git a/rxjava-core/src/main/java/rx/joins/Plan3.java b/rxjava-core/src/main/java/rx/joins/Plan3.java
new file mode 100644
index 00000000000..386ef82dfec
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/joins/Plan3.java
@@ -0,0 +1,83 @@
+/**
+ * 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.joins;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import rx.Observer;
+import static rx.joins.Plan0.createObserver;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+import rx.util.functions.Action3;
+import rx.util.functions.Actions;
+import rx.util.functions.Func3;
+
+/**
+ * Represents an execution plan for join patterns.
+ */
+public class Plan3<T1, T2, T3, R> extends Plan0<R> {
+    protected Pattern3<T1, T2, T3> expression;
+    protected Func3<T1, T2, T3, R> selector;
+    public Plan3(Pattern3<T1, T2, T3> expression, Func3<T1, T2, T3, R> selector) {
+        this.expression = expression;
+        this.selector = selector;
+    }
+
+    @Override
+    public ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions, 
+            final Observer<R> observer, final Action1<ActivePlan0> deactivate) {
+        Action1<Throwable> onError = Actions.onErrorFrom(observer);
+        
+        final JoinObserver1<T1> firstJoinObserver = createObserver(externalSubscriptions, expression.first(), onError);
+        final JoinObserver1<T2> secondJoinObserver = createObserver(externalSubscriptions, expression.second(), onError);
+        final JoinObserver1<T3> thirdJoinObserver = createObserver(externalSubscriptions, expression.third(), onError);
+        
+        final AtomicReference<ActivePlan3<T1, T2, T3>> self = new AtomicReference<ActivePlan3<T1, T2, T3>>();
+        
+        ActivePlan3<T1, T2, T3> activePlan = new ActivePlan3<T1, T2, T3>(firstJoinObserver, secondJoinObserver, 
+                thirdJoinObserver, new Action3<T1, T2, T3>() {
+            @Override
+            public void call(T1 t1, T2 t2, T3 t3) {
+                R result;
+                try {
+                    result = selector.call(t1, t2, t3);
+                } catch (Throwable t) {
+                    observer.onError(t);
+                    return;
+                }
+                observer.onNext(result);
+            }
+        },
+        new Action0() {
+            @Override
+            public void call() {
+                firstJoinObserver.removeActivePlan(self.get());
+                secondJoinObserver.removeActivePlan(self.get());
+                thirdJoinObserver.removeActivePlan(self.get());
+                deactivate.call(self.get());
+            }
+        });
+        
+        self.set(activePlan);
+        
+        firstJoinObserver.addActivePlan(activePlan);
+        secondJoinObserver.addActivePlan(activePlan);
+        thirdJoinObserver.addActivePlan(activePlan);
+        
+        return activePlan;
+    }
+
+}
diff --git a/rxjava-core/src/main/java/rx/operators/OperationJoinPatterns.java b/rxjava-core/src/main/java/rx/operators/OperationJoinPatterns.java
new file mode 100644
index 00000000000..6ff1c4346ea
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/operators/OperationJoinPatterns.java
@@ -0,0 +1,131 @@
+ /**
+  * 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.operators;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.joins.ActivePlan0;
+import rx.joins.JoinObserver;
+import rx.joins.Pattern1;
+import rx.joins.Pattern2;
+import rx.joins.Plan0;
+import rx.subjects.PublishSubject;
+import rx.subscriptions.CompositeSubscription;
+import rx.util.functions.Action1;
+import rx.util.functions.Func1;
+import rx.util.functions.Func2;
+
+/**
+ * Join patterns: And, Then, When.
+ */
+public class OperationJoinPatterns {
+    /**
+     * Creates a pattern that matches when both observable sequences have an available element.
+     */
+    public static <T1, T2> Pattern2<T1, T2> and(/* this */Observable<T1> left, Observable<T2> right) {
+        if (left == null) {
+            throw new NullPointerException("left");
+        }
+        if (right == null) {
+            throw new NullPointerException("right");
+        }
+        return new Pattern2<T1, T2>(left, right);
+    }
+    /**
+     * Matches when the observable sequence has an available element and projects the element by invoking the selector function.
+     */
+    public static <T1, R> Plan0<R> then(/* this */Observable<T1> source, Func1<T1, R> selector) {
+        if (source == null) {
+            throw new NullPointerException("source");
+        }
+        if (selector == null) {
+            throw new NullPointerException("selector");
+        }
+        return new Pattern1<T1>(source).then(selector);
+    }
+    /**
+     * Joins together the results from several patterns.
+     */
+    public static <R> OnSubscribeFunc<R> when(Plan0<R>... plans) {
+        if (plans == null) {
+            throw new NullPointerException("plans");
+        }
+        return when(Arrays.asList(plans));
+    }
+    /**
+     * Joins together the results from several patterns.
+     */
+    public static <R> OnSubscribeFunc<R> when(final Iterable<? extends Plan0<R>> plans) {
+        if (plans == null) {
+            throw new NullPointerException("plans");
+        }
+        return new OnSubscribeFunc<R>() {
+            @Override
+            public Subscription onSubscribe(final Observer<? super R> t1) {
+                final Map<Object, JoinObserver> externalSubscriptions = new HashMap<Object, JoinObserver>();
+                final Object gate = new Object();
+                final List<ActivePlan0> activePlans = new ArrayList<ActivePlan0>();
+                
+                final Observer<R> out = new Observer<R>() {
+                    @Override
+                    public void onNext(R args) {
+                        t1.onNext(args);
+                    }
+                    @Override
+                    public void onError(Throwable e) {
+                        for (JoinObserver po : externalSubscriptions.values()) {
+                            po.unsubscribe();
+                        }
+                        t1.onError(e);
+                    }
+                    @Override
+                    public void onCompleted() {
+                        t1.onCompleted();
+                    }
+                };
+                
+                try {
+                    for (Plan0<R> plan : plans) {
+                        activePlans.add(plan.activate(externalSubscriptions, out, new Action1<ActivePlan0>() {
+                            @Override
+                            public void call(ActivePlan0 activePlan) {
+                                activePlans.remove(activePlan);
+                                if (activePlans.isEmpty()) {
+                                    out.onCompleted();
+                                }
+                            }
+                        }));
+                    }
+                } catch (Throwable t) {
+                    return Observable.<R>error(t).subscribe(t1);
+                }
+                CompositeSubscription group = new CompositeSubscription();
+                for (JoinObserver jo : externalSubscriptions.values()) {
+                    jo.subscribe(gate);
+                    group.add(jo);
+                }
+                return group;
+            }
+        };
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/subscriptions/SingleAssignmentSubscription.java b/rxjava-core/src/main/java/rx/subscriptions/SingleAssignmentSubscription.java
new file mode 100644
index 00000000000..492c4e6dbf4
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/subscriptions/SingleAssignmentSubscription.java
@@ -0,0 +1,81 @@
+/**
+ * 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.AtomicReference;
+import rx.Subscription;
+
+/**
+ * A subscription that allows only a single resource to be assigned.
+ * <p>
+ * If this subscription is live, no other subscription may be set() and
+ * yields an {@link IllegalStateException}.
+ * <p>
+ * If the unsubscribe has been called, setting a new subscription will
+ * unsubscribe it immediately.
+ */
+public final class SingleAssignmentSubscription implements Subscription {
+    /** Holds the current resource. */
+    private final AtomicReference<Subscription> current = new AtomicReference<Subscription>();
+    /** Sentinel for the unsubscribed state. */
+    private static final Subscription SENTINEL = new Subscription() {
+        @Override
+        public void unsubscribe() {
+        }
+    };
+    /**
+     * Returns the current subscription or null if not yet set.
+     */
+    public Subscription get() {
+        Subscription s = current.get();
+        if (s == SENTINEL) {
+            return Subscriptions.empty();
+        }
+        return s;
+    }
+    /**
+     * Sets a new subscription if not already set.
+     * @param s the new subscription
+     * @throws IllegalStateException if this subscription is live and contains
+     * another subscription.
+     */
+    public void set(Subscription s) {
+        if (current.compareAndSet(null, s)) {
+            return;
+        }
+        if (current.get() != SENTINEL) {
+            throw new IllegalStateException("Subscription already set");
+        }
+        if (s != null) {
+            s.unsubscribe();
+        }
+    }
+    @Override
+    public void unsubscribe() {
+        Subscription old = current.getAndSet(SENTINEL);
+        if (old != null) {
+            old.unsubscribe();
+        }
+    }
+    /**
+     * Test if this subscription is already unsubscribed.
+     */
+    public boolean isUnsubscribed() {
+        return current.get() == SENTINEL;
+    }
+    
+}
diff --git a/rxjava-core/src/main/java/rx/util/functions/Actions.java b/rxjava-core/src/main/java/rx/util/functions/Actions.java
new file mode 100644
index 00000000000..2bffbb9ec8a
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/util/functions/Actions.java
@@ -0,0 +1,72 @@
+/**
+ * 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.util.functions;
+
+import rx.Observer;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+
+/**
+ * Utility class for the Action interfaces.
+ */
+public final class Actions {
+    private Actions() { throw new IllegalStateException("No instances!"); }
+    /**
+     * Extracts a method reference to the observer's onNext method
+     * in the form of an Action1.
+     * <p>Java 8: observer::onNext</p>
+     * @param observer the observer to use
+     * @return an action which calls the observer's onNext method.
+     */
+    public static <T> Action1<T> onNextFrom(final Observer<T> observer) {
+        return new Action1<T>() {
+            @Override
+            public void call(T t1) {
+                observer.onNext(t1);
+            }
+        };
+    }
+    /**
+     * Extracts a method reference to the observer's onError method
+     * in the form of an Action1.
+     * <p>Java 8: observer::onError</p>
+     * @param observer the observer to use
+     * @return an action which calls the observer's onError method.
+     */
+    public static <T> Action1<Throwable> onErrorFrom(final Observer<T> observer) {
+        return new Action1<Throwable>() {
+            @Override
+            public void call(Throwable t1) {
+                observer.onError(t1);
+            }
+        };
+    }
+    /**
+     * Extracts a method reference to the observer's onCompleted method
+     * in the form of an Action0.
+     * <p>Java 8: observer::onCompleted</p>
+     * @param observer the observer to use
+     * @return an action which calls the observer's onCompleted method.
+     */
+    public static <T> Action0 onCompletedFrom(final Observer<T> observer) {
+        return new Action0() {
+            @Override
+            public void call() {
+                observer.onCompleted();
+            }
+        };
+    }
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperationJoinsTest.java b/rxjava-core/src/test/java/rx/operators/OperationJoinsTest.java
new file mode 100644
index 00000000000..8a3b0c79c98
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/operators/OperationJoinsTest.java
@@ -0,0 +1,382 @@
+/**
+ * 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.operators;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InOrder;
+import static org.mockito.Matchers.any;
+import org.mockito.Mock;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import org.mockito.MockitoAnnotations;
+import rx.Observable;
+import rx.Observer;
+import rx.joins.Plan0;
+import rx.subjects.PublishSubject;
+import rx.util.functions.Func1;
+import rx.util.functions.Func2;
+import rx.util.functions.Func3;
+import rx.util.functions.Functions;
+
+public class OperationJoinsTest {
+    @Mock
+    Observer<Object> observer;
+    
+    
+    Func2<Integer, Integer, Integer> add2 = new Func2<Integer, Integer, Integer>() {
+        @Override
+        public Integer call(Integer t1, Integer t2) {
+            return t1 + t2;
+        }
+    };
+    Func2<Integer, Integer, Integer> mul2 = new Func2<Integer, Integer, Integer>() {
+        @Override
+        public Integer call(Integer t1, Integer t2) {
+            return t1 * t2;
+        }
+    };
+    Func2<Integer, Integer, Integer> sub2 = new Func2<Integer, Integer, Integer>() {
+        @Override
+        public Integer call(Integer t1, Integer t2) {
+            return t1 - t2;
+        }
+    };
+    
+    Func3<Integer, Integer, Integer, Integer> add3 = new Func3<Integer, Integer, Integer, Integer>() {
+        @Override
+        public Integer call(Integer t1, Integer t2, Integer t3) {
+            return t1 + t2 + t3;
+        }
+    };
+    Func1<Integer, Integer> func1Throw = new Func1<Integer, Integer>() {
+        @Override
+        public Integer call(Integer t1) {
+            throw new RuntimeException("Forced failure");
+        }
+    };
+    Func2<Integer, Integer, Integer> func2Throw = new Func2<Integer, Integer, Integer>() {
+        @Override
+        public Integer call(Integer t1, Integer t2) {
+            throw new RuntimeException("Forced failure");
+        }
+    };
+    Func3<Integer, Integer, Integer, Integer> func3Throw = new Func3<Integer, Integer, Integer, Integer>() {
+        @Override
+        public Integer call(Integer t1, Integer t2, Integer t3) {
+            throw new RuntimeException("Forced failure");
+        }
+    };
+    
+    @Before
+    public void before() {
+        MockitoAnnotations.initMocks(this);
+    }
+    
+    @Test(expected = NullPointerException.class)
+    public void and2ArgumentNull() {
+        Observable<Integer> some = Observable.just(1);
+        some.and(null);
+    }
+    @Test(expected = NullPointerException.class)
+    public void and3argumentNull() {
+        Observable<Integer> some = Observable.just(1);
+        some.and(some).and(null);
+    }
+    @Test
+    public void and2() {
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.and(some).then(add2));
+        
+        m.subscribe(observer);
+        
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, times(1)).onNext(2);
+        verify(observer, times(1)).onCompleted();
+    }
+    @Test
+    public void and2Error1() {
+        Observable<Integer> error = Observable.error(new RuntimeException("Forced failure"));
+        
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(error.and(some).then(add2));
+        
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void and2Error2() {
+        Observable<Integer> error = Observable.error(new RuntimeException("Forced failure"));
+        
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.and(error).then(add2));
+        
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void and3() {
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.and(some).and(some).then(add3));
+        
+        m.subscribe(observer);
+        
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, times(1)).onNext(3);
+        verify(observer, times(1)).onCompleted();
+    }
+    @Test
+    public void and3Error1() {
+        Observable<Integer> error = Observable.error(new RuntimeException("Forced failure"));
+        
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(error.and(some).and(some).then(add3));
+        
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void and3Error2() {
+        Observable<Integer> error = Observable.error(new RuntimeException("Forced failure"));
+        
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.and(error).and(some).then(add3));
+        
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void and3Error3() {
+        Observable<Integer> error = Observable.error(new RuntimeException("Forced failure"));
+        
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.and(some).and(error).then(add3));
+        
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test(expected = NullPointerException.class)
+    public void thenArgumentNull() {
+        Observable<Integer> some = Observable.just(1);
+        
+        some.then(null);
+    }
+    @Test(expected = NullPointerException.class)
+    public void then2ArgumentNull() {
+        Observable<Integer> some = Observable.just(1);
+        
+        some.and(some).then(null);
+    }
+    @Test(expected = NullPointerException.class)
+    public void then3ArgumentNull() {
+        Observable<Integer> some = Observable.just(1);
+        
+        some.and(some).and(some).then(null);
+    }
+    @Test
+    public void then1() {
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.then(Functions.<Integer>identity()));
+        m.subscribe(observer);
+        
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, times(1)).onNext(1);
+        verify(observer, times(1)).onCompleted();
+    }
+    @Test
+    public void then1Error() {
+        Observable<Integer> some = Observable.error(new RuntimeException("Forced failure"));
+        
+        Observable<Integer> m = Observable.when(some.then(Functions.<Integer>identity()));
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void then1Throws() {
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.then(func1Throw));
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void then2Throws() {
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.and(some).then(func2Throw));
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void then3Throws() {
+        Observable<Integer> some = Observable.just(1);
+        
+        Observable<Integer> m = Observable.when(some.and(some).and(some).then(func3Throw));
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test(expected = NullPointerException.class)
+    public void whenArgumentNull1() {
+        Observable.when((Plan0<Object>[])null);
+    }
+    @Test(expected = NullPointerException.class)
+    public void whenArgumentNull2() {
+        Observable.when((Iterable<Plan0<Object>>)null);
+    }
+    @Test
+    public void whenMultipleSymmetric() {
+        Observable<Integer> source1 = Observable.from(1, 2, 3);
+        Observable<Integer> source2 = Observable.from(4, 5, 6);
+        
+        Observable<Integer> m = Observable.when(source1.and(source2).then(add2));
+        m.subscribe(observer);
+        
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, times(1)).onNext(1 + 4);
+        verify(observer, times(1)).onNext(2 + 5);
+        verify(observer, times(1)).onNext(3 + 6);
+        verify(observer, times(1)).onCompleted();        
+    }
+    
+    @Test
+    public void whenMultipleAsymSymmetric() {
+        Observable<Integer> source1 = Observable.from(1, 2, 3);
+        Observable<Integer> source2 = Observable.from(4, 5);
+        
+        Observable<Integer> m = Observable.when(source1.and(source2).then(add2));
+        m.subscribe(observer);
+        
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, times(1)).onNext(1 + 4);
+        verify(observer, times(1)).onNext(2 + 5);
+        verify(observer, times(1)).onCompleted();        
+    }
+    @Test
+    public void whenEmptyEmpty() {
+        Observable<Integer> source1 = Observable.empty();
+        Observable<Integer> source2 = Observable.empty();
+        
+        Observable<Integer> m = Observable.when(source1.and(source2).then(add2));
+        m.subscribe(observer);
+        
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, times(1)).onCompleted();        
+    }
+    
+    @Test
+    public void whenNeverNever() {
+        Observable<Integer> source1 = Observable.never();
+        Observable<Integer> source2 = Observable.never();
+        
+        Observable<Integer> m = Observable.when(source1.and(source2).then(add2));
+        m.subscribe(observer);
+        
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();        
+    }
+    @Test
+    public void whenThrowNonEmpty() {
+        Observable<Integer> source1 = Observable.empty();
+        Observable<Integer> source2 = Observable.error(new RuntimeException("Forced failure"));
+        
+        Observable<Integer> m = Observable.when(source1.and(source2).then(add2));
+        m.subscribe(observer);
+        
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();        
+    }
+    @Test
+    public void whenComplicated() {
+        PublishSubject<Integer> xs = PublishSubject.create();
+        PublishSubject<Integer> ys = PublishSubject.create();
+        PublishSubject<Integer> zs = PublishSubject.create();
+        
+        Observable<Integer> m = Observable.when(
+            xs.and(ys).then(add2),
+            xs.and(zs).then(mul2),
+            ys.and(zs).then(sub2)
+        );
+        m.subscribe(observer);
+        
+        xs.onNext(1); // t == 210
+        
+        xs.onNext(2); // t == 220
+        zs.onNext(7); // t == 220
+        
+        xs.onNext(3); // t == 230
+        zs.onNext(8); // t == 230
+
+        ys.onNext(4); // t == 240
+        zs.onNext(9); // t == 240
+        xs.onCompleted(); // t == 240
+
+        ys.onNext(5); // t == 250
+        
+        ys.onNext(6); // t == 260
+        
+        ys.onCompleted(); // t == 270
+        
+        zs.onCompleted(); // t == 300
+        
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer, times(1)).onNext(1 * 7);
+        inOrder.verify(observer, times(1)).onNext(2 * 8);
+        inOrder.verify(observer, times(1)).onNext(3 + 4);
+        inOrder.verify(observer, times(1)).onNext(5 - 9);
+        inOrder.verify(observer, times(1)).onCompleted();         
+        verify(observer, never()).onError(any(Throwable.class));
+    }
+}