subscribeOn and publishOn

Author: meowpunch

src/main/java/concurrency/reactive/SchedulerEx.java

@@ -1,19 +1,25 @@
 package concurrency.reactive;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class SchedulerEx {
+
   private static final Logger LOG = LoggerFactory.getLogger(SchedulerEx.class.getName());
 
   public static void main(String[] args) {
+    LOG.debug("enter");
+
     Publisher<Integer> pub = subscriber ->
         subscriber.onSubscribe(new Subscription() {
           @Override
           public void request(long n) {
+            LOG.debug("request");
             subscriber.onNext(1);
             subscriber.onNext(2);
             subscriber.onNext(3);
@@ -28,8 +34,44 @@ public void cancel() {
           }
         });
 
+    // subscribeOn
+    Publisher<Integer> subOnPub = subscriber ->
+        Executors.newSingleThreadExecutor().execute(() -> pub.subscribe(subscriber));
+
+    // publishOn
+    Publisher<Integer> pubOnPub = sub ->
+        pub.subscribe(new Subscriber<>() {
+          final ExecutorService executorService = Executors.newSingleThreadExecutor();
+
+          @Override
+          public void onSubscribe(Subscription s) {
+            sub.onSubscribe(s);
+          }
+
+          @Override
+          public void onNext(Integer integer) {
+            executorService.execute(() -> sub.onNext(integer));
+          }
+
+          @Override
+          public void onError(Throwable t) {
+            executorService.execute(() -> sub.onError(t));
+          }
+
+          @Override
+          public void onComplete() {
+            executorService.execute(sub::onComplete);
+          }
+        });
+
+    subOnPub.subscribe(getSubscriber());
+    pubOnPub.subscribe(getSubscriber());
+
+    LOG.debug("exit");
+  }
 
-    pub.subscribe(new Subscriber<>() {
+  private static Subscriber<Integer> getSubscriber() {
+    return new Subscriber<>() {
       @Override
       public void onSubscribe(Subscription s) {
         LOG.debug("onSubscribe");
@@ -50,6 +92,6 @@ public void onError(Throwable t) {
       public void onComplete() {
         LOG.debug("onComplete");
       }
-    });
+    };
   }
 }