Take advantage of abstract base classes in JMS implementation

Author: rickbw

src/main/java/crud/jms/MessageConsumerDataSource.java

@@ -25,23 +25,24 @@
 
 import crud.core.DataSource;
 import crud.core.MiddlewareException;
+import crud.implementer.SessionWorker;
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
 
 
 /*package*/ final class MessageConsumerDataSource<M extends Message> implements DataSource<M> {
 
-    private @Nonnull final SessionWrapper session;
+    private @Nonnull final SessionWorker worker;
     private @Nonnull final MessageConsumer consumer;
     private @Nonnull final Observable<M> hotObservable;
 
 
     public MessageConsumerDataSource(
-            @Nonnull final SessionWrapper session,
+            @Nonnull final SessionWorker worker,
             @Nonnull final MessageConsumer consumer,
             @Nonnull final Class<M> messageType) {
-        this.session = Objects.requireNonNull(session);
+        this.worker = Objects.requireNonNull(worker);
         this.consumer = Objects.requireNonNull(consumer);
 
         this.hotObservable = Observable.create(new MessageListenerToSubscriberHandoff(messageType)).share();
@@ -59,7 +60,7 @@ public Observable<M> read() {
 
     @Override
     public Observable<Void> shutdown() {
-        return this.session.submit(new Callable<Void>() {
+        return this.worker.submit(new Callable<Void>() {
             @Override
             public Void call() throws JMSException {
                 /* TODO: Should this result in an onCompleted() to the
@@ -84,6 +85,7 @@ public MessageListenerToSubscriberHandoff(final Class<M> messageType) {
         public void call(final Subscriber<? super M> sub) {
             try {
                 sub.add(new MessageListenerRemover());
+                // FIXME: Adding listener in wrong thread?
                 MessageConsumerDataSource.this.consumer.setMessageListener(new MessageListener() {
                     @Override
                     public void onMessage(final Message message) {

src/main/java/crud/jms/MessageConsumingDataSet.java

@@ -55,7 +55,7 @@ public DataSource<M> dataSource(final String key, final Session session) {
             final MessageConsumer messageConsumer = key.isEmpty()
                     ? realSession.createConsumer(this.destination)
                     : realSession.createConsumer(this.destination, key);
-            return new MessageConsumerDataSource<>(sessionImpl, messageConsumer, this.id.getElementType());
+            return new MessageConsumerDataSource<>(sessionImpl.worker(), messageConsumer, this.id.getElementType());
         } catch (final JMSException jx) {
             throw new MiddlewareException(jx.getMessage(), jx);
         }

src/main/java/crud/jms/MessageProducerDataSink.java

@@ -25,20 +25,21 @@
 
 import crud.core.DataSink;
 import crud.core.MiddlewareException;
+import crud.implementer.SessionWorker;
 import rx.Observable;
 import rx.Subscriber;
 
 
 /*package*/ final class MessageProducerDataSink<M extends Message> implements DataSink<M, Void> {
 
-    private @Nonnull final SessionWrapper session;
+    private @Nonnull final SessionWorker worker;
     private @Nonnull final MessageProducer producer;
 
 
     public MessageProducerDataSink(
-            @Nonnull final SessionWrapper session,
+            @Nonnull final SessionWorker worker,
             @Nonnull final MessageProducer producer) {
-        this.session = Objects.requireNonNull(session);
+        this.worker = Objects.requireNonNull(worker);
         this.producer = Objects.requireNonNull(producer);
     }
 
@@ -55,7 +56,7 @@ public Observable<Void> write(final M message) {
 
     @Override
     public Observable<Void> shutdown() {
-        return this.session.submit(new Callable<Void>() {
+        return this.worker.submit(new Callable<Void>() {
             @Override
             public Void call() throws Exception {
                 MessageProducerDataSink.this.producer.close();

src/main/java/crud/jms/MessageProducingDataSet.java

@@ -62,7 +62,7 @@ public DataSink<M, Void> dataSink(final String key, final Session session) {
         final javax.jms.Session realSession = sessionImpl.getDelegate();
         try {
             final MessageProducer messageProducer = realSession.createProducer(this.destination);
-            return new MessageProducerDataSink<>(sessionImpl, messageProducer);
+            return new MessageProducerDataSink<>(sessionImpl.worker(), messageProducer);
         } catch (final JMSException jx) {
             throw new MiddlewareException(jx.getMessage(), jx);
         }

src/main/java/crud/jms/NonTransactedJmsSession.java

@@ -20,14 +20,9 @@
 /*package*/ final class NonTransactedJmsSession extends SessionWrapper implements Session {
 
     public NonTransactedJmsSession(final javax.jms.Session delegate) {
-        super(delegate);
+        super(Session.Ordering.ORDERED, delegate);
         // Assumed, but illegal to check in this thread:
         //assert !getDelegate().getTransacted();
     }
 
-    @Override
-    public Session.Ordering getOrdering() {
-        return Session.Ordering.ORDERED;
-    }
-
 }

src/main/java/crud/jms/SessionWrapper.java

@@ -15,47 +15,36 @@
 package crud.jms;
 
 import java.util.Objects;
-import java.util.concurrent.Callable;
-import java.util.concurrent.TimeUnit;
 
 import javax.annotation.Nonnull;
 import javax.jms.JMSException;
 
+import crud.core.Session;
+import crud.implementer.AbstractSession;
 import crud.implementer.SessionWorker;
 
-import crud.core.AsyncCloseable;
-import rx.Observable;
 
+/*package*/ abstract class SessionWrapper extends AbstractSession {
 
-/*package*/ abstract class SessionWrapper implements AsyncCloseable {
-
-    private final SessionWorker worker = new SessionWorker();
     private @Nonnull final javax.jms.Session delegate;
 
-    private final Callable<Void> closeTask = new Callable<Void>() {
-        @Override
-        public Void call() throws JMSException {
-            getDelegate().close();
-            return null;
-        }
-    };
-
-
-    @Override
-    public final Observable<Void> shutdown() {
-        return this.worker.shutdown(this.closeTask, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
-    }
 
-    protected SessionWrapper(@Nonnull final javax.jms.Session delegate) {
+    protected SessionWrapper(@Nonnull final Session.Ordering ordering, @Nonnull final javax.jms.Session delegate) {
+        super(ordering);
         this.delegate = Objects.requireNonNull(delegate);
     }
 
-    protected final Observable<Void> submit(final Callable<Void> task) {
-        return this.worker.submit(task);
+    protected final @Nonnull SessionWorker worker() {
+        return getWorker();
     }
 
     protected final @Nonnull javax.jms.Session getDelegate() {
         return this.delegate;
     }
 
+    @Override
+    protected void doShutdown() throws JMSException {
+        getDelegate().close();
+    }
+
 }

src/main/java/crud/jms/TransactedJmsSession.java

@@ -26,19 +26,14 @@
 /*package*/ final class TransactedJmsSession extends SessionWrapper implements TransactedSession {
 
     public TransactedJmsSession(final javax.jms.Session delegate) {
-        super(delegate);
+        super(Session.Ordering.TRANSACTED, delegate);
         // Assumed, but illegal to check in this thread:
         //assert getDelegate().getTransacted();
     }
 
-    @Override
-    public Session.Ordering getOrdering() {
-        return Session.Ordering.TRANSACTED;
-    }
-
     @Override
     public Observable<Void> commit() {
-        return submit(new Callable<Void>() {
+        return getWorker().submit(new Callable<Void>() {
             @Override
             public Void call() throws JMSException {
                 getDelegate().commit();
@@ -49,7 +44,7 @@ public Void call() throws JMSException {
 
     @Override
     public Observable<Void> rollback() {
-        return submit(new Callable<Void>() {
+        return getWorker().submit(new Callable<Void>() {
             @Override
             public Void call() throws JMSException {
                 getDelegate().rollback();