001/************************************************************************ 002* Licensed under Public Domain (CC0) * 003* * 004* To the extent possible under law, the person who associated CC0 with * 005* this code has waived all copyright and related or neighboring * 006* rights to this code. * 007* * 008* You should have received a copy of the CC0 legalcode along with this * 009* work. If not, see <http://creativecommons.org/publicdomain/zero/1.0/>.* 010************************************************************************/ 011 012package org.reactivestreams.example.unicast; 013 014import org.reactivestreams.Subscriber; 015import org.reactivestreams.Subscription; 016 017import java.util.concurrent.Executor; 018import java.util.concurrent.atomic.AtomicBoolean; 019import java.util.concurrent.ConcurrentLinkedQueue; 020 021/** 022 * AsyncSubscriber is an implementation of Reactive Streams `Subscriber`, 023 * it runs asynchronously (on an Executor), requests one element 024 * at a time, and invokes a user-defined method to process each element. 025 * 026 * NOTE: The code below uses a lot of try-catches to show the reader where exceptions can be expected, and where they are forbidden. 027 */ 028public abstract class AsyncSubscriber<T> implements Subscriber<T>, Runnable { 029 030 // Signal represents the asynchronous protocol between the Publisher and Subscriber 031 private static interface Signal {} 032 033 private enum OnComplete implements Signal { Instance; } 034 035 private static class OnError implements Signal { 036 public final Throwable error; 037 public OnError(final Throwable error) { this.error = error; } 038 } 039 040 private static class OnNext<T> implements Signal { 041 public final T next; 042 public OnNext(final T next) { this.next = next; } 043 } 044 045 private static class OnSubscribe implements Signal { 046 public final Subscription subscription; 047 public OnSubscribe(final Subscription subscription) { this.subscription = subscription; } 048 } 049 050 private Subscription subscription; // Obeying rule 3.1, we make this private! 051 private boolean done; // It's useful to keep track of whether this Subscriber is done or not 052 private final Executor executor; // This is the Executor we'll use to be asynchronous, obeying rule 2.2 053 054 // Only one constructor, and it's only accessible for the subclasses 055 protected AsyncSubscriber(Executor executor) { 056 if (executor == null) throw null; 057 this.executor = executor; 058 } 059 060 // Showcases a convenience method to idempotently marking the Subscriber as "done", so we don't want to process more elements 061 // herefor we also need to cancel our `Subscription`. 062 private final void done() { 063 //On this line we could add a guard against `!done`, but since rule 3.7 says that `Subscription.cancel()` is idempotent, we don't need to. 064 done = true; // If `whenNext` throws an exception, let's consider ourselves done (not accepting more elements) 065 if (subscription != null) { // If we are bailing out before we got a `Subscription` there's little need for cancelling it. 066 try { 067 subscription.cancel(); // Cancel the subscription 068 } catch(final Throwable t) { 069 //Subscription.cancel is not allowed to throw an exception, according to rule 3.15 070 (new IllegalStateException(subscription + " violated the Reactive Streams rule 3.15 by throwing an exception from cancel.", t)).printStackTrace(System.err); 071 } 072 } 073 } 074 075 // This method is invoked when the OnNext signals arrive 076 // Returns whether more elements are desired or not, and if no more elements are desired, 077 // for convenience. 078 protected abstract boolean whenNext(final T element); 079 080 // This method is invoked when the OnComplete signal arrives 081 // override this method to implement your own custom onComplete logic. 082 protected void whenComplete() { } 083 084 // This method is invoked if the OnError signal arrives 085 // override this method to implement your own custom onError logic. 086 protected void whenError(Throwable error) { } 087 088 private final void handleOnSubscribe(final Subscription s) { 089 if (s == null) { 090 // Getting a null `Subscription` here is not valid so lets just ignore it. 091 } else if (subscription != null) { // If someone has made a mistake and added this Subscriber multiple times, let's handle it gracefully 092 try { 093 s.cancel(); // Cancel the additional subscription to follow rule 2.5 094 } catch(final Throwable t) { 095 //Subscription.cancel is not allowed to throw an exception, according to rule 3.15 096 (new IllegalStateException(s + " violated the Reactive Streams rule 3.15 by throwing an exception from cancel.", t)).printStackTrace(System.err); 097 } 098 } else { 099 // We have to assign it locally before we use it, if we want to be a synchronous `Subscriber` 100 // Because according to rule 3.10, the Subscription is allowed to call `onNext` synchronously from within `request` 101 subscription = s; 102 try { 103 // If we want elements, according to rule 2.1 we need to call `request` 104 // And, according to rule 3.2 we are allowed to call this synchronously from within the `onSubscribe` method 105 s.request(1); // Our Subscriber is unbuffered and modest, it requests one element at a time 106 } catch(final Throwable t) { 107 // Subscription.request is not allowed to throw according to rule 3.16 108 (new IllegalStateException(s + " violated the Reactive Streams rule 3.16 by throwing an exception from request.", t)).printStackTrace(System.err); 109 } 110 } 111 } 112 113 private final void handleOnNext(final T element) { 114 if (!done) { // If we aren't already done 115 if(subscription == null) { // Technically this check is not needed, since we are expecting Publishers to conform to the spec 116 // Check for spec violation of 2.1 and 1.09 117 (new IllegalStateException("Someone violated the Reactive Streams rule 1.09 and 2.1 by signalling OnNext before `Subscription.request`. (no Subscription)")).printStackTrace(System.err); 118 } else { 119 try { 120 if (whenNext(element)) { 121 try { 122 subscription.request(1); // Our Subscriber is unbuffered and modest, it requests one element at a time 123 } catch(final Throwable t) { 124 // Subscription.request is not allowed to throw according to rule 3.16 125 (new IllegalStateException(subscription + " violated the Reactive Streams rule 3.16 by throwing an exception from request.", t)).printStackTrace(System.err); 126 } 127 } else { 128 done(); // This is legal according to rule 2.6 129 } 130 } catch(final Throwable t) { 131 done(); 132 try { 133 onError(t); 134 } catch(final Throwable t2) { 135 //Subscriber.onError is not allowed to throw an exception, according to rule 2.13 136 (new IllegalStateException(this + " violated the Reactive Streams rule 2.13 by throwing an exception from onError.", t2)).printStackTrace(System.err); 137 } 138 } 139 } 140 } 141 } 142 143 // Here it is important that we do not violate 2.2 and 2.3 by calling methods on the `Subscription` or `Publisher` 144 private void handleOnComplete() { 145 if (subscription == null) { // Technically this check is not needed, since we are expecting Publishers to conform to the spec 146 // Publisher is not allowed to signal onComplete before onSubscribe according to rule 1.09 147 (new IllegalStateException("Publisher violated the Reactive Streams rule 1.09 signalling onComplete prior to onSubscribe.")).printStackTrace(System.err); 148 } else { 149 done = true; // Obey rule 2.4 150 whenComplete(); 151 } 152 } 153 154 // Here it is important that we do not violate 2.2 and 2.3 by calling methods on the `Subscription` or `Publisher` 155 private void handleOnError(final Throwable error) { 156 if (subscription == null) { // Technically this check is not needed, since we are expecting Publishers to conform to the spec 157 // Publisher is not allowed to signal onError before onSubscribe according to rule 1.09 158 (new IllegalStateException("Publisher violated the Reactive Streams rule 1.09 signalling onError prior to onSubscribe.")).printStackTrace(System.err); 159 } else { 160 done = true; // Obey rule 2.4 161 whenError(error); 162 } 163 } 164 165 // We implement the OnX methods on `Subscriber` to send Signals that we will process asycnhronously, but only one at a time 166 167 @Override public final void onSubscribe(final Subscription s) { 168 // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Subscription` is `null` 169 if (s == null) throw null; 170 171 signal(new OnSubscribe(s)); 172 } 173 174 @Override public final void onNext(final T element) { 175 // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `element` is `null` 176 if (element == null) throw null; 177 178 signal(new OnNext<T>(element)); 179 } 180 181 @Override public final void onError(final Throwable t) { 182 // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Throwable` is `null` 183 if (t == null) throw null; 184 185 signal(new OnError(t)); 186 } 187 188 @Override public final void onComplete() { 189 signal(OnComplete.Instance); 190 } 191 192 // This `ConcurrentLinkedQueue` will track signals that are sent to this `Subscriber`, like `OnComplete` and `OnNext` , 193 // and obeying rule 2.11 194 private final ConcurrentLinkedQueue<Signal> inboundSignals = new ConcurrentLinkedQueue<Signal>(); 195 196 // We are using this `AtomicBoolean` to make sure that this `Subscriber` doesn't run concurrently with itself, 197 // obeying rule 2.7 and 2.11 198 private final AtomicBoolean on = new AtomicBoolean(false); 199 200 @SuppressWarnings("unchecked") 201 @Override public final void run() { 202 if(on.get()) { // establishes a happens-before relationship with the end of the previous run 203 try { 204 final Signal s = inboundSignals.poll(); // We take a signal off the queue 205 if (!done) { // If we're done, we shouldn't process any more signals, obeying rule 2.8 206 // Below we simply unpack the `Signal`s and invoke the corresponding methods 207 if (s instanceof OnNext<?>) 208 handleOnNext(((OnNext<T>)s).next); 209 else if (s instanceof OnSubscribe) 210 handleOnSubscribe(((OnSubscribe)s).subscription); 211 else if (s instanceof OnError) // We are always able to handle OnError, obeying rule 2.10 212 handleOnError(((OnError)s).error); 213 else if (s == OnComplete.Instance) // We are always able to handle OnComplete, obeying rule 2.9 214 handleOnComplete(); 215 } 216 } finally { 217 on.set(false); // establishes a happens-before relationship with the beginning of the next run 218 if(!inboundSignals.isEmpty()) // If we still have signals to process 219 tryScheduleToExecute(); // Then we try to schedule ourselves to execute again 220 } 221 } 222 } 223 224 // What `signal` does is that it sends signals to the `Subscription` asynchronously 225 private void signal(final Signal signal) { 226 if (inboundSignals.offer(signal)) // No need to null-check here as ConcurrentLinkedQueue does this for us 227 tryScheduleToExecute(); // Then we try to schedule it for execution, if it isn't already 228 } 229 230 // This method makes sure that this `Subscriber` is only executing on one Thread at a time 231 private final void tryScheduleToExecute() { 232 if(on.compareAndSet(false, true)) { 233 try { 234 executor.execute(this); 235 } catch(Throwable t) { // If we can't run on the `Executor`, we need to fail gracefully and not violate rule 2.13 236 if (!done) { 237 try { 238 done(); // First of all, this failure is not recoverable, so we need to cancel our subscription 239 } finally { 240 inboundSignals.clear(); // We're not going to need these anymore 241 // This subscription is cancelled by now, but letting the Subscriber become schedulable again means 242 // that we can drain the inboundSignals queue if anything arrives after clearing 243 on.set(false); 244 } 245 } 246 } 247 } 248 } 249}