import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
private static ClientSession startSession(
        ClientSessionFactory csf,
        FunctionE<ClientSessionFactory, ClientSession> createSession)
            throws ActiveMQException {
    boolean created = false;
    try {
        ClientSession session = createSession.apply(csf);
        session.start();
        created = true;

        return session;
    } finally {
        if (!created) {    // (1)
            csf.close();
        }
    }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
protected void initMocks() throws ActiveMQException {
    CoreQueueConfiguration q = new CoreQueueConfiguration();
    q.setName("q");
    ClientSession sesh = mock(ClientSession.class);
    
    producer = mock(ClientProducer.class);
    when(sesh.createProducer(q.getAddress())).thenReturn(producer);
    
    msgToQueue = mock(ClientMessage.class);
    boolean durable = true;
    when(sesh.createMessage(durable)).thenReturn(msgToQueue);
    
    msgBody = mock(ActiveMQBuffer.class);
    when(msgToQueue.getBodyBuffer()).thenReturn(msgBody);
    
    connector = new ArtemisQConnector(q, sesh);
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public boolean intercept(Packet packet, RemotingConnection connection) throws ActiveMQException {
   if (packet.getType() == PacketImpl.BACKUP_REGISTRATION) {
      try {
         SharedNothingBackupActivation activation = (SharedNothingBackupActivation) backup.getActivation();
         ReplicationEndpoint repEnd = activation.getReplicationEndpoint();
         handler.addSubHandler(repEnd);
         Channel repChannel = repEnd.getChannel();
         repChannel.setHandler(handler);
         handler.setChannel(repChannel);
         live.getRemotingService().removeIncomingInterceptor(this);
      } catch (Exception e) {
         throw new RuntimeException(e);
      }
   }
   return true;
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Test
public void testAddressDoesNotExist() throws Exception {
   ClientSession sendSession = cf.createSession(false, true, true);
   server.getAddressSettingsRepository().addMatch(addressA.toString(), new AddressSettings().setAutoCreateAddresses(false));
   Set<RoutingType> routingTypes = new HashSet<>();
   routingTypes.add(RoutingType.ANYCAST);
   try {
      sendSession.createQueue(addressA, RoutingType.MULTICAST, queueA);
      fail("Creating a queue here should fail since the queue's address doesn't exist and auto-create-addresses = false.");
   } catch (Exception e) {
      assertTrue(e instanceof ActiveMQException);
      ActiveMQException ae = (ActiveMQException) e;
      assertEquals(ActiveMQExceptionType.ADDRESS_DOES_NOT_EXIST, ae.getType());
   }
   sendSession.close();
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
public void closeSession() throws ActiveMQException {
	if (consumer != null) {
		consumer.close();
		consumer = null;
	}

	if (session != null) {
		session.close();
		session = null;
	}

	if (sessionFactory != null) {
		sessionFactory.close();
		sessionFactory = null;
	}

	if (locator != null) {
		locator.close();
		locator = null;
	}
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public void xaStart(Xid xid, int flags) throws XAException, ActiveMQException {
   Packet packet;
   if (flags == XAResource.TMJOIN) {
      packet = new SessionXAJoinMessage(xid);
   } else if (flags == XAResource.TMRESUME) {
      packet = new SessionXAResumeMessage(xid);
   } else if (flags == XAResource.TMNOFLAGS) {
      // Don't need to flush since the previous end will have done this
      packet = new SessionXAStartMessage(xid);
   } else {
      throw new XAException(XAException.XAER_INVAL);
   }

   SessionXAResponseMessage response = (SessionXAResponseMessage) sessionChannel.sendBlocking(packet, PacketImpl.SESS_XA_RESP);

   if (response.isError()) {
      ActiveMQClientLogger.LOGGER.errorCallingStart(response.getMessage(), response.getResponseCode());
      throw new XAException(response.getResponseCode());
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
private void callSessionFailureListeners(final ActiveMQException me,
                                         final boolean afterReconnect,
                                         final boolean failedOver,
                                         final String scaleDownTargetNodeID) {
   final List<SessionFailureListener> listenersClone = new ArrayList<>(listeners);

   for (final SessionFailureListener listener : listenersClone) {
      try {
         if (afterReconnect) {
            listener.connectionFailed(me, failedOver, scaleDownTargetNodeID);
         } else {
            listener.beforeReconnect(me);
         }
      } catch (final Throwable t) {
         // Failure of one listener to execute shouldn't prevent others
         // from
         // executing
         ActiveMQClientLogger.LOGGER.failedToExecuteListener(t);
      }
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public ChannelHandler getActivationChannelHandler(final Channel channel, final Acceptor acceptorUsed) {
   return new ChannelHandler() {
      @Override
      public void handlePacket(Packet packet) {
         if (packet.getType() == PacketImpl.BACKUP_REGISTRATION) {
            BackupRegistrationMessage msg = (BackupRegistrationMessage) packet;
            ClusterConnection clusterConnection = acceptorUsed.getClusterConnection();
            try {
               startReplication(channel.getConnection(), clusterConnection, getPair(msg.getConnector(), true), msg.isFailBackRequest());
            } catch (ActiveMQAlreadyReplicatingException are) {
               channel.send(new BackupReplicationStartFailedMessage(BackupReplicationStartFailedMessage.BackupRegistrationProblem.ALREADY_REPLICATING));
            } catch (ActiveMQException e) {
               logger.debug("Failed to process backup registration packet", e);
               channel.send(new BackupReplicationStartFailedMessage(BackupReplicationStartFailedMessage.BackupRegistrationProblem.EXCEPTION));
            }
         }
      }
   };
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public JournalLoader createJournalLoader(PostOffice postOffice,
                                         PagingManager pagingManager,
                                         StorageManager storageManager,
                                         QueueFactory queueFactory,
                                         NodeManager nodeManager,
                                         ManagementService managementService,
                                         GroupingHandler groupingHandler,
                                         Configuration configuration,
                                         ActiveMQServer parentServer) throws ActiveMQException {
   if (sharedStoreSlavePolicy.getScaleDownPolicy() != null && sharedStoreSlavePolicy.getScaleDownPolicy().isEnabled()) {
      return new BackupRecoveryJournalLoader(postOffice, pagingManager, storageManager, queueFactory, nodeManager, managementService, groupingHandler, configuration, parentServer, ScaleDownPolicy.getScaleDownConnector(sharedStoreSlavePolicy.getScaleDownPolicy(), activeMQServer), activeMQServer.getClusterManager().getClusterController());
   } else {
      return super.createJournalLoader(postOffice, pagingManager, storageManager, queueFactory, nodeManager, managementService, groupingHandler, configuration, parentServer);
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public void setClientID(final String clientID) throws JMSException {
   checkClosed();

   if (this.clientID != null) {
      throw new IllegalStateException("Client id has already been set");
   }

   if (!justCreated) {
      throw new IllegalStateException("setClientID can only be called directly after the connection is created");
   }

   validateClientID(initialSession, clientID);

   this.clientID = clientID;
   try {
      this.addSessionMetaData(initialSession);
   } catch (ActiveMQException e) {
      JMSException ex = new JMSException("Internal error setting metadata jms-client-id");
      ex.setLinkedException(e);
      ex.initCause(e);
      throw ex;
   }

   justCreated = false;
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
/**
 * @param persistDeliveryCountBeforeDelivery
 * @throws Exception
 */
private void setUp(final boolean persistDeliveryCountBeforeDelivery) throws Exception {
   Configuration config = createDefaultInVMConfig().setPersistDeliveryCountBeforeDelivery(persistDeliveryCountBeforeDelivery);

   server = createServer(true, config);

   server.start();
   locator = createInVMNonHALocator();
   factory = createSessionFactory(locator);

   ClientSession session = addClientSession(factory.createSession(false, false, false));
   try {
      session.createQueue(ADDRESS, ADDRESS, true);
   } catch (ActiveMQException expected) {
      // in case of restart
   }

   session.close();
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public void connectionDestroyed(final Object connectionID) {
   // The exception has to be created in the same thread where it's being called
   // as to avoid a different stack trace cause
   final ActiveMQException ex = ActiveMQClientMessageBundle.BUNDLE.channelDisconnected();

   // It has to use the same executor as the disconnect message is being sent through

   closeExecutor.execute(new Runnable() {
      @Override
      public void run() {
         handleConnectionFailure(connectionID, ex);
      }
   });

}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
private ActiveMQTopic lookupTopic(final String topicName, final boolean isTemporary) throws ActiveMQException {

      ActiveMQTopic topic;

      if (isTemporary) {
         topic = ActiveMQDestination.createTemporaryTopic(topicName);
      } else {
         topic = ActiveMQDestination.createTopic(topicName);
      }

      AddressQuery query = session.addressQuery(topic.getSimpleAddress());

      if (!query.isExists() && !query.isAutoCreateAddresses()) {
         return null;
      } else {
         return topic;
      }
   }
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
public void publish(HttpHeaders headers,
                    byte[] body,
                    String dup,
                    boolean durable,
                    Long ttl,
                    Long expiration,
                    Integer priority) throws Exception {
   Pooled pooled = getPooled();
   try {
      ClientProducer producer = pooled.producer;
      ClientMessage message = createActiveMQMessage(headers, body, durable, ttl, expiration, priority, pooled.session);
      message.putStringProperty(Message.HDR_DUPLICATE_DETECTION_ID.toString(), dup);
      producer.send(message);
      ActiveMQRestLogger.LOGGER.debug("Sent message: " + message);
      pool.add(pooled);
   } catch (Exception ex) {
      try {
         pooled.session.close();
      } catch (ActiveMQException e) {
      }
      addPooled();
      throw ex;
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
private ActiveMQQueue lookupQueue(final String queueName, boolean isTemporary) throws ActiveMQException {
   ActiveMQQueue queue;

   if (isTemporary) {
      queue = ActiveMQDestination.createTemporaryQueue(queueName);
   } else {
      queue = ActiveMQDestination.createQueue(queueName);
   }

   QueueQuery response = session.queueQuery(queue.getSimpleAddress());

   if (!response.isExists() && !response.isAutoCreateQueues()) {
      return null;
   } else {
      return queue;
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的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.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public TemporaryQueue createTemporaryQueue() throws JMSException {
   // As per spec. section 4.11
   if (sessionType == ActiveMQSession.TYPE_TOPIC_SESSION) {
      throw new IllegalStateException("Cannot create a temporary queue using a TopicSession");
   }

   try {
      ActiveMQTemporaryQueue queue = ActiveMQDestination.createTemporaryQueue(this);

      SimpleString simpleAddress = queue.getSimpleAddress();

      session.createTemporaryQueue(simpleAddress, RoutingType.ANYCAST, simpleAddress);

      connection.addTemporaryQueue(simpleAddress);

      return queue;
   } catch (ActiveMQException e) {
      throw JMSExceptionHelper.convertFromActiveMQException(e);
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public void sendACK(boolean individual,
                    boolean block,
                    final ClientConsumer consumer,
                    final Message message) throws ActiveMQException {
   PacketImpl messagePacket;
   if (individual) {
      messagePacket = new SessionIndividualAcknowledgeMessage(getConsumerID(consumer), message.getMessageID(), block);
   } else {
      messagePacket = new SessionAcknowledgeMessage(getConsumerID(consumer), message.getMessageID(), block);
   }

   if (block) {
      sessionChannel.sendBlocking(messagePacket, PacketImpl.NULL_RESPONSE);
   } else {
      sessionChannel.sendBatched(messagePacket);
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
@Override
public Queue createQueue(final String queueName) throws JMSException {
   // As per spec. section 4.11
   if (sessionType == ActiveMQSession.TYPE_TOPIC_SESSION) {
      throw new IllegalStateException("Cannot create a queue using a TopicSession");
   }

   try {
      Queue queue = null;
      if (cacheDestination) {
         queue = queueCache.get(queueName);
      }
      if (queue == null) {
         queue = internalCreateQueue(queueName, false);
      }
      if (cacheDestination) {
         queueCache.put(queueName, queue);
      }
      return queue;
   } catch (ActiveMQException e) {
      throw JMSExceptionHelper.convertFromActiveMQException(e);
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQException; //导入依赖的package包/类
private static ClientSession createSession(ClientSessionFactory csf,
                                           String username,
                                           String password,
                                           boolean xa,
                                           boolean autoCommitSends,
                                           boolean autoCommitAcks,
                                           boolean preAcknowledge,
                                           int ackBatchSize)
        throws ActiveMQException {
    requireNonNull(csf, "csf");
    return csf.createSession(username, password, xa, autoCommitSends,
            autoCommitAcks, preAcknowledge, ackBatchSize);
}