import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
/**
 * Check if the connection is internal
 * 
 * @param context
 *            the context
 * @param info
 *            the connection info
 * @return true if the connection is embedded
 */
private boolean isEmbeddedConnection(ConnectionContext context, ConnectionInfo info) {
    if (embeddedConfiguration != null) {
        // check the username and password of the connection
        if (embeddedConfiguration.getUsername().equals(info.getUserName())
                && embeddedConfiguration.getPassword()
                        .equals(info.getPassword())) {

            // check if its local
            if (CommunoteJaasLoginModule.isAddressLocal(context
                    .getConnection().getRemoteAddress())) {

                // it is as an embedded connection
                return true;
            }
        }
    }
    return false;
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
/**
 * Determine if a client certificate authentication should be tried.
 * 
 * @param connector
 *            the connector
 * @param info
 *            the connection info
 * @return true if the certificate authentication should be used.
 * @throws IOException
 *             in case of an error
 * @throws URISyntaxException
 *             in case of an error
 */
private boolean useCertificateAuthentication(Connector connector, ConnectionInfo info)
        throws IOException,
        URISyntaxException {
    boolean isSSL = isSslConnection(connector);

    boolean haveUsernamePassword = info.getUserName() != null && info.getPassword() != null;

    // if its as ssl request and we force a client certificate we will do the certificate
    // authentication
    if (isSSL && this.settingsDao.isForceSSLClientAuthentication()) {
        return true;
    }

    // if we have a password, use it
    if (haveUsernamePassword) {
        return false;
    }

    // finally depend on the SSL flag
    return isSSL;
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
/**
 * Overridden to allow for authentication using different Jaas
 * configurations depending on if the connection is SSL or not.
 *
 * @param context The context for the incoming Connection.
 * @param info The ConnectionInfo Command representing the incoming
 *                connection.
 */
@Override
public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
    if (context.getSecurityContext() == null) {
        boolean isSSL;
        Connector connector = context.getConnector();
        if (connector instanceof TransportConnector) {
            TransportConnector transportConnector = (TransportConnector) connector;
            isSSL = transportConnector.getServer().isSslServer();
        } else {
            isSSL = false;
        }

        if (isSSL) {
            this.sslBroker.addConnection(context, info);
        } else {
            this.nonSslBroker.addConnection(context, info);
        }
        super.addConnection(context, info);
    }
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
/**
 * Overriding removeConnection to make sure the security context is cleaned.
 */
@Override
public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
    boolean isSSL;
    Connector connector = context.getConnector();
    if (connector instanceof TransportConnector) {
        TransportConnector transportConnector = (TransportConnector) connector;
        isSSL = (transportConnector.getServer() instanceof SslTransportServer);
    } else {
        isSSL = false;
    }
    super.removeConnection(context, info, error);
    if (isSSL) {
        this.sslBroker.removeConnection(context, info, error);
    } else {
        this.nonSslBroker.removeConnection(context, info, error);
    }
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
@Override
protected void populateObject(Object object) throws Exception {
   super.populateObject(object);
   ConnectionInfo info = (ConnectionInfo) object;

   info.setConnectionId(createConnectionId("ConnectionId:1"));
   info.setClientId("ClientId:2");
   info.setPassword("Password:3");
   info.setUserName("UserName:4");
   {
      BrokerId value[] = new BrokerId[2];
      for (int i = 0; i < 2; i++) {
         value[i] = createBrokerId("BrokerPath:5");
      }
      info.setBrokerPath(value);
   }
   info.setBrokerMasterConnector(true);
   info.setManageable(false);
   info.setClientMaster(true);
   info.setFaultTolerant(false);
   info.setFailoverReconnect(true);
   info.setClientIp("ClientIp:6");
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
@Override
protected void populateObject(Object object) throws Exception {
   super.populateObject(object);
   ConnectionInfo info = (ConnectionInfo) object;

   info.setConnectionId(createConnectionId("ConnectionId:1"));
   info.setClientId("ClientId:2");
   info.setPassword("Password:3");
   info.setUserName("UserName:4");
   {
      BrokerId value[] = new BrokerId[2];
      for (int i = 0; i < 2; i++) {
         value[i] = createBrokerId("BrokerPath:5");
      }
      info.setBrokerPath(value);
   }
   info.setBrokerMasterConnector(true);
   info.setManageable(false);
   info.setClientMaster(true);
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
public AMQSession(ConnectionInfo connInfo,
                  SessionInfo sessInfo,
                  ActiveMQServer server,
                  OpenWireConnection connection,
                  OpenWireProtocolManager protocolManager) {
   this.connInfo = connInfo;
   this.sessInfo = sessInfo;

   this.server = server;
   this.connection = connection;
   this.protocolManager = protocolManager;
   this.scheduledPool = protocolManager.getScheduledPool();
   OpenWireFormat marshaller = (OpenWireFormat) connection.getMarshaller();

   this.converter = new OpenWireMessageConverter(marshaller.copy());
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
public void testAddConsumerThenSend() throws Exception {

      // Start a producer and consumer
      StubConnection connection = createConnection();
      ConnectionInfo connectionInfo = createConnectionInfo();
      SessionInfo sessionInfo = createSessionInfo(connectionInfo);
      ProducerInfo producerInfo = createProducerInfo(sessionInfo);
      connection.send(connectionInfo);
      connection.send(sessionInfo);
      connection.send(producerInfo);

      destination = createDestinationInfo(connection, connectionInfo, destinationType);

      ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
      connection.send(consumerInfo);

      connection.send(createMessage(producerInfo, destination, deliveryMode));

      // Make sure the message was delivered.
      Message m = receiveMessage(connection);
      assertNotNull(m);
   }
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
@Override
protected void populateObject(Object object) throws Exception {
   super.populateObject(object);
   ConnectionInfo info = (ConnectionInfo) object;

   info.setConnectionId(createConnectionId("ConnectionId:1"));
   info.setClientId("ClientId:2");
   info.setPassword("Password:3");
   info.setUserName("UserName:4");
   {
      BrokerId value[] = new BrokerId[2];
      for (int i = 0; i < 2; i++) {
         value[i] = createBrokerId("BrokerPath:5");
      }
      info.setBrokerPath(value);
   }
   info.setBrokerMasterConnector(true);
   info.setManageable(false);
   info.setClientMaster(true);
   info.setFaultTolerant(false);
   info.setFailoverReconnect(true);
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
@Override
public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
    String clientId = info.getClientId();
    if (clientId == null) {
        throw new InvalidClientIDException("No clientID specified for connection disconnect request");
    }
    synchronized (clientIdSet) {
        ConnectionContext oldValue = clientIdSet.get(clientId);
        // we may be removing the duplicate connection, not the first
        // connection to be created
        // so lets check that their connection IDs are the same
        if (oldValue == context) {
            if (isEqual(oldValue.getConnectionId(), info.getConnectionId())) {
                clientIdSet.remove(clientId);
            }
        }
    }
    connections.remove(context.getConnection());
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
/**
 * Write a object instance to data output stream
 * 
 * @param o the instance to be marshaled
 * @param dataOut the output stream
 * @throws IOException thrown if an error occurs
 */
public void tightMarshal2(OpenWireFormat wireFormat, Object o, DataOutput dataOut, BooleanStream bs)
    throws IOException {
    super.tightMarshal2(wireFormat, o, dataOut, bs);

    ConnectionInfo info = (ConnectionInfo)o;
    tightMarshalCachedObject2(wireFormat, (DataStructure)info.getConnectionId(), dataOut, bs);
    tightMarshalString2(info.getClientId(), dataOut, bs);
    tightMarshalString2(info.getPassword(), dataOut, bs);
    tightMarshalString2(info.getUserName(), dataOut, bs);
    tightMarshalObjectArray2(wireFormat, info.getBrokerPath(), dataOut, bs);
    bs.readBoolean();
    bs.readBoolean();
    bs.readBoolean();

}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
/**
 * Un-marshal an object instance from the data input stream
 * 
 * @param o the object to un-marshal
 * @param dataIn the data input stream to build the object from
 * @throws IOException
 */
public void looseUnmarshal(OpenWireFormat wireFormat, Object o, DataInput dataIn) throws IOException {
    super.looseUnmarshal(wireFormat, o, dataIn);

    ConnectionInfo info = (ConnectionInfo)o;
    info.setConnectionId((org.apache.activemq.command.ConnectionId)looseUnmarsalCachedObject(wireFormat,
                                                                                             dataIn));
    info.setClientId(looseUnmarshalString(dataIn));
    info.setPassword(looseUnmarshalString(dataIn));
    info.setUserName(looseUnmarshalString(dataIn));

    if (dataIn.readBoolean()) {
        short size = dataIn.readShort();
        org.apache.activemq.command.BrokerId value[] = new org.apache.activemq.command.BrokerId[size];
        for (int i = 0; i < size; i++) {
            value[i] = (org.apache.activemq.command.BrokerId)looseUnmarsalNestedObject(wireFormat, dataIn);
        }
        info.setBrokerPath(value);
    } else {
        info.setBrokerPath(null);
    }
    info.setBrokerMasterConnector(dataIn.readBoolean());
    info.setManageable(dataIn.readBoolean());
    info.setClientMaster(dataIn.readBoolean());

}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
@Override
protected void populateObject(Object object) throws Exception {
   super.populateObject(object);
   ConnectionInfo info = (ConnectionInfo) object;
   info.setConnectionId(createConnectionId("ConnectionId:1"));
   info.setClientId("ClientId:2");
   info.setPassword("Password:3");
   info.setUserName("UserName:4");

   {
      BrokerId value[] = new BrokerId[2];
      for (int i = 0; i < 2; i++) {
         value[i] = createBrokerId("BrokerPath:5");
      }
      info.setBrokerPath(value);
   }
   info.setBrokerMasterConnector(true);
   info.setManageable(false);

}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
/**
 * Write the booleans that this object uses to a BooleanStream
 */
public int tightMarshal1(OpenWireFormat wireFormat, Object o, BooleanStream bs) throws IOException {

    ConnectionInfo info = (ConnectionInfo)o;

    int rc = super.tightMarshal1(wireFormat, o, bs);
    rc += tightMarshalCachedObject1(wireFormat, (DataStructure)info.getConnectionId(), bs);
    rc += tightMarshalString1(info.getClientId(), bs);
    rc += tightMarshalString1(info.getPassword(), bs);
    rc += tightMarshalString1(info.getUserName(), bs);
    rc += tightMarshalObjectArray1(wireFormat, info.getBrokerPath(), bs);
    bs.writeBoolean(info.isBrokerMasterConnector());
    bs.writeBoolean(info.isManageable());
    bs.writeBoolean(info.isClientMaster());

    return rc + 0;
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
public void testQueueSendThenAddConsumer() throws Exception {

      // Start a producer
      StubConnection connection = createConnection();
      ConnectionInfo connectionInfo = createConnectionInfo();
      SessionInfo sessionInfo = createSessionInfo(connectionInfo);
      ProducerInfo producerInfo = createProducerInfo(sessionInfo);
      connection.send(connectionInfo);
      connection.send(sessionInfo);
      connection.send(producerInfo);

      destination = createDestinationInfo(connection, connectionInfo, destinationType);

      // Send a message to the broker.
      connection.send(createMessage(producerInfo, destination, deliveryMode));

      // Start the consumer
      ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
      connection.send(consumerInfo);

      // Make sure the message was delivered.
      Message m = receiveMessage(connection);
      assertNotNull(m);

   }
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
@Override
public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
    SecurityContext securityContext = context.getSecurityContext();
    if (securityContext == null) {
        securityContext = authenticate(info.getUserName(), info.getPassword(), null);
        context.setSecurityContext(securityContext);
        securityContexts.add(securityContext);
    }

    try {
        super.addConnection(context, info);
    } catch (Exception e) {
        securityContexts.remove(securityContext);
        context.setSecurityContext(null);
        throw e;
    }
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
public void removeConnection(ConnectionInfo info, Throwable error) throws InvalidClientIDException {
   synchronized (clientIdSet) {
      String clientId = info.getClientId();
      if (clientId != null) {
         AMQConnectionContext context = this.clientIdSet.get(clientId);
         if (context != null && context.decRefCount() == 0) {
            //connection is still there and need to close
            context.getConnection().disconnect(error != null);
            this.connections.remove(context.getConnection());
            this.clientIdSet.remove(clientId);
         }
      } else {
         throw new InvalidClientIDException("No clientID specified for connection disconnect request");
      }
   }
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
public void reconnect(AMQConnectionContext existingContext, ConnectionInfo info) {
   this.context = existingContext;
   WireFormatInfo wireFormatInfo = wireFormat.getPreferedWireFormatInfo();
   // Older clients should have been defaulting this field to true.. but
   // they were not.
   if (wireFormatInfo != null && wireFormatInfo.getVersion() <= 2) {
      info.setClientMaster(true);
   }
   if (info.getClientIp() == null) {
      info.setClientIp(getRemoteAddress());
   }

   state = new ConnectionState(info);
   state.reset(info);

   context.setConnection(this);
   context.setConnectionState(state);
   context.setClientMaster(info.isClientMaster());
   context.setFaultTolerant(info.isFaultTolerant());
   context.setReconnect(true);
   context.incRefCount();
}
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
public void testConcurrentConnection() throws Exception {

      StubConnection connection1 = createConnection();
      StubConnection connection2 = createConnection();

      // reuse same connection info
      ConnectionInfo connectionInfo = createConnectionInfo();
      connection1.request(connectionInfo);
      connection2.request(connectionInfo);

      // second one should win out, verify using consumer on default session (watchAdvisories)
      ConsumerId consumerId = new ConsumerId(new SessionId(connectionInfo.getConnectionId(), -1), 1);
      ConsumerInfo consumerInfo = new ConsumerInfo(consumerId);
      consumerInfo.setDestination(AdvisorySupport.TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC);

      connection2.request(consumerInfo);
   }
 

import org.apache.activemq.command.ConnectionInfo; //导入依赖的package包/类
@Override
public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {

    super.removeConnection(context, info, error);
    if (this.securityContexts.remove(context.getSecurityContext())) {
        context.setSecurityContext((SecurityContext) null);
    }

}