diff --git a/src/main/java/rx/Observable.java b/src/main/java/rx/Observable.java
index 27a97b4b3d..70a5655011 100644
--- a/src/main/java/rx/Observable.java
+++ b/src/main/java/rx/Observable.java
@@ -15,12 +15,12 @@
 import java.util.*;
 import java.util.concurrent.*;
 
+import rx.annotations.Experimental;
 import rx.exceptions.*;
 import rx.functions.*;
 import rx.internal.operators.*;
 import rx.internal.util.ScalarSynchronousObservable;
 import rx.internal.util.UtilityFunctions;
-
 import rx.observables.*;
 import rx.observers.SafeSubscriber;
 import rx.plugins.*;
@@ -182,6 +182,7 @@ public void call(Subscriber<? super R> o) {
      * @return the source Observable, transformed by the transformer function
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
      */
+    @SuppressWarnings("unchecked")
     public <R> Observable<R> compose(Transformer<? super T, ? extends R> transformer) {
         return ((Transformer<T, R>) transformer).call(this);
     }
@@ -5054,6 +5055,47 @@ public final Observable<T> onBackpressureDrop() {
         return lift(new OperatorOnBackpressureDrop<T>());
     }
     
+    /**
+     * Instructs an Observable that is emitting items faster than its observer can consume them is to
+     * block the producer thread.
+     * <p>
+     * The producer side can emit up to {@code maxQueueLength} onNext elements without blocking, but the
+     * consumer side considers the amount its downstream requested through {@code Producer.request(n)}
+     * and doesn't emit more than requested even if more is available. For example, using 
+     * {@code onBackpressureBlock(384).observeOn(Schedulers.io())} will not throw a MissingBackpressureException.
+     * <p>
+     * Note that if the upstream Observable does support backpressure, this operator ignores that capability
+     * and doesn't propagate any backpressure requests from downstream.
+     *  
+     * @param maxQueueLength the maximum number of items the producer can emit without blocking
+     * @return the source Observable modified to block {@code onNext} notifications on overflow
+     * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
+     * @Experimental The behavior of this can change at any time. 
+     */
+    @Experimental
+    public final Observable<T> onBackpressureBlock(int maxQueueLength) {
+        return lift(new OperatorOnBackpressureBlock<T>(maxQueueLength));
+    }
+    /**
+     * Instructs an Observable that is emitting items faster than its observer can consume them is to
+     * block the producer thread if the number of undelivered onNext events reaches the system-wide ring buffer size.
+     * <p>
+     * The producer side can emit up to the system-wide ring buffer size onNext elements without blocking, but the
+     * consumer side considers the amount its downstream requested through {@code Producer.request(n)}
+     * and doesn't emit more than requested even if available.
+     * <p>
+     * Note that if the upstream Observable does support backpressure, this operator ignores that capability
+     * and doesn't propagate any backpressure requests from downstream.
+     * 
+     * @return the source Observable modified to block {@code onNext} notifications on overflow
+     * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
+     * @Experimental The behavior of this can change at any time. 
+     */
+    @Experimental
+    public final Observable<T> onBackpressureBlock() {
+        return onBackpressureBlock(rx.internal.util.RxRingBuffer.SIZE);
+    }
+    
     /**
      * Instructs an Observable to pass control to another Observable rather than invoking
      * {@link Observer#onError onError} if it encounters an error.
diff --git a/src/main/java/rx/internal/operators/OperatorOnBackpressureBlock.java b/src/main/java/rx/internal/operators/OperatorOnBackpressureBlock.java
new file mode 100644
index 0000000000..bb328788e9
--- /dev/null
+++ b/src/main/java/rx/internal/operators/OperatorOnBackpressureBlock.java
@@ -0,0 +1,157 @@
+/**
+ * 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.operators;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+import rx.Observable.Operator;
+import rx.Producer;
+import rx.Subscriber;
+
+/**
+ * Operator that blocks the producer thread in case a backpressure is needed.
+ */
+public class OperatorOnBackpressureBlock<T> implements Operator<T, T> {
+    final int max;
+    public OperatorOnBackpressureBlock(int max) {
+        this.max = max;
+    }
+    @Override
+    public Subscriber<? super T> call(Subscriber<? super T> child) {
+        BlockingSubscriber<T> s = new BlockingSubscriber<T>(max, child);
+        s.init();
+        return s;
+    }
+    
+    static final class BlockingSubscriber<T> extends Subscriber<T> {
+        final NotificationLite<T> nl = NotificationLite.instance();
+        final BlockingQueue<Object> queue;
+        final Subscriber<? super T> child;
+        /** Guarded by this. */
+        long requestedCount;
+        /** Guarded by this. */
+        boolean emitting;
+        volatile boolean terminated;
+        /** Set before terminated, read after terminated. */
+        Throwable exception;
+        public BlockingSubscriber(int max, Subscriber<? super T> child) {
+            this.queue = new ArrayBlockingQueue<Object>(max);
+            this.child = child;
+        }
+        void init() {
+            child.add(this);
+            child.setProducer(new Producer() {
+                @Override
+                public void request(long n) {
+                    synchronized (BlockingSubscriber.this) {
+                        if (n == Long.MAX_VALUE || requestedCount == Long.MAX_VALUE) {
+                            requestedCount = Long.MAX_VALUE;
+                        } else {
+                            requestedCount += n;
+                        }
+                    }
+                    drain();
+                }
+            });
+        }
+        @Override
+        public void onNext(T t) {
+            try {
+                queue.put(nl.next(t));
+                drain();
+            } catch (InterruptedException ex) {
+                if (!isUnsubscribed()) {
+                    onError(ex);
+                }
+            }
+        }
+        @Override
+        public void onError(Throwable e) {
+            if (!terminated) {
+                exception = e;
+                terminated = true;
+                drain();
+            }
+        }
+        @Override
+        public void onCompleted() {
+            terminated = true;
+            drain();
+        }
+        void drain() {
+            long n;
+            synchronized (this) {
+                if (emitting) {
+                    return;
+                }
+                emitting = true;
+                n = requestedCount;
+            }
+            boolean skipFinal = false;
+            try {
+                while (true) {
+                    int emitted = 0;
+                    while (n > 0) {
+                        Object o = queue.poll();
+                        if (o == null) {
+                            if (terminated) {
+                                if (exception != null) {
+                                    child.onError(exception);
+                                } else {
+                                    child.onCompleted();
+                                }
+                                return;
+                            }
+                            break;
+                        } else {
+                            child.onNext(nl.getValue(o));
+                            n--;
+                            emitted++;
+                        }
+                    }
+                    synchronized (this) {
+                        // if no backpressure below
+                        if (requestedCount == Long.MAX_VALUE) {
+                            // no new data arrived since the last poll
+                            if (queue.peek() == null) {
+                                skipFinal = true;
+                                emitting = false;
+                                return;
+                            }
+                            n = Long.MAX_VALUE;
+                        } else {
+                            if (emitted == 0) {
+                                skipFinal = true;
+                                emitting = false;
+                                return;
+                            }
+                            requestedCount -= emitted;
+                            n = requestedCount;
+                        }
+                    }
+                }
+            } finally {
+                if (!skipFinal) {
+                    synchronized (this) {
+                        emitting = false;
+                    }
+                }
+            }
+        }
+    }
+}
diff --git a/src/test/java/rx/internal/operators/OnBackpressureBlockTest.java b/src/test/java/rx/internal/operators/OnBackpressureBlockTest.java
new file mode 100644
index 0000000000..b247b15e6b
--- /dev/null
+++ b/src/test/java/rx/internal/operators/OnBackpressureBlockTest.java
@@ -0,0 +1,262 @@
+/**
+ * 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.operators;
+
+import static org.mockito.Mockito.*;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observable.OnSubscribe;
+import rx.Observer;
+import rx.Subscriber;
+import rx.exceptions.MissingBackpressureException;
+import rx.exceptions.TestException;
+import rx.internal.util.RxRingBuffer;
+import rx.observers.TestObserver;
+import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
+
+/**
+ * Test the onBackpressureBlock() behavior.
+ */
+public class OnBackpressureBlockTest {
+    static final int WAIT = 200;
+    
+    @Test(timeout = 1000)
+    public void testSimpleBelowCapacity() {
+        Observable<Integer> source = Observable.just(1).onBackpressureBlock(10);
+        
+        TestObserver<Integer> o = new TestObserver<Integer>();
+        source.subscribe(o);
+        
+        o.assertReceivedOnNext(Arrays.asList(1));
+        o.assertTerminalEvent();
+        assertTrue(o.getOnErrorEvents().isEmpty());
+    }
+    @Test(timeout = 10000)
+    public void testSimpleAboveCapacity() throws InterruptedException {
+        Observable<Integer> source = Observable.range(1, 11).subscribeOn(Schedulers.newThread())
+                .onBackpressureBlock(10);
+        
+        TestSubscriber<Integer> o = new TestSubscriber<Integer>() {
+            @Override
+            public void onStart() {
+                request(0); // make sure it doesn't start in unlimited mode
+            }
+        };
+        source.subscribe(o);
+        o.requestMore(10);
+
+        Thread.sleep(WAIT);
+        
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
+        
+        o.requestMore(10);
+        
+        Thread.sleep(WAIT);
+
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11));
+
+        o.assertTerminalEvent();
+        assertTrue(o.getOnErrorEvents().isEmpty());
+    }
+    
+    @Test(timeout = 3000)
+    public void testNoMissingBackpressureException() {
+        final int NUM_VALUES = RxRingBuffer.SIZE * 3;
+        Observable<Integer> source = Observable.create(new OnSubscribe<Integer>() {
+            @Override
+            public void call(Subscriber<? super Integer> t1) {
+                for (int i = 0; i < NUM_VALUES; i++) {
+                    t1.onNext(i);
+                }
+                t1.onCompleted();
+            }
+        }).subscribeOn(Schedulers.newThread());
+        
+        @SuppressWarnings("unchecked")
+        Observer<Integer> o = mock(Observer.class);
+
+        TestSubscriber<Integer> s = new TestSubscriber<Integer>(o);
+        
+        source.onBackpressureBlock(RxRingBuffer.SIZE).observeOn(Schedulers.newThread()).subscribe(s);
+        
+        s.awaitTerminalEvent();
+        
+        verify(o, never()).onError(any(MissingBackpressureException.class));
+        
+        s.assertNoErrors();
+        verify(o, times(NUM_VALUES)).onNext(any(Integer.class));
+        verify(o).onCompleted();
+    }
+    @Test(timeout = 10000)
+    public void testBlockedProducerCanBeUnsubscribed() throws InterruptedException {
+        Observable<Integer> source = Observable.range(1, 11).subscribeOn(Schedulers.newThread())
+                .onBackpressureBlock(5);
+        
+        TestSubscriber<Integer> o = new TestSubscriber<Integer>() {
+            @Override
+            public void onStart() {
+                request(0); // make sure it doesn't start in unlimited mode
+            }
+        };
+        source.subscribe(o);
+        
+        o.requestMore(5);
+        
+        Thread.sleep(WAIT);
+        
+        o.unsubscribe();
+
+        Thread.sleep(WAIT);
+
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5));
+        o.assertNoErrors();
+        assertTrue(o.getOnCompletedEvents().isEmpty());
+    }
+    @Test(timeout = 10000)
+    public void testExceptionIsDelivered() throws InterruptedException {
+        Observable<Integer> source = Observable.range(1, 10)
+                .concatWith(Observable.<Integer>error(new TestException("Forced failure")))
+                .subscribeOn(Schedulers.newThread())
+                .onBackpressureBlock(5);
+        
+        TestSubscriber<Integer> o = new TestSubscriber<Integer>() {
+            @Override
+            public void onStart() {
+                request(0); // make sure it doesn't start in unlimited mode
+            }
+        };
+        source.subscribe(o);
+
+        o.requestMore(7);
+
+        Thread.sleep(WAIT);
+        
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+        o.assertNoErrors();
+        assertTrue(o.getOnCompletedEvents().isEmpty());
+
+        o.requestMore(3);
+        
+        Thread.sleep(WAIT);
+
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
+        o.assertNoErrors();
+        assertTrue(o.getOnCompletedEvents().isEmpty());
+
+        o.requestMore(10);
+        
+        Thread.sleep(WAIT);
+        
+        o.assertTerminalEvent();
+        assertEquals(1, o.getOnErrorEvents().size());
+        assertTrue(o.getOnErrorEvents().get(0) instanceof TestException);
+    }
+    @Test(timeout = 10000)
+    public void testExceptionIsDeliveredAfterValues() throws InterruptedException {
+        Observable<Integer> source = Observable.range(1, 10)
+                .concatWith(Observable.<Integer>error(new TestException("Forced failure")))
+                .subscribeOn(Schedulers.newThread())
+                .onBackpressureBlock(5);
+        
+        TestSubscriber<Integer> o = new TestSubscriber<Integer>() {
+            @Override
+            public void onStart() {
+                request(0); // make sure it doesn't start in unlimited mode
+            }
+        };
+        source.subscribe(o);
+
+        o.requestMore(7);
+
+        Thread.sleep(WAIT);
+        
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+        o.assertNoErrors();
+        assertTrue(o.getOnCompletedEvents().isEmpty());
+
+        o.requestMore(7);
+        
+        Thread.sleep(WAIT);
+
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
+        assertEquals(1, o.getOnErrorEvents().size());
+        assertTrue(o.getOnErrorEvents().get(0) instanceof TestException);
+        assertTrue(o.getOnCompletedEvents().isEmpty());
+    }
+    @Test(timeout = 10000)
+    public void testTakeWorksWithSubscriberRequesting() {
+        Observable<Integer> source = Observable.range(1, 10)
+                .concatWith(Observable.<Integer>error(new TestException("Forced failure")))
+                .subscribeOn(Schedulers.newThread())
+                .onBackpressureBlock(5).take(7);
+        
+        TestSubscriber<Integer> o = new TestSubscriber<Integer>() {
+            @Override
+            public void onStart() {
+                request(0); // make sure it doesn't start in unlimited mode
+            }
+        };
+        source.subscribe(o);
+
+        o.requestMore(7);
+
+        o.awaitTerminalEvent();
+        
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+        o.assertNoErrors();
+        o.assertTerminalEvent();
+    }
+    @Test(timeout = 10000)
+    public void testTakeWorksSubscriberRequestUnlimited() {
+        Observable<Integer> source = Observable.range(1, 10)
+                .concatWith(Observable.<Integer>error(new TestException("Forced failure")))
+                .subscribeOn(Schedulers.newThread())
+                .onBackpressureBlock(5).take(7);
+        
+        TestSubscriber<Integer> o = new TestSubscriber<Integer>();
+        source.subscribe(o);
+
+        o.awaitTerminalEvent();
+        
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+        o.assertNoErrors();
+        o.assertTerminalEvent();
+    }
+    @Test(timeout = 10000)
+    public void testTakeWorksSubscriberRequestUnlimitedBufferedException() {
+        Observable<Integer> source = Observable.range(1, 10)
+                .concatWith(Observable.<Integer>error(new TestException("Forced failure")))
+                .subscribeOn(Schedulers.newThread())
+                .onBackpressureBlock(11).take(7);
+        
+        TestSubscriber<Integer> o = new TestSubscriber<Integer>();
+        source.subscribe(o);
+
+        o.awaitTerminalEvent();
+        
+        o.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+        o.assertNoErrors();
+        o.assertTerminalEvent();
+    }
+}