QFJ-943: optional watermarks-based back pressure propagation from inbo…

quickfixj-core/src/main/java/quickfix/AbstractSessionConnectorBuilder.java

@@ -0,0 +1,74 @@
+package quickfix;
+
+public abstract class AbstractSessionConnectorBuilder<Derived, Product> {
+    private final Class<Derived> derived;
+    Application application;
+    MessageStoreFactory messageStoreFactory;
+    SessionSettings settings;
+    LogFactory logFactory;
+    MessageFactory messageFactory;
+
+    int queueCapacity = -1;
+    int queueLowerWatermark = -1;
+    int queueUpperWatermark = -1;
+
+    AbstractSessionConnectorBuilder(Class<Derived> derived) {
+        this.derived = derived;
+    }
+
+    public Derived withApplication(Application val) throws ConfigError {
+        application = val;
+        return derived.cast(this);
+    }
+
+    public Derived withMessageStoreFactory(MessageStoreFactory val) throws ConfigError {
+        messageStoreFactory = val;
+        return derived.cast(this);
+    }
+
+    public Derived withSettings(SessionSettings val) {
+        settings = val;
+        return derived.cast(this);
+    }
+
+    public Derived withLogFactory(LogFactory val) throws ConfigError {
+        logFactory = val;
+        return derived.cast(this);
+    }
+
+    public Derived withMessageFactory(MessageFactory val) throws ConfigError {
+        messageFactory = val;
+        return derived.cast(this);
+    }
+
+    public Derived withQueueCapacity(int val) throws  ConfigError {
+        if (queueLowerWatermark >= 0) {
+            throw new ConfigError("queue capacity and watermarks may not be configured together");
+        } else if (queueCapacity < 0) {
+            throw new ConfigError("negative queue capacity");
+        }
+        queueCapacity = val;
+        return derived.cast(this);
+    }
+
+    public Derived withQueueWatermarks(int lower, int upper) throws ConfigError {
+        if (queueCapacity >= 0) {
+            throw new ConfigError("queue capacity and watermarks may not be configured together");
+        } else if (queueLowerWatermark < 0 || queueUpperWatermark <= queueLowerWatermark) {
+            throw new ConfigError("invalid queue watermarks, required: 0 <= lower watermark < upper watermark");
+        }
+        queueLowerWatermark = lower;
+        queueUpperWatermark = upper;
+        return derived.cast(this);
+    }
+
+    public final Product build() throws ConfigError {
+        if (logFactory == null) {
+            logFactory = new ScreenLogFactory(settings);
+        }
+
+        return doBuild();
+    }
+
+    protected abstract Product doBuild() throws ConfigError;
+}

quickfixj-core/src/main/java/quickfix/SocketAcceptor.java

