Fix for https://issues.apache.org/jira/browse/AMQ-4719

yape · Nov 12, 2013 · 7e000d5 · 7e000d5
1 parent bc9751a
commit 7e000d5
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 1 deletion.
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/TransportServer.java b/activemq-client/src/main/java/org/apache/activemq/transport/TransportServer.java
@@ -65,4 +65,11 @@ public interface TransportServer extends Service {
      *          connections.
      */
     boolean isSslServer();
+
+    /**
+     * Some protocols allow link stealing by default (if 2 connections have the same clientID - the youngest wins).
+     * This is the default for AMQP and MQTT. However, JMS 1.1 spec requires the opposite
+     * @return
+     */
+    boolean isAllowLinkStealing();
 }
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/TransportServerFilter.java b/activemq-client/src/main/java/org/apache/activemq/transport/TransportServerFilter.java
@@ -59,4 +59,8 @@ public InetSocketAddress getSocketAddress() {
     public boolean isSslServer() {
         return next.isSslServer();
     }
+
+    public boolean isAllowLinkStealing() {
+        return next.isAllowLinkStealing();
+    }
 }
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransportServer.java b/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransportServer.java
@@ -72,6 +72,8 @@ public class TcpTransportServer extends TransportServerThreadSupport implements
     protected long maxInactivityDurationInitalDelay = 10000;
     protected int minmumWireFormatVersion;
     protected boolean useQueueForAccept = true;
+    protected boolean allowLinkStealing;
+
 
     /**
      * trace=true -> the Transport stack where this TcpTransport object will be, will have a TransportLogger layer
@@ -343,7 +345,7 @@ public String toString() {
 
     /**
      * @param socket
-     * @param inetAddress
+     * @param bindAddress
      * @return real hostName
      * @throws UnknownHostException
      */
@@ -511,4 +513,13 @@ public void stopped(Service service) {
     public boolean isSslServer() {
         return false;
     }
+
+    @Override
+    public boolean isAllowLinkStealing() {
+        return allowLinkStealing;
+    }
+
+    public void setAllowLinkStealing(boolean allowLinkStealing) {
+        this.allowLinkStealing = allowLinkStealing;
+    }
 }
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java b/activemq-client/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java
@@ -52,6 +52,7 @@ public class UdpTransportServer extends TransportServerSupport {
     private final Transport configuredTransport;
     private boolean usingWireFormatNegotiation;
     private final Map<DatagramEndpoint, Transport> transports = new HashMap<DatagramEndpoint, Transport>();
+    private boolean allowLinkStealing;
 
     public UdpTransportServer(URI connectURI, UdpTransport serverTransport, Transport configuredTransport, ReplayStrategy replayStrategy) {
         super(connectURI);
@@ -189,4 +190,13 @@ public InetSocketAddress getSocketAddress() {
     public boolean isSslServer() {
         return false;
     }
+
+    @Override
+    public boolean isAllowLinkStealing() {
+        return allowLinkStealing;
+    }
+
+    public void setAllowLinkStealing(boolean allowLinkStealing) {
+        this.allowLinkStealing = allowLinkStealing;
+    }
 }