Merge branch 'release/13.1.2'

Author: iitsoftware

pom.xml

@@ -7,7 +7,7 @@
 
   <groupId>com.swiftmq</groupId>
   <artifactId>swiftmq-client</artifactId>
-  <version>13.1.1</version>
+  <version>13.1.2</version>
 
   <name>SwiftMQ Client</name>
   <description>Client for SwiftMQ Messaging System with JMS, AMQP 1.0 and file transfer over JMS.</description>

src/main/java/com/swiftmq/amqp/v100/client/Connection.java

@@ -25,6 +25,7 @@
 import com.swiftmq.amqp.v100.client.po.POOpen;
 import com.swiftmq.amqp.v100.client.po.POProtocolRequest;
 import com.swiftmq.amqp.v100.client.po.POSendClose;
+import com.swiftmq.amqp.v100.generated.transport.definitions.Fields;
 import com.swiftmq.amqp.v100.transport.AMQPFrame;
 import com.swiftmq.amqp.v100.types.AMQPString;
 import com.swiftmq.amqp.v100.types.AMQPSymbol;
@@ -92,6 +93,7 @@ public class Connection implements ExceptionHandler {
     int outputBufferExtendSize = 65536;
     SocketFactory socketFactory = new PlainSocketFactory();
     ExceptionListener exceptionListener = null;
+    Fields properties = null;
 
     /**
      * Creates a Connection and uses SASL for authentication.
@@ -223,6 +225,23 @@ public void setOpenHostname(String openHostname) {
         this.openHostname = openHostname;
     }
 
+    /**
+     * Sets the of the OpenFrame to be sent to the remote endpoint
+     *
+     * @param properties Fields for the OpenFrame
+     */
+    public void setProperties(Fields properties) {
+        this.properties = properties;
+    }
+
+    /**
+     * Retrieves the properties of the OpenFrame from the remote endpoint. Available after connect
+     *
+     * @return Fields object representing the remote properties, or null if not available
+     */
+    public Fields getRemoteProperties() {
+        return connectionDispatcher.getRemoteProperties();
+    }
     /**
      * Sets the idle timeout. Default is Long.MAX_VALUE.
      * <p/>
@@ -365,6 +384,8 @@ protected ProtocolInputHandler createInputHandler() {
                 return connectionDispatcher.getProtocolHandler();
             }
         };
+        if (properties != null)
+            connectionDispatcher.setProperties(properties);
         connectionDispatcher.setMyConnection(this);
         networkConnection.start();
         dos = new DataStreamOutputStream(networkConnection.getOutputStream());

src/main/java/com/swiftmq/amqp/v100/client/ConnectionDispatcher.java

@@ -26,6 +26,7 @@
 import com.swiftmq.amqp.v100.generated.security.sasl.*;
 import com.swiftmq.amqp.v100.generated.transport.definitions.ConnectionError;
 import com.swiftmq.amqp.v100.generated.transport.definitions.ErrorConditionFactory;
+import com.swiftmq.amqp.v100.generated.transport.definitions.Fields;
 import com.swiftmq.amqp.v100.generated.transport.definitions.Milliseconds;
 import com.swiftmq.amqp.v100.generated.transport.performatives.*;
 import com.swiftmq.amqp.v100.transport.AMQPFrame;
@@ -76,7 +77,7 @@ public class ConnectionDispatcher
     POProtocolRequest protPO = null;
     POAuthenticate authPO = null;
     POOpen openPO = null;
-    OpenFrame remoteOpen = null;
+    volatile OpenFrame remoteOpen = null;
     POSendClose closePO = null;
     CloseFrame remoteClose = null;
     SaslMechanismsFrame saslMechanisms = null;
@@ -89,6 +90,7 @@ public class ConnectionDispatcher
     long idleTimeoutDelay = 0;
     SaslClient saslClient = null;
     volatile boolean connectionDisabled = false;
+    Fields properties = null;
 
     int maxLocalFrameSize = Integer.MAX_VALUE;
     int maxRemoteFrameSize = Integer.MAX_VALUE;
@@ -133,6 +135,16 @@ public int getMaxFrameSize() {
         return Math.max(512, Math.min(maxLocalFrameSize, maxRemoteFrameSize));
     }
 
+    public void setProperties(Fields properties) {
+        this.properties = properties;
+    }
+
+    public Fields getRemoteProperties() {
+        if (remoteOpen != null)
+            return remoteOpen.getProperties();
+        return null;
+    }
+
     public void setSaslActive(boolean saslActive) {
         this.saslActive = saslActive;
     }
@@ -317,6 +329,8 @@ public void visit(POOpen po) {
             OpenFrame openFrame = new OpenFrame(0);
             openFrame.setContainerId(new AMQPString(po.getContainerId()));
             openFrame.setChannelMax(new AMQPUnsignedShort(po.getMaxChannel()));
+            if (properties != null)
+                openFrame.setProperties(properties);
             if (myConnection.getOpenHostname() == null)
                 openFrame.setHostname(new AMQPString(remoteHostname));
             else

src/main/java/com/swiftmq/client/thread/DefaultPoolManager.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 public class DefaultPoolManager extends PoolManager {
+    public static final String PROP_JAC_ACTIVE = "swiftmq.jac.active";
     public static final String PROP_CONNECTOR_POOL_MIN_THREADS = "swiftmq.pool.connector.threads.min";
     public static final String PROP_CONNECTOR_POOL_MAX_THREADS = "swiftmq.pool.connector.threads.max";
     public static final String PROP_CONNECTOR_POOL_PRIO = "swiftmq.pool.connector.priority";
@@ -45,6 +46,16 @@ public class DefaultPoolManager extends PoolManager {
     ThreadPool connectionPool = null;
     ThreadPool sessionPool = null;
     ThreadPool connectorPool = null;
+    boolean JAC_ACTIVE = System.getProperty(PROP_JAC_ACTIVE, "false").equals("true");
+
+    public DefaultPoolManager() {
+        if (JAC_ACTIVE) {
+            ThreadPool pool = new IVMTaskScheduler();
+            connectionPool = pool;
+            sessionPool = pool;
+            connectorPool = pool;
+        }
+    }
 
     ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
 

src/main/java/com/swiftmq/client/thread/IVMTaskScheduler.java

@@ -0,0 +1,106 @@
+package com.swiftmq.client.thread;
+
+import com.swiftmq.swiftlet.SwiftletManager;
+import com.swiftmq.swiftlet.threadpool.AsyncTask;
+import com.swiftmq.swiftlet.threadpool.ThreadPool;
+import com.swiftmq.swiftlet.threadpool.ThreadpoolSwiftlet;
+import com.swiftmq.swiftlet.threadpool.event.FreezeCompletionListener;
+
+public class IVMTaskScheduler implements ThreadPool {
+
+    ThreadpoolSwiftlet threadpoolSwiftlet = null;
+
+    public IVMTaskScheduler() {
+        threadpoolSwiftlet = (ThreadpoolSwiftlet) SwiftletManager.getInstance().getSwiftlet("sys$threadpool");
+        if (threadpoolSwiftlet == null)
+            System.err.println("[" + this + "] Failed to get ThreadpoolSwiftlet");
+        else
+            System.out.println("[" + this + "] App uses virtual threads, scheduled in: adhocvirtual");
+    }
+
+    /**
+     * Closes the pool.
+     * Internal use only.
+     */
+    @Override
+    public void close() {
+
+    }
+
+    /**
+     * Returns the pool name.
+     *
+     * @return pool name.
+     */
+    @Override
+    public String getPoolName() {
+        return toString();
+    }
+
+    /**
+     * Returns the number of currently idling threads.
+     * Used from management tools only.
+     *
+     * @return number of idling threads.
+     */
+    @Override
+    public int getNumberIdlingThreads() {
+        return 0;
+    }
+
+    /**
+     * Returns the number of currently running threads.
+     * Used from management tools only.
+     *
+     * @return number of running threads.
+     */
+    @Override
+    public int getNumberRunningThreads() {
+        return 0;
+    }
+
+    /**
+     * Dispatch a task into the pool.
+     *
+     * @param asyncTask the task to dispatch.
+     */
+    @Override
+    public void dispatchTask(AsyncTask asyncTask) {
+        threadpoolSwiftlet.runAsync(asyncTask, true);
+    }
+
+    /**
+     * Freezes this pool. That is, the current running tasks are completed but
+     * no further tasks will be scheduled until unfreeze() is called. It is possible
+     * to dispatch tasks during freeze. However, these will be executed after unfreeze()
+     * is called.
+     *
+     * @param listener will be called when the pool is freezed.
+     */
+    @Override
+    public void freeze(FreezeCompletionListener listener) {
+
+    }
+
+    /**
+     * Unfreezes this pool.
+     */
+    @Override
+    public void unfreeze() {
+
+    }
+
+    /**
+     * Stops the pool.
+     * Internal use only.
+     */
+    @Override
+    public void stop() {
+
+    }
+
+    @Override
+    public String toString() {
+        return IVMTaskScheduler.class.getSimpleName();
+    }
+}

src/main/java/com/swiftmq/jms/v750/MessageConsumerImpl.java

@@ -33,6 +33,7 @@
 import javax.jms.IllegalStateException;
 import javax.jms.*;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
@@ -371,9 +372,9 @@ Message receiveMessage(boolean block, long timeout) throws JMSException {
                             } else {
                                 long to = timeout;
                                 do {
-                                    long startWait = System.currentTimeMillis();
+                                    long startWait = System.nanoTime();
                                     waiter.doWait(to);
-                                    long delta = System.currentTimeMillis() - startWait;
+                                    long delta = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startWait);
                                     to -= delta;
                                 }
                                 while (to > 0 && messageCache.getSize() == 0 && fillCachePending.get() && !cancelled.get() && !isClosed());