Added network and fault tolerant flags to ConnectionContext

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@631241 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/Connection.java

@@ -105,4 +105,10 @@ public interface Connection extends Service {
 
     String getConnectionId();
     
+    /**
+     * return true if a network connection
+     * @return
+     */
+    boolean isNetworkConnection();
+
 }

activemq-core/src/main/java/org/apache/activemq/broker/ConnectionContext.java

@@ -53,6 +53,7 @@ public class ConnectionContext {
     private boolean producerFlowControl = true;
     private MessageAuthorizationPolicy messageAuthorizationPolicy;
     private boolean networkConnection;
+    private boolean faultTolerant;
     private final AtomicBoolean stopping = new AtomicBoolean();
     private final MessageEvaluationContext messageEvaluationContext = new MessageEvaluationContext();
     private boolean dontSendReponse;
@@ -278,4 +279,12 @@ public class ConnectionContext {
     public void setClientMaster(boolean clientMaster) {
         this.clientMaster = clientMaster;
     }
+
+    public boolean isFaultTolerant() {
+        return faultTolerant;
+    }
+
+    public void setFaultTolerant(boolean faultTolerant) {
+        this.faultTolerant = faultTolerant;
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java

@@ -146,6 +146,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
     private CountDownLatch dispatchStoppedLatch = new CountDownLatch(1);
     private ConnectionContext context;
     private boolean networkConnection;
+    private boolean faultTolerantConnection;
     private AtomicInteger protocolVersion = new AtomicInteger(CommandTypes.PROTOCOL_VERSION);
     private DemandForwardingBridge duplexBridge;
     private final TaskRunnerFactory taskRunnerFactory;
@@ -647,6 +648,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
         context.setConnector(connector);
         context.setMessageAuthorizationPolicy(getMessageAuthorizationPolicy());
         context.setNetworkConnection(networkConnection);
+        context.setFaultTolerant(faultTolerantConnection);
         context.setTransactions(new ConcurrentHashMap<TransactionId, Transaction>());
         context.setUserName(info.getUserName());
         context.setWireFormatInfo(wireFormatInfo);
@@ -1077,6 +1079,10 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
         return starting;
     }
     
+    public synchronized boolean isNetworkConnection() {
+        return networkConnection;
+    }
+
     protected synchronized void setStarting(boolean starting) {
         this.starting = starting;
     }
@@ -1255,6 +1261,9 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
     }
 
     public Response processConnectionControl(ConnectionControl control) throws Exception {
+        if(control != null) {
+            faultTolerantConnection=control.isFaultTolerant();
+        }
         return null;
     }