@@ -31,6 +31,34 @@ public class SocketAcceptor extends AbstractSocketAcceptor {
     private volatile Boolean isStarted = Boolean.FALSE;
     private final SingleThreadedEventHandlingStrategy eventHandlingStrategy;
 
+    private SocketAcceptor(Builder builder) throws ConfigError {
+        super(builder.application, builder.messageStoreFactory, builder.settings,
+                builder.logFactory, builder.messageFactory);
+
+        if (builder.queueCapacity >= 0) {
+            eventHandlingStrategy
+                    = new SingleThreadedEventHandlingStrategy(this, builder.queueCapacity);
+        } else {
+            eventHandlingStrategy
+                    = new SingleThreadedEventHandlingStrategy(this, builder.queueLowerWatermark, builder.queueUpperWatermark);
+        }
+    }
+
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+    public static final class Builder extends AbstractSessionConnectorBuilder<Builder, SocketAcceptor> {
+        private Builder() {
+            super(Builder.class);
+        }
+
+        @Override
+        protected SocketAcceptor doBuild() throws ConfigError {
+            return new SocketAcceptor(this);
+        }
+    }
+
     public SocketAcceptor(Application application, MessageStoreFactory messageStoreFactory,
             SessionSettings settings, LogFactory logFactory, MessageFactory messageFactory,
             int queueCapacity)

quickfixj-core/src/main/java/quickfix/SocketInitiator.java

@@ -31,6 +31,34 @@ public class SocketInitiator extends AbstractSocketInitiator {
     private volatile Boolean isStarted = Boolean.FALSE;
     private final SingleThreadedEventHandlingStrategy eventHandlingStrategy;
 
+    private SocketInitiator(Builder builder) throws ConfigError {
+        super(builder.application, builder.messageStoreFactory, builder.settings,
+                builder.logFactory, builder.messageFactory);
+
+        if (builder.queueCapacity >= 0) {
+            eventHandlingStrategy
+                    = new SingleThreadedEventHandlingStrategy(this, builder.queueCapacity);
+        } else {
+            eventHandlingStrategy
+                    = new SingleThreadedEventHandlingStrategy(this, builder.queueLowerWatermark, builder.queueUpperWatermark);
+        }
+    }
+
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+    public static final class Builder extends AbstractSessionConnectorBuilder<Builder, SocketInitiator> {
+        private Builder() {
+            super(Builder.class);
+        }
+
+        @Override
+        protected SocketInitiator doBuild() throws ConfigError {
+            return new SocketInitiator(this);
+        }
+    }
+
     public SocketInitiator(Application application, MessageStoreFactory messageStoreFactory,
             SessionSettings settings, MessageFactory messageFactory, int queueCapacity) throws ConfigError {
         super(application, messageStoreFactory, settings, new ScreenLogFactory(settings),

quickfixj-core/src/main/java/quickfix/ThreadedSocketAcceptor.java

@@ -29,6 +29,34 @@
 public class ThreadedSocketAcceptor extends AbstractSocketAcceptor {
     private final ThreadPerSessionEventHandlingStrategy eventHandlingStrategy;
 
+    private ThreadedSocketAcceptor(Builder builder) throws ConfigError {
+        super(builder.application, builder.messageStoreFactory, builder.settings,
+                builder.logFactory, builder.messageFactory);
+
+        if (builder.queueCapacity >= 0) {
+            eventHandlingStrategy
+                    = new ThreadPerSessionEventHandlingStrategy(this, builder.queueCapacity);
+        } else {
+            eventHandlingStrategy
+                    = new ThreadPerSessionEventHandlingStrategy(this, builder.queueLowerWatermark, builder.queueUpperWatermark);
+        }
+    }
+
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+    public static final class Builder extends AbstractSessionConnectorBuilder<Builder, ThreadedSocketAcceptor> {
+        private Builder() {
+            super(Builder.class);
+        }
+
+        @Override
+        protected ThreadedSocketAcceptor doBuild() throws ConfigError {
+            return new ThreadedSocketAcceptor(this);
+        }
+    }
+
     public ThreadedSocketAcceptor(Application application, MessageStoreFactory messageStoreFactory,
                                   SessionSettings settings, LogFactory logFactory, MessageFactory messageFactory,
                                   int queueCapacity )

quickfixj-core/src/main/java/quickfix/ThreadedSocketInitiator.java

@@ -29,6 +29,34 @@
 public class ThreadedSocketInitiator extends AbstractSocketInitiator {
     private final ThreadPerSessionEventHandlingStrategy eventHandlingStrategy;
 
+    private ThreadedSocketInitiator(Builder builder) throws ConfigError {
+        super(builder.application, builder.messageStoreFactory, builder.settings,
+                builder.logFactory, builder.messageFactory);
+
+        if (builder.queueCapacity >= 0) {
+            eventHandlingStrategy
+                    = new ThreadPerSessionEventHandlingStrategy(this, builder.queueCapacity);
+        } else {
+            eventHandlingStrategy
+                    = new ThreadPerSessionEventHandlingStrategy(this, builder.queueLowerWatermark, builder.queueUpperWatermark);
+        }
+    }
+
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+    public static final class Builder extends AbstractSessionConnectorBuilder<Builder, ThreadedSocketInitiator> {
+        private Builder() {
+            super(Builder.class);
+        }
+
+        @Override
+        protected ThreadedSocketInitiator doBuild() throws ConfigError {
+            return new ThreadedSocketInitiator(this);
+        }
+    }
+
     public ThreadedSocketInitiator(Application application,
             MessageStoreFactory messageStoreFactory, SessionSettings settings,
             LogFactory logFactory, MessageFactory messageFactory, int queueCapacity) throws ConfigError {

quickfixj-core/src/main/java/quickfix/mina/EventHandlingStrategy.java

@@ -28,7 +28,6 @@
  * it only handles message reception events.
  */
 public interface EventHandlingStrategy {
-
     /**
      * Constant indicating how long we wait for an incoming message. After
      * thread has been asked to stop, it can take up to this long to terminate.

quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java

@@ -95,4 +95,7 @@ public String getRemoteAddress() {
         return null;
     }
 
+    IoSession getIoSession() {
+        return ioSession;
+    }
 }

quickfixj-core/src/main/java/quickfix/mina/QueueTracker.java

@@ -0,0 +1,11 @@
+package quickfix.mina;
+
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+interface QueueTracker<E> {
+    void put(E e) throws InterruptedException;
+    E poll(long timeout, TimeUnit unit) throws InterruptedException;
+    int drainTo(Collection<E> collection);
+}

quickfixj-core/src/main/java/quickfix/mina/QueueTrackers.java

@@ -0,0 +1,89 @@
+package quickfix.mina;
+
+import org.apache.mina.core.session.IoSession;
+import quickfix.Responder;
+import quickfix.Session;
+
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static java.lang.String.format;
+
+/**
+ * Queue trackers factory methods
+ */
+final class QueueTrackers {
+    private static final String LOWER_WATERMARK_FMT = "inbound queue size < lower watermark (%d), socket reads resumed";
+    private static final String UPPER_WATERMARK_FMT = "inbound queue size > upper watermark (%d), socket reads suspended";
+
+    /**
+     * Watermarks-based queue tracker
+     */
+    static <E> WatermarkTracker<E, Session> newMultiSessionWatermarkTracker(
+            BlockingQueue<E> queue,
+            long lowerWatermark, long upperWatermark,
+            Function<E, Session> classifier) {
+        return WatermarkTracker.newMulti(queue, lowerWatermark, upperWatermark, classifier,
+                qfSession -> resumeReads(qfSession, (int)lowerWatermark),
+                qfSession -> suspendReads(qfSession, (int)upperWatermark));
+    }
+
+    /**
+     * Default no-op queue tracker
+     */
+    static <E> QueueTracker<E> newDefaultQueueTracker(BlockingQueue<E> queue) {
+        return new QueueTracker<E>() {
+            @Override
+            public void put(E e) throws InterruptedException {
+                queue.put(e);
+            }
+
+            @Override
+            public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+                return queue.poll(timeout, unit);
+            }
+
+            @Override
+            public int drainTo(Collection<E> collection) {
+                return queue.drainTo(collection);
+            }
+        };
+    }
+
+    private static IoSession lookupIoSession(Session qfSession) {
+        final Responder responder = qfSession.getResponder();
+
+        if (responder instanceof IoSessionResponder) {
+            return ((IoSessionResponder)responder).getIoSession();
+        } else {
+            return null;
+        }
+    }
+
+    private static void resumeReads(Session qfSession, int queueLowerWatermark) {
+        final IoSession ioSession = lookupIoSession(qfSession);
+        if (ioSession != null && ioSession.isReadSuspended()) {
+            ioSession.resumeRead();
+            qfSession.getLog().onEvent(format(LOWER_WATERMARK_FMT, queueLowerWatermark));
+        }
+    }
+
+    private static void suspendReads(Session qfSession, int queueUpperWatermark) {
+        final IoSession ioSession = lookupIoSession(qfSession);
+        if (ioSession != null && !ioSession.isReadSuspended()) {
+            ioSession.suspendRead();
+            qfSession.getLog().onEvent(format(UPPER_WATERMARK_FMT, queueUpperWatermark));
+        }
+    }
+
+    static <E, Void> WatermarkTracker<E, Void> newSingleSessionWatermarkTracker(
+            BlockingQueue<E> queue,
+            long lowerWatermark, long upperWatermark,
+            Session qfSession) {
+        return WatermarkTracker.newMono(queue, lowerWatermark, upperWatermark,
+                () -> resumeReads(qfSession, (int)lowerWatermark),
+                () -> suspendReads(qfSession, (int)upperWatermark));
+    }
+}