Schedule Session work using Rx Schedulers; don't reinvent the wheel.

Also improved error handling.

Author: rickbw

src/main/java/crud/implementer/AbstractDataSink.java

@@ -14,8 +14,6 @@
  */
 package crud.implementer;
 
-import java.util.concurrent.Callable;
-
 import javax.annotation.Nonnull;
 
 import crud.core.DataSink;
@@ -39,13 +37,12 @@ public abstract class AbstractDataSink<E, R>
      */
     @Override
     public Observable<R> write(final E value) {
-        final Observable<R> result = Observable.create(new SubmitWriteOnSubscribe(value)).cache();
-        /* Start a subscription now, so that the write is scheduled
-         * immediately. The no-argument subscribe() does not handle errors,
-         * so materialize() to prevent it from seeing any.
-         */
-        result.materialize().subscribe();
-        return result;
+        return getWorker().scheduleHot(new SessionWorker.Task<R>() {
+            @Override
+            public void call(final Subscriber<? super R> sub) throws Exception {
+                doWrite(value, sub);
+            }
+        });
     }
 
     protected AbstractDataSink(@Nonnull final SessionWorker worker) {
@@ -68,29 +65,13 @@ protected AbstractDataSink(@Nonnull final SessionWorker worker) {
      * @param writeMe   The value to write.
      * @param resultSub The subscriber to which the result of the write should
      *                  be reported.
+     *
+     * @throws Exception    Subclasses may throw whatever they wish.
+     *                      Exceptions will be passed to
+     *                      {@link Observer#onError(Throwable)}.
      */
-    protected void doWrite(final E writeMe, final Subscriber<? super R> resultSub) {
+    protected void doWrite(final E writeMe, final Subscriber<? super R> resultSub) throws Exception {
         throw new AssertionError("Must override this method if not overriding write()");
     }
 
-
-    private final class SubmitWriteOnSubscribe implements Observable.OnSubscribe<R> {
-        private final E writeMe;
-
-        public SubmitWriteOnSubscribe(final E writeMe) {
-            this.writeMe = writeMe;
-        }
-
-        @Override
-        public void call(final Subscriber<? super R> sub) {
-            submit(new Callable<Void>() {
-                @Override
-                public Void call() throws Exception {
-                    doWrite(SubmitWriteOnSubscribe.this.writeMe, sub);
-                    return null;
-                }
-            });
-        }
-    }
-
 }

src/main/java/crud/implementer/AbstractDataSource.java

@@ -14,8 +14,6 @@
  */
 package crud.implementer;
 
-import java.util.concurrent.Callable;
-
 import javax.annotation.Nonnull;
 
 import crud.core.DataSource;
@@ -28,16 +26,10 @@ public abstract class AbstractDataSource<E> extends AbstractSessionParticipant i
 
     @Override
     public Observable<E> read() {
-        return Observable.create(new Observable.OnSubscribe<E>() {
+        return getWorker().scheduleCold(new SessionWorker.Task<E>() {
             @Override
-            public void call(final Subscriber<? super E> sub) {
-                submit(new Callable<Void>() {
-                    @Override
-                    public Void call() throws Exception {
-                        onReadSubscribe(sub);
-                        return null;
-                    }
-                });
+            public void call(final Subscriber<? super E> sub) throws Exception {
+                onReadSubscribe(sub);
             }
         });
     }
@@ -58,8 +50,12 @@ protected AbstractDataSource(@Nonnull final SessionWorker worker) {
      * directly, then there is no need to override this method.
      *
      * @param sub   The {@link Subscriber} to notify of the elements being read.
+     *
+     * @throws Exception    Subclasses may throw whatever they wish.
+     *                      Exceptions will be passed to
+     *                      {@link Observer#onError(Throwable)}.
      */
-    protected void onReadSubscribe(final Subscriber<? super E> sub) {
+    protected void onReadSubscribe(final Subscriber<? super E> sub) throws Exception {
         throw new AssertionError("Must override this method if not overriding read()");
     }
 

src/main/java/crud/implementer/AbstractSession.java

@@ -15,26 +15,25 @@
 package crud.implementer;
 
 import java.util.Objects;
-import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 
 import javax.annotation.Nonnull;
 
 import crud.core.Session;
 import rx.Observable;
 import rx.Observer;
+import rx.Subscriber;
 
 
 public abstract class AbstractSession extends AbstractAsyncCloseable implements Session {
 
     private @Nonnull final SessionWorker worker;
     private @Nonnull final Session.Ordering ordering;
 
-    private final Callable<Void> shutdownTask = new Callable<Void>() {
+    private final SessionWorker.Task<Void> shutdownTask = new SessionWorker.Task<Void>() {
         @Override
-        public Void call() throws Exception {
+        public void call(final Subscriber<? super Void> sub) throws Exception {
             doShutdown();
-            return null;
         }
     };
 

src/main/java/crud/implementer/AbstractSessionParticipant.java

@@ -15,7 +15,6 @@
 package crud.implementer;
 
 import java.util.Objects;
-import java.util.concurrent.Callable;
 
 import javax.annotation.Nonnull;
 
@@ -24,6 +23,7 @@
 import crud.core.DataSource;
 import rx.Observable;
 import rx.Observer;
+import rx.Subscriber;
 
 
 /**
@@ -42,11 +42,10 @@ public abstract class AbstractSessionParticipant implements AsyncCloseable {
      */
     @Override
     public Observable<Void> shutdown() {
-        return submit(new Callable<Void>() {
+        return this.worker.scheduleHot(new SessionWorker.Task<Void>() {
             @Override
-            public Void call() throws Exception {
+            public void call(final Subscriber<? super Void> sub) throws Exception {
                 doShutdown();
-                return null;
             }
         });
     }
@@ -72,8 +71,8 @@ protected void doShutdown() throws Exception {
         // do nothing
     }
 
-    protected final Observable<Void> submit(final Callable<Void> task) {
-        return this.worker.submit(task);
+    public final @Nonnull SessionWorker getWorker() {
+        return this.worker;
     }
 
 }

src/main/java/crud/implementer/SessionWorker.java

@@ -14,10 +14,8 @@
  */
 package crud.implementer;
 
-import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -28,7 +26,10 @@
 import crud.core.Session;
 import rx.Observable;
 import rx.Observer;
+import rx.Scheduler;
 import rx.Subscriber;
+import rx.functions.Action1;
+import rx.schedulers.Schedulers;
 
 
 /**
@@ -41,43 +42,81 @@ public final class SessionWorker {
 
     private final AtomicBoolean stopped = new AtomicBoolean(false);
     private final ExecutorService executor = Executors.newSingleThreadExecutor();
+    private final Scheduler scheduler = Schedulers.from(this.executor);
 
 
     /**
-     * Schedule the given task to run on this worker's asynchronous thread.
-     * Return a {@link Observable#cache() cached} {@link Observable} that
-     * will emit the success or error result of the task to all subscribers.
-     * <p/>
-     * If {@link #shutdown(Callable, long, TimeUnit)} was already called, the
-     * resulting {@link Observable} will emit
-     * {@link RejectedExecutionException}.
+     * Wrap the given {@link Action1 action} in an {@link Observable} and
+     * schedule its subscription to run on the {@link Scheduler} encapsulated
+     * by this {@link SessionWorker worker}. This method only creates the
+     * {@link Observable}; it does not subscribe to it.
+     *
+     * @see #subscribeHot(Observable)
      */
-    public Observable<Void> submit(@Nonnull final Callable<Void> task) {
-        if (!this.stopped.get()) {
-            return doSubmit(task);
-        } else {
-            return Observable.error(new RejectedExecutionException("Already stopped"));
-        }
+    public <T> Observable<T> scheduleCold(@Nonnull final Task<T> task) {
+        final Observable.OnSubscribe<T> onSubscribe = new Observable.OnSubscribe<T>() {
+            @Override
+            public void call(final Subscriber<? super T> sub) {
+                try {
+                    task.call(sub);
+                    sub.onCompleted();
+                } catch (final MiddlewareException mx) {
+                    sub.onError(mx);
+                } catch (final Exception ex) {
+                    sub.onError(new MiddlewareException(ex.getMessage(), ex));
+                }
+            }
+        };
+        return Observable.create(onSubscribe).subscribeOn(this.scheduler);
+    }
+
+    /**
+     * Wrap the given {@link Action1 action} in an {@link Observable} and
+     * schedule its subscription to run on the {@link Scheduler} encapsulated
+     * by this {@link SessionWorker worker}. Then
+     * {@link Observable#cache() cache} the result, and begin the
+     * subscription. This allows the subscription action to begin immediately,
+     * but allows the caller to see every resulting value. Note that the use
+     * of cache() assumes that the number of those results is relatively
+     * small.
+     *
+     * @see #scheduleCold(Task)
+     * @see #subscribeHot(Observable)
+     */
+    public <T> Observable<T> scheduleHot(@Nonnull final Task<T> task) {
+        final Observable<T> result = scheduleCold(task).cache();
+        doSubscribeHot(result);
+        return result;
+    }
+
+    /**
+     * Start subscribing to the given {@link Observable} on the
+     * {@link Scheduler} encapsulated by this {@link SessionWorker worker}.
+     * Note that the caller may miss some results if it does not e.g.
+     * {@link Observable#share() share} the input Observable before calling
+     * this method.
+     */
+    public <T> void subscribeHot(final Observable<T> obs) {
+        doSubscribeHot(obs.subscribeOn(this.scheduler));
     }
 
     /**
-     * Stop accepting new tasks via {@link #submit(Callable)}, and initiate
+     * {@link #scheduleHot(Task) Schedule} the given
+     * task, then stop accepting any new tasks, and initiate
      * the termination of this worker's background thread. The resulting
      * {@link Observable} will emit {@link Observer#onCompleted()} once the
      * termination is complete, a {@link TimeoutException} if it fails to
      * complete within the given duration, or {@link InterruptedException} if
      * the shutdown is interrupted.
      *
      * @param finalTask The caller should perform any of its own cleanup in
-     *                  this task, scheduled here as opposed to in
-     *                  {@link #submit(Callable)} to avoid race conditions on
-     *                  shutdown.
+     *                  this task, scheduled here to avoid race conditions.
      */
     public Observable<Void> shutdown(
-            final @Nonnull Callable<Void> finalTask,
+            @Nonnull final Task<Void> finalTask,
             final long waitDuration, @Nonnull final TimeUnit waitUnit) {
         if (!this.stopped.getAndSet(true)) {
-            final Observable<Void> taskResult = doSubmit(finalTask);
+            final Observable<Void> taskResult = scheduleHot(finalTask);
             this.executor.shutdown();   // non-blocking
             final Observable<Void> await = Observable.create(new Observable.OnSubscribe<Void>() {
                 @Override
@@ -112,53 +151,23 @@ public void call(final Subscriber<? super Void> sub) {
         }
     }
 
-    private Observable<Void> doSubmit(final Callable<Void> task) {
-        final Observable<Void> result = Observable.create(actionOf(task)).cache();
-        /* Start a subscription now, so that the given task is immediately
-         * scheduled. The no-argument subscribe() does not handle errors, so
-         * materialize so that it won't see any.
+    private static <T> void doSubscribeHot(final Observable<T> obs) {
+        /* The no-argument subscribe() does not handle errors, so
+         * materialize() so that it won't see any.
          */
-        result.materialize().subscribe();
-        return result;
+        obs.materialize().subscribe();
     }
 
-    private Observable.OnSubscribe<Void> actionOf(final Callable<Void> task) {
-        return new Observable.OnSubscribe<Void>() {
-            @Override
-            public void call(final Subscriber<? super Void> sub) {
-                SessionWorker.this.executor.submit(runAndNotify(task, sub));
-            }
-        };
-    }
 
-    private static Runnable runAndNotify(
-            final Callable<Void> runMe,
-            final Subscriber<? super Void> notifyMe) {
-        return new Runnable() {
-            @Override
-            public void run() {
-                try {
-                    runMe.call();
-                    notifyMe.onCompleted();
-                } catch (final RuntimeException rex) {
-                    /* RuntimeExceptions are assumed to indicate program bugs.
-                     * Report them to the subscriber as-is. This clause will
-                     * also include MiddlewareExceptions as-is, which is
-                     * desirable.
-                     */
-                    notifyMe.onError(rex);
-                } catch (final Exception ex) {
-                    /* Any checked exception is assumed to represent a
-                     * middleware-specific failure condition, and so is mapped
-                     * to MiddlewareException.
-                     *
-                     * TODO: Provide a pluggable exception-translation
-                     * capability.
-                     */
-                    notifyMe.onError(new MiddlewareException(ex.getMessage(), ex));
-                }
-            }
-        };
+    /**
+     * An asynchronous task to be scheduled by a {@link SessionWorker}.
+     * It will be wrapped by an instance of {@link rx.Observable.OnSubscribe},
+     * and execution of {@link Observer#onCompleted()} and
+     * {@link Observer#onError(Throwable)} will be taken care of on behalf of
+     * the task; it need only invoke {@link Observer#onNext(Object)}.
+     */
+    public static interface Task<T> {
+        void call(Subscriber<? super T> sub) throws Exception;
     }
 
 }