import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public boolean intercept(final Packet packet, final RemotingConnection connection) throws ActiveMQException {
   System.out.println("SimpleInterceptor gets called!");
   System.out.println("Packet: " + packet.getClass().getName());
   System.out.println("RemotingConnection: " + connection.getRemoteAddress());

   if (packet instanceof SessionReceiveMessage) {
      SessionReceiveMessage realPacket = (SessionReceiveMessage) packet;
      Message msg = realPacket.getMessage();
      msg.putStringProperty(new SimpleString("newproperty"), new SimpleString("Hello from interceptor!"));
   }
   // We return true which means "call next interceptor" (if there is one) or target.
   // If we returned false, it means "abort call" - no more interceptors would be called and neither would
   // the target
   return true;
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public boolean intercept(Packet packet, RemotingConnection connection) throws ActiveMQException {
   if (packet.getType() != PacketImpl.BACKUP_REGISTRATION_FAILED)
      return true;

   if (logger.isTraceEnabled()) {
      logger.trace("Received ReplicationError::" + packet);
   }
   BackupReplicationStartFailedMessage message = (BackupReplicationStartFailedMessage) packet;
   switch (message.getRegistrationProblem()) {
      case ALREADY_REPLICATING:
         tryNext();
         break;
      case AUTHENTICATION:
         failed();
         break;
      case EXCEPTION:
         failed();
         break;
      default:
         failed();

   }
   return false;
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
protected ServerSessionImpl internalCreateSession(String name,
                                                  String username,
                                                  String password,
                                                  String validatedUser,
                                                  int minLargeMessageSize,
                                                  RemotingConnection connection,
                                                  boolean autoCommitSends,
                                                  boolean autoCommitAcks,
                                                  boolean preAcknowledge,
                                                  boolean xa,
                                                  String defaultAddress,
                                                  SessionCallback callback,
                                                  OperationContext context,
                                                  boolean autoCreateJMSQueues,
                                                  Map<SimpleString, RoutingType> prefixes) throws Exception {
   return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, configuration.isPersistDeliveryCountBeforeDelivery(), xa, connection, storageManager, postOffice, resourceManager, securityStore, managementService, this, configuration.getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), callback, context, pagingManager, prefixes);
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public String[] listConnectionIDs() {
   checkStarted();

   clearIO();
   try {
      Set<RemotingConnection> connections = remotingService.getConnections();
      String[] connectionIDs = new String[connections.size()];
      int i = 0;
      for (RemotingConnection connection : connections) {
         connectionIDs[i++] = connection.getID().toString();
      }
      return connectionIDs;
   } finally {
      blockOnIO();
   }
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public boolean intercept(final Packet packet, final RemotingConnection conn) throws ActiveMQException {
   log.debug("DummyFilter packet = " + packet.getClass().getName());
   syncCounter.addAndGet(1);
   if (sendException) {
      throw new ActiveMQInternalErrorException();
   }
   if (changeMessage) {
      if (packet instanceof SessionReceiveMessage) {
         SessionReceiveMessage deliver = (SessionReceiveMessage) packet;
         log.debug("msg = " + deliver.getMessage().getClass().getName());
         deliver.getMessage().putStringProperty(new SimpleString("DummyInterceptor"), new SimpleString("was here"));
      }
   }
   return true;
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
protected ServerSessionImpl internalCreateSession(String name,
                                                  String username,
                                                  String password,
                                                  String validatedUser,
                                                  int minLargeMessageSize,
                                                  RemotingConnection connection,
                                                  boolean autoCommitSends,
                                                  boolean autoCommitAcks,
                                                  boolean preAcknowledge,
                                                  boolean xa,
                                                  String defaultAddress,
                                                  SessionCallback callback,
                                                  OperationContext context,
                                                  boolean autoCreateQueue,
                                                  Map<SimpleString, RoutingType> prefixes) throws Exception {
   return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, getConfiguration().isPersistDeliveryCountBeforeDelivery(), xa, connection, getStorageManager(), getPostOffice(), getResourceManager(), getSecurityStore(), getManagementService(), this, getConfiguration().getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), new MyCallback(callback), context, getPagingManager(), prefixes);
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Test
public void testProducerWithSmallWindowSizeAndLargeMessage() throws Exception {
   final CountDownLatch latch = new CountDownLatch(1);
   server.getRemotingService().addIncomingInterceptor(new Interceptor() {
      @Override
      public boolean intercept(final Packet packet, final RemotingConnection connection) throws ActiveMQException {
         if (packet.getType() == PacketImpl.SESS_SEND) {
            latch.countDown();
         }
         return true;
      }
   });
   ServerLocator locator = createInVMNonHALocator().setConfirmationWindowSize(100);
   ClientSessionFactory cf = locator.createSessionFactory();
   ClientSession session = cf.createSession(false, true, true);
   ClientProducer producer = session.createProducer(QUEUE);
   ClientMessage message = session.createMessage(true);
   byte[] body = new byte[1000];
   message.getBodyBuffer().writeBytes(body);
   producer.send(message);
   Assert.assertTrue(latch.await(5, TimeUnit.SECONDS));
   session.close();
   locator.close();
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Test
public void testUsingDeadConnection() throws Exception {
   for (int i = 0; i < 100; i++) {
      final Connection conn1 = cf1.createConnection();

      Session sess1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
      RemotingConnection rc1 = ((ClientSessionInternal) ((ActiveMQSession) sess1).getCoreSession()).getConnection();

      rc1.fail(new ActiveMQNotConnectedException("blah"));

      try {
         conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Assert.fail("should throw exception");
      } catch (JMSException e) {
         //pass
      }

      conn1.close();
   }
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public void connectionDestroyed(final Object connectionID) {

   if (logger.isTraceEnabled()) {
      logger.trace("Connection removed " + connectionID + " from server " + this.server, new Exception("trace"));
   }

   ConnectionEntry conn = connections.get(connectionID);

   if (conn != null && !conn.connection.isSupportReconnect()) {
      RemotingConnection removedConnection = removeConnection(connectionID);
      if (removedConnection != null) {
         try {
            server.callBrokerPlugins(server.hasBrokerPlugins() ? plugin -> plugin.afterDestroyConnection(removedConnection) : null);
         } catch (ActiveMQException t) {
            logger.warn("Error executing afterDestroyConnection plugin method: {}", t.getMessage(), t);
            conn.connection.fail(t);
            return;
         }
      }
      conn.connection.fail(new ActiveMQRemoteDisconnectException());
   }
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
private RemotingConnection getForwardingConnection(final Bridge bridge) throws Exception {
   long start = System.currentTimeMillis();

   do {
      RemotingConnection forwardingConnection = ((BridgeImpl) bridge).getForwardingConnection();

      if (forwardingConnection != null) {
         return forwardingConnection;
      }

      Thread.sleep(10);
   }
   while (System.currentTimeMillis() - start < 50000);

   throw new IllegalStateException("Failed to get forwarding connection");
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public boolean closeConnectionsForAddress(final String ipAddress) {
   checkStarted();

   clearIO();
   try {
      boolean closed = false;
      Set<RemotingConnection> connections = remotingService.getConnections();
      for (RemotingConnection connection : connections) {
         String remoteAddress = connection.getRemoteAddress();
         if (remoteAddress.contains(ipAddress)) {
            connection.fail(ActiveMQMessageBundle.BUNDLE.connectionsClosedByManagement(ipAddress));
            remotingService.removeConnection(connection.getID());
            closed = true;
         }
      }

      return closed;
   } finally {
      blockOnIO();
   }

}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public String[] listRemoteAddresses(final String ipAddress) {
   checkStarted();

   clearIO();
   try {
      Set<RemotingConnection> connections = remotingService.getConnections();
      List<String> remoteConnections = new ArrayList<>();
      for (RemotingConnection connection : connections) {
         String remoteAddress = connection.getRemoteAddress();
         if (remoteAddress.contains(ipAddress)) {
            remoteConnections.add(connection.getRemoteAddress());
         }
      }
      return remoteConnections.toArray(new String[remoteConnections.size()]);
   } finally {
      blockOnIO();
   }

}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
public boolean validateUser(String login, String passcode, RemotingConnection remotingConnection) {
   boolean validated = true;

   ActiveMQSecurityManager sm = server.getSecurityManager();

   if (sm != null && server.getConfiguration().isSecurityEnabled()) {
      if (sm instanceof ActiveMQSecurityManager3) {
         validated = ((ActiveMQSecurityManager3) sm).validateUser(login, passcode, remotingConnection) != null;
      } else if (sm instanceof ActiveMQSecurityManager2) {
         validated = ((ActiveMQSecurityManager2) sm).validateUser(login, passcode, CertificateUtil.getCertsFromConnection(remotingConnection));
      } else {
         validated = sm.validateUser(login, passcode);
      }
   }

   return validated;
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public synchronized void freeze(final String scaleDownNodeID, final CoreRemotingConnection connectionToKeepOpen) {
   if (!started)
      return;
   failureCheckAndFlushThread.close(false);
   HashMap<Object, ConnectionEntry> connectionEntries = new HashMap<>(connections);

   // Now we ensure that no connections will process any more packets after this method is
   // complete then send a disconnect packet
   for (Entry<Object, ConnectionEntry> entry : connectionEntries.entrySet()) {
      RemotingConnection conn = entry.getValue().connection;

      if (conn.equals(connectionToKeepOpen))
         continue;

      if (logger.isTraceEnabled()) {
         logger.trace("Sending connection.disconnection packet to " + conn);
      }

      if (!conn.isClient()) {
         conn.disconnect(scaleDownNodeID, false);
         removeConnection(entry.getKey());
      }
   }
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
private Subject getAuthenticatedSubject(final String user,
                                        final String password,
                                        final RemotingConnection remotingConnection) throws LoginException {
   LoginContext lc;
   ClassLoader currentLoader = Thread.currentThread().getContextClassLoader();
   ClassLoader thisLoader = this.getClass().getClassLoader();
   try {
      if (thisLoader != currentLoader) {
         Thread.currentThread().setContextClassLoader(thisLoader);
      }
      if (certificateConfigurationName != null && certificateConfigurationName.length() > 0 && getCertsFromConnection(remotingConnection) != null) {
         lc = new LoginContext(certificateConfigurationName, null, new JaasCallbackHandler(user, password, remotingConnection), certificateConfiguration);
      } else {
         lc = new LoginContext(configurationName, null, new JaasCallbackHandler(user, password, remotingConnection), configuration);
      }
      lc.login();
      return lc.getSubject();
   } finally {
      if (thisLoader != currentLoader) {
         Thread.currentThread().setContextClassLoader(currentLoader);
      }
   }
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
protected RemotingConnection establishNewConnection() {
   Connection transportConnection = createTransportConnection();

   if (transportConnection == null) {
      if (ClientSessionFactoryImpl.logger.isTraceEnabled()) {
         logger.trace("Neither backup or live were active, will just give up now");
      }
      return null;
   }

   RemotingConnection newConnection = clientProtocolManager.connect(transportConnection, callTimeout, callFailoverTimeout, incomingInterceptors, outgoingInterceptors, new SessionFactoryTopologyHandler());

   newConnection.addFailureListener(new DelegatingFailureListener(newConnection.getID()));

   schedulePing();

   if (logger.isTraceEnabled()) {
      logger.trace("returning " + newConnection);
   }

   return newConnection;
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public void nodeDisconnected(RemotingConnection conn, String nodeID, String scaleDownTargetNodeID) {

   if (logger.isTraceEnabled()) {
      logger.trace("Disconnect being called on client:" +
                      " server locator = " +
                      serverLocator +
                      " notifying node " +
                      nodeID +
                      " as down", new Exception("trace"));
   }

   serverLocator.notifyNodeDown(System.currentTimeMillis(), nodeID);

   closeExecutor.execute(new CloseRunnable(conn, scaleDownTargetNodeID));

}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public boolean intercept(final MqttMessage mqttMessage, RemotingConnection connection) {
   System.out.println("MQTT control packet was intercepted " + mqttMessage.fixedHeader().messageType());

   // If you need to handle an specific packet type:
   if (mqttMessage instanceof MqttPublishMessage) {
      MqttPublishMessage message = (MqttPublishMessage) mqttMessage;


      String originalMessage = message.payload().toString(Charset.forName("UTF-8"));
      System.out.println("Original message: " + originalMessage);

      // The new message content must not be bigger that the original content.
      String modifiedMessage = "Modified message ";

      message.payload().setBytes(0, modifiedMessage.getBytes());
   } else {
      if (mqttMessage instanceof MqttConnectMessage) {
         MqttConnectMessage connectMessage = (MqttConnectMessage) mqttMessage;
         System.out.println("MQTT CONNECT control packet was intercepted " + connectMessage);
      }
   }


   // We return true which means "call next interceptor" (if there is one) or target.
   // If we returned false, it means "abort call" - no more interceptors would be called and neither would
   // the target
   return true;
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public void beforeCreateSession(String name, String username, int minLargeMessageSize, RemotingConnection connection,
                                boolean autoCommitSends, boolean autoCommitAcks, boolean preAcknowledge, boolean xa,
                                String defaultAddress, SessionCallback callback, boolean autoCreateQueues,
                                OperationContext context, Map<SimpleString, RoutingType> prefixes) {
   Preconditions.checkNotNull(connection);
   methodCalled(BEFORE_CREATE_SESSION);
}
 

import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; //导入依赖的package包/类
@Override
public void connectionException(Object connectionID, ActiveMQException me) {
   RemotingConnection connection = connectionMap.get(connectionID);
   if (connection != null) {
      log.info("Connection " + connection.getRemoteAddress() + " exception: " + me.getMessage());
      connection.fail(me);
   } else {
      log.error("Connection with id " + connectionID + " not found in connectionException");
   }
}