Example and proper test for subscribe_on

Author: Bittrance

examples/subscribe_on.rb

@@ -0,0 +1,38 @@
+require 'rx'
+
+def in_a_second(n)
+  Rx::Observable.create do |observer|
+    sleep(1)
+    observer.on_next(n)
+    observer.on_completed
+  end
+end
+
+# subscribe_on will schedule ancestry on given scheduler; each sleep happens in a separate thread
+source = Rx::Observable.of(
+  in_a_second(1)
+    .map {|v| v + 1 }
+    .subscribe_on(Rx::DefaultScheduler.instance),
+  in_a_second(2)
+    .map {|v| v + 1 }
+    .subscribe_on(Rx::DefaultScheduler.instance)
+).merge_all
+.time_interval
+
+subscription = source.subscribe(
+    lambda {|x|
+        puts 'Next: ' + x.to_s
+    },
+    lambda {|err|
+        puts 'Error: ' + err.to_s
+    },
+    lambda {
+        puts 'Completed'
+    })
+
+# => Next: (3)@(1.004153)
+# => Next: (2)@(0.000251)
+
+while Thread.list.size > 1
+  (Thread.list - [Thread.current]).each &:join
+end

test/rx/operators/test_synchronization.rb

@@ -5,6 +5,17 @@
 class TestObservableSynchronization < Minitest::Test
   include Rx::ReactiveTest
 
+  def test_subscribe_on
+    scheduler = Rx::TestScheduler.new
+    mock = Rx::MockObserver.new scheduler
+    Rx::Observable.just(1)
+      .subscribe_on(scheduler)
+      .subscribe(mock)
+    assert_equal 0, mock.messages.length
+    scheduler.advance_by 100
+    assert_equal 2, mock.messages.length
+  end
+
   def test_subscribe_on_default_scheduler_does_not_raise
     Rx::Observable.just(1).subscribe_on(Rx::DefaultScheduler.instance).subscribe
   end