DelegateSubscriber

Author: meowpunch

src/main/java/concurrency/reactive/DelegateSubscriber.java

@@ -0,0 +1,29 @@
+package concurrency.reactive;
+
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+abstract class DelegateSubscriber<T> implements Subscriber<T> {
+
+  private final Subscriber<T> subscriber;
+
+  DelegateSubscriber(Subscriber<T> subscriber) {
+    this.subscriber = subscriber;
+  }
+
+  @Override
+  public void onSubscribe(Subscription s) {
+    subscriber.onSubscribe(s);
+  }
+
+  @Override
+  public void onError(Throwable t) {
+    subscriber.onError(t);
+  }
+
+  @Override
+  public void onComplete() {
+    subscriber.onComplete();
+  }
+}

src/main/java/concurrency/reactive/PubSub.java

@@ -23,26 +23,11 @@ private static Publisher<Integer> mapPub(Publisher<Integer> publisher, Function<
     return new Publisher<>() {
       @Override
       public void subscribe(Subscriber<? super Integer> subscriber) {
-        publisher.subscribe(new Subscriber<Integer>() {
-          @Override
-          public void onSubscribe(Subscription subscription) {
-            subscriber.onSubscribe(subscription);
-          }
-
+        publisher.subscribe(new DelegateSubscriber<Integer>((Subscriber<Integer>) subscriber) {
           @Override
           public void onNext(Integer integer) {
             subscriber.onNext(mapFunction.apply(integer));
           }
-
-          @Override
-          public void onError(Throwable t) {
-            subscriber.onError(t);
-          }
-
-          @Override
-          public void onComplete() {
-            subscriber.onComplete();
-          }
         });
       }
     };