Shutting down a DataBus automatically shuts down any contained Sessions.

Author: rickbw

src/main/java/crud/core/DataBus.java

@@ -117,4 +117,13 @@ public interface DataBus extends AsyncCloseable {
      */
     public @Nonnull TransactedSession startTransactedSession();
 
+    /**
+     * {@inheritDoc}
+     * <p/>
+     * Shutting down a {@link DataBus} implicitly shuts down all
+     * {@link Session}s created in the context of that {@code DataBus}.
+     */
+    @Override
+    public Observable<Void> shutdown();
+
 }

src/main/java/crud/file/FileSystem.java

@@ -24,6 +24,7 @@
 import crud.core.Session.Ordering;
 import crud.core.WritableResourceSet;
 import crud.implementer.AbstractDataBus;
+import crud.implementer.DataBusWorker;
 import crud.implementer.DefaultSession;
 
 
@@ -39,14 +40,18 @@ public final class FileSystem extends AbstractDataBus {
             Session.Ordering.ORDERED));
 
 
+    public FileSystem() {
+        super(DataBusWorker.create());
+    }
+
     @Override
     public Set<Session.Ordering> getSupportedSessionOrderings() {
         return supportedOrderings;
     }
 
     @Override
     protected Session doStartOrderedSession() throws Exception {
-        return new DefaultSession(Session.Ordering.ORDERED);
+        return new DefaultSession(getWorker(), Session.Ordering.ORDERED);
     }
 
     @Override

src/main/java/crud/implementer/AbstractDataBus.java

@@ -14,6 +14,7 @@
  */
 package crud.implementer;
 
+import java.util.Objects;
 import java.util.Set;
 
 import javax.annotation.Nonnull;
@@ -28,7 +29,9 @@
 import crud.core.TransactedSession;
 import crud.core.UnsupportedSessionOrderingException;
 import crud.core.WritableResourceSet;
+import rx.Observable;
 import rx.Observer;
+import rx.Subscriber;
 
 
 /**
@@ -41,7 +44,28 @@
  *
  * @author Rick Warren
  */
-public abstract class AbstractDataBus extends AbstractAsyncCloseable implements DataBus {
+public abstract class AbstractDataBus implements DataBus {
+
+    private @Nonnull final DataBusWorker worker;
+
+    private final Task<Void> shutdownTask = new Task<Void>() {
+        @Override
+        public void call(final Subscriber<? super Void> sub) throws Exception {
+            doShutdown();
+        }
+    };
+
+
+    /**
+     * Provided for use by subclasses and their clients.
+     * <p/>
+     * <em>ATTN</em>: This method is not declared {@code final} in order to
+     * support mocking in unit tests. Nevertheless, it is not intended for
+     * overriding, and the behavior in that case is unspecified.
+     */
+    public @Nonnull DataBusWorker getWorker() {
+        return this.worker;
+    }
 
     /**
      * Do nothing. Subclasses can override this method to do more.
@@ -161,8 +185,16 @@ public TransactedSession startTransactedSession() {
         }
     }
 
-    protected AbstractDataBus() {
-        // nothing to do
+    /**
+     * Subclasses should override {@link #doShutdown()} instead.
+     */
+    @Override
+    public final Observable<Void> shutdown() {
+        return this.worker.shutdown(this.shutdownTask);
+    }
+
+    protected AbstractDataBus(@Nonnull final DataBusWorker worker) {
+        this.worker = Objects.requireNonNull(worker);
     }
 
     /**
@@ -301,4 +333,20 @@ protected boolean isResourceSetAvailable(final WritableResourceSet.Id<?, ?, ?> i
         throw new AssertionError("isResourceSetAvailable() indicated ResourceSet available, but this method was not overridden");
     }
 
+    /**
+     * Subclasses should override this method to perform any shutdown task
+     * they have to do.
+     *
+     * By default, this method does nothing.
+     *
+     * @throws Exception    Subclasses may throw whatever they wish.
+     *                      Exceptions will be passed to
+     *                      {@link Observer#onError(Throwable)}.
+     *
+     * @see #shutdown()
+     */
+    protected void doShutdown() throws Exception {
+        // do nothing
+    }
+
 }

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

@@ -70,6 +70,14 @@ protected AbstractSession(
         this.ordering = Objects.requireNonNull(ordering);
     }
 
+    protected AbstractSession(
+            @Nonnull final DataBusWorker dataBusWorker,
+            @Nonnull final SessionWorker myWorker,
+            @Nonnull final Session.Ordering ordering) {
+        this(myWorker, ordering);
+        dataBusWorker.addPreShutdownHook(this);
+    }
+
     /**
      * Subclasses should override this method to perform any shutdown task
      * they have to do. It will be run in the context of the

src/main/java/crud/implementer/AbstractTransactedSession.java

@@ -45,20 +45,16 @@ public final Observable<Void> rollback() {
         return this.tx.rollback();
     }
 
-    protected AbstractTransactedSession(@Nonnull final SessionWorker worker) {
-        super(worker, Session.Ordering.TRANSACTED);
-
-        this.tx = new TransactionLifecycle(worker) {
-            @Override
-            protected void doCommit() throws Exception {
-                AbstractTransactedSession.this.doCommit();
-            }
+    protected AbstractTransactedSession(@Nonnull final SessionWorker myWorker) {
+        super(myWorker, Session.Ordering.TRANSACTED);
+        this.tx = initTransactionLifecycle(myWorker);
+    }
 
-            @Override
-            protected void doRollback() throws Exception {
-                AbstractTransactedSession.this.doRollback();
-            }
-        };
+    protected AbstractTransactedSession(
+            @Nonnull final DataBusWorker dataBusWorker,
+            @Nonnull final SessionWorker myWorker) {
+        super(dataBusWorker, myWorker, Session.Ordering.TRANSACTED);
+        this.tx = initTransactionLifecycle(myWorker);
     }
 
     /**
@@ -79,4 +75,18 @@ protected void doRollback() throws Exception {
      */
     protected abstract void doRollback() throws Exception;
 
+    private TransactionLifecycle initTransactionLifecycle(final SessionWorker myWorker) {
+        return new TransactionLifecycle(myWorker) {
+            @Override
+            protected void doCommit() throws Exception {
+                AbstractTransactedSession.this.doCommit();
+            }
+
+            @Override
+            protected void doRollback() throws Exception {
+                AbstractTransactedSession.this.doRollback();
+            }
+        };
+    }
+
 }

