1.x: observeOn + immediate scheduler to be a request rebatcher

src/main/java/rx/internal/operators/OperatorObserveOn.java

@@ -63,10 +63,7 @@ public OperatorObserveOn(Scheduler scheduler, boolean delayError, int bufferSize
 
     @Override
     public Subscriber<? super T> call(Subscriber<? super T> child) {
-        if (scheduler instanceof ImmediateScheduler) {
-            // avoid overhead, execute directly
-            return child;
-        } else if (scheduler instanceof TrampolineScheduler) {
+        if (scheduler instanceof TrampolineScheduler) {
             // avoid overhead, execute directly
             return child;
         } else {

src/test/java/rx/internal/operators/OperatorObserveOnTest.java

@@ -23,7 +23,7 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
 
-import org.junit.Test;
+import org.junit.*;
 import org.mockito.InOrder;
 
 import rx.*;
@@ -933,4 +933,27 @@ public void bufferSizesWork() {
             ts.assertNoErrors();
         }
     }
+
+    @Test
+    public void synchronousRebatching() {
+        final List<Long> requests = new ArrayList<Long>();
+
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Observable.range(1, 50)
+        .doOnRequest(new Action1<Long>() {
+            @Override
+            public void call(Long r) {
+                requests.add(r);
+            }
+        })
+        .observeOn(Schedulers.immediate(), 20)
+        .subscribe(ts);
+
+        ts.assertValueCount(50);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+
+        Assert.assertEquals(Arrays.asList(20L, 15L, 15L, 15L), requests);
+    }
 }