title: How to write operators in Reactor 3?
body: <p>I've implemented a Reactor operator by implementing <code>org.reactivestreams.Publisher</code>, as you can see below. However, I wonder whether this is the <em>right way ™</em> to use Reactor. Implementing a subscriber by hand looks a bit cumbersome. And the <a href="http://projectreactor.io/docs/core/release/api/reactor/core/publisher/Operators.html" rel="nofollow noreferrer">Operators</a> class seems not to be helpful in this regard.</p>
  
  <pre><code>class MyOperator implements Publisher&lt;Integer&gt; {
  
      private final Publisher&lt;Integer&gt; source;
  
      public MyOperator(Publisher&lt;Integer&gt; source) {
          this.source = source;
      }
  
      @Override
      public void subscribe(Subscriber&lt;? super Integer&gt; target) {
          final Subscriber&lt;Integer&gt; wrappedSubscriber = createSubscriber(target);
          source.subscribe(wrappedSubscriber);
      }
  
      private Subscriber&lt;Integer&gt; createSubscriber(Subscriber&lt;? super Integer&gt; target) {
          return new Subscriber&lt;Integer&gt;() {
              private Subscription subscription;
  
              @Override
              public void onSubscribe(Subscription subscription) {
                  this.subscription = subscription;
                  subscription.request(Long.MAX_VALUE);
              }
  
              @Override
              public void onNext(Integer integer) {
                  target.onNext(integer + 1); // actual behaviour
                  subscription.request(Long.MAX_VALUE);
              }
  
              @Override
              public void onError(Throwable t) {
                  target.onError(t);
              }
  
              @Override
              public void onComplete() {
                  target.onComplete();
              }
          };
      }
  }
  </code></pre>
  
  <p>Or is the following example the <em>right way ™</em>?</p>
  
  <pre><code>class MyCompactOperator implements Publisher&lt;Integer&gt; {
  
      final Flux&lt;Integer&gt; flux;
  
      public MyCompactOperator(Publisher&lt;Integer&gt; source) {
          flux = Flux.from(source).map(number -&gt; number + 1);
      }
  
      @Override
      public void subscribe(Subscriber&lt;? super Integer&gt; subscriber) {
          flux.subscribe(subscriber);
      }
  }
  </code></pre>
  
  <p>At least this requires a lot less code.</p>
  
  <p>And variant 3 with a <code>Flux</code> as source, as suggest by Simon Baslé:</p>
  
  <pre><code>class MyFluxOperator extends Flux&lt;Integer&gt; {
  
      private final Flux&lt;Integer&gt; source;
  
      public MyFluxOperator(Flux&lt;Integer&gt; source) {
          this.source = source;
      }
  
      @Override
      public void subscribe(Subscriber&lt;? super Integer&gt; subscriber) {
          source.map(number -&gt; number + 1).subscribe(subscriber);
      }
  }
  </code></pre>
  
  <p>All implementations are working as expected:</p>
  
  <pre><code>Flux&lt;Integer&gt; source = Flux.just(1, 2, 3, 4, 5);
  Flux.from(new MyOperator(source)).subscribe(System.out::println);
  
  // for variant 3
  new MyFluxOperator(source).subscribe(System.out::println);
  </code></pre>
  
  <p>I've used a <code>Flux</code> in the second line to avoid the implementation of another subscriber.</p>
  
  <p>Output:</p>
  
  <pre><code>2
  3
  4
  5
  6
  </code></pre>
  
  <p>Questions:</p>
  
  <ul>
  <li>Is there something missing in my implementation? </li>
  <li>Is there a better way (with less code, better error handling or whatever) to implement operators in Reactor 3?</li>
  <li>Are there relevant functional or non-functional differences between both approaches?</li>
  </ul>