src/main/java/crud/implementer/DataBusWorker.java

@@ -0,0 +1,92 @@
+/* Copyright 2015 Rick Warren
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package crud.implementer;
+
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.ThreadSafe;
+
+import crud.core.DataBus;
+import crud.core.Session;
+import rx.Observable;
+import rx.Observer;
+import rx.schedulers.Schedulers;
+
+
+/**
+ * Manages lifecycle concerns on behalf of {@link DataBus} implementations.
+ *
+ * @author Rick Warren
+ */
+@ThreadSafe
+public class DataBusWorker {
+
+    private final WorkerDelegate delegate = new WorkerDelegate(Schedulers.immediate());
+
+
+    public static DataBusWorker create() {
+        return new DataBusWorker();
+    }
+
+    /**
+     * Add a listener that will be called prior to this {@link DataBusWorker}
+     * being {@link #shutdown(Task) shut down}. The listener will only be
+     * called once, just before shutting down.
+     *
+     * @throws IllegalStateException    If this method is called after
+     *              {@link #shutdown(Task)} itself.
+     */
+    public void addPreShutdownHook(@Nonnull final Session shutMeDown) {
+        this.delegate.addPreShutdownHook(shutMeDown);
+    }
+
+    /**
+     * Schedule the given task after shutting down the
+     * {@link #addPreShutdownHook(Session) pre-shutdown hooks}. The
+     * resulting {@link Observable} will emit one of the following:
+     * <ol>
+     *  <li>Any {@link Observer#onError(Throwable) error} emitted by the given
+     *      final task. (This task will be run after all
+     *      {@link #addPreShutdownHook(Session) pre-shutdown hooks}
+     *      have been shut down, but any error from it will receive precedence
+     *      with respect to reporting, because it is likely to be the most
+     *      relevant to the caller.)</li>
+     *  <li>The first {@link Observer#onError(Throwable) error}, if any,
+     *      emitted by any of the
+     *      {@link #addPreShutdownHook(Session) pre-shutdown hooks}.
+     *      (These hooks are shut down before the final task runs, but their
+     *      errors are not allowed to hide any errors from that task.)</li>
+     *  <li>{@link Observer#onCompleted()} once the termination is complete,
+     *      if no errors occurred.</li>
+     * </ol>
+     * <p/>
+     * This method only operates once. Calling it additional times has no
+     * effect, and will return an {@link Observable} that emits
+     * {@link Observer#onCompleted()}.
+     *
+     * @param finalTask The caller should perform any of its own cleanup in
+     *                  this task, scheduled here to avoid race conditions.
+     */
+    public Observable<Void> shutdown(final Task<Void> finalTask) {
+        return this.delegate.shutdown(finalTask);
+    }
+
+    private DataBusWorker() {
+        /* Private to prevent subclassing. We could just make the class final,
+         * but that would prevent mocking as well. Fortunately, Mockito can
+         * call a private constructor reflectively.
+         */
+    }
+
+}

src/main/java/crud/implementer/DefaultSession.java

@@ -14,6 +14,8 @@
  */
 package crud.implementer;
 
+import javax.annotation.Nonnull;
+
 import com.google.common.base.Preconditions;
 
 import crud.core.Session;
@@ -28,8 +30,10 @@
  */
 public class DefaultSession extends AbstractSession {
 
-    public DefaultSession(final Session.Ordering ordering) {
-        super(SessionWorker.create(), ordering);
+    public DefaultSession(
+            @Nonnull final DataBusWorker dataBusWorker,
+            @Nonnull final Session.Ordering ordering) {
+        super(dataBusWorker, SessionWorker.create(), ordering);
         Preconditions.checkArgument(
                 getOrdering() != Session.Ordering.TRANSACTED,
                 "Not a TransactedSession");

src/main/java/crud/jdbc/JdbcDataBus.java

@@ -30,13 +30,23 @@
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 
+import crud.core.DataBus;
 import crud.core.ReadableResourceSet;
 import crud.core.Session;
 import crud.core.TransactedSession;
 import crud.core.WritableResourceSet;
 import crud.implementer.AbstractDataBus;
+import crud.implementer.DataBusWorker;
 
 
+/**
+ * A {@link DataBus} implementation that wraps a JDBC {@link DataSource}.
+ * It supports {@link TransactedSession}s by means of the usual JDBC
+ * transactions. It also supports ordered {@link Session}s by means of
+ * {@link Connection#setAutoCommit(boolean) auto-commit}.
+ *
+ * @author Rick Warren
+ */
 public class JdbcDataBus extends AbstractDataBus {
 
     private static final Logger log = LoggerFactory.getLogger(JdbcDataBus.class);
@@ -55,15 +65,16 @@ public JdbcDataBus(@Nonnull final DataSource dataSource) {
 
     public JdbcDataBus(
             @Nonnull final DataSource dataSource,
-            final String username,
-            final String password) {
+            @Nonnull final String username,
+            @Nonnull final String password) {
         this(dataSource, Optional.of(username), Optional.of(password));
     }
 
     private JdbcDataBus(
             @Nonnull final DataSource dataSource,
             final Optional<String> username,
             final Optional<String> password) {
+        super(DataBusWorker.create());
         this.dataSource = Objects.requireNonNull(dataSource);
         this.username = Objects.requireNonNull(username);
         this.password = Objects.requireNonNull(password);
@@ -82,15 +93,15 @@ public Set<Session.Ordering> getSupportedSessionOrderings() {
     protected Session doStartOrderedSession() throws SQLException {
         final Connection connection = getConnection();
         connection.setAutoCommit(true);
-        return new JdbcSession(connection);
+        return new JdbcSession(getWorker(), connection);
     }
 
     @Override
     @SuppressWarnings("resource")
     protected TransactedSession doStartTransactedSession() throws SQLException {
         final Connection connection = getConnection();
         connection.setAutoCommit(false);
-        return new JdbcTransactedSession(connection);
+        return new JdbcTransactedSession(getWorker(), connection);
     }
 
     @Override

src/main/java/crud/jdbc/JdbcSession.java

@@ -24,6 +24,7 @@
 import crud.core.Session;
 import crud.core.WritableResource;
 import crud.implementer.AbstractSession;
+import crud.implementer.DataBusWorker;
 import crud.implementer.SessionWorker;
 
 
@@ -32,8 +33,10 @@
     private @Nonnull final Connection connection;
 
 
-    public JdbcSession(@Nonnull final Connection connection) {
-        super(SessionWorker.create(), Session.Ordering.ORDERED);
+    public JdbcSession(
+            @Nonnull final DataBusWorker dataBusWorker,
+            @Nonnull final Connection connection) {
+        super(dataBusWorker, SessionWorker.create(), Session.Ordering.ORDERED);
         this.connection = Objects.requireNonNull(connection);
     }