import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
private void connectInternal(Runnable command) {
	final JMSContext jmsCtx = mock(JMSContext.class);
	when(this.jmsCtxProvider.get()).thenReturn(jmsCtx);

	when(this.artemisConfig.getManagementNotificationAddress()).thenReturn(new SimpleString("notif"));
	final JMSConsumer consumer = mock(JMSConsumer.class);
	when(jmsCtx.createConsumer(any())).thenReturn(consumer);

	command.run();

	verify(this.jmsCtxProvider).get();
	verify(this.artemisConfig).getManagementNotificationAddress();
	verify(this.log).info("Connecting to broker for sourcing destination events.");
	verify(jmsCtx).createConsumer(any());
	verify(consumer).setMessageListener(this.eventProducer);
	verifyNoMoreInteractions(consumer);
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Override
public synchronized boolean moveReference(final long messageID,
                                          final SimpleString toAddress,
                                          final boolean rejectDuplicate) throws Exception {
   try (LinkedListIterator<MessageReference> iter = iterator()) {
      while (iter.hasNext()) {
         MessageReference ref = iter.next();
         if (ref.getMessage().getMessageID() == messageID) {
            iter.remove();
            refRemoved(ref);
            incDelivering();
            try {
               move(null, toAddress, ref, rejectDuplicate, AckReason.NORMAL);
            } catch (Exception e) {
               decDelivering();
               throw e;
            }
            return true;
         }
      }
      return false;
   }
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Override
public void sendQueueInfoToQueue(final String queueName, final String address) throws Exception {
   checkStarted();

   clearIO();
   try {
      postOffice.sendQueueInfoToQueue(new SimpleString(queueName), new SimpleString(address == null ? "" : address));

      GroupingHandler handler = server.getGroupingHandler();
      if (handler != null) {
         // the group handler would miss responses if the group was requested before the reset was done
         // on that case we ask the groupinghandler to replay its send in case it's waiting for the information
         handler.resendPending();
      }
   } finally {
      blockOnIO();
   }
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
/**
 * @param queueID
 * @param pageSubscriptions
 * @param queueInfos
 * @return
 */
private static PageSubscription locateSubscription(final long queueID,
                                                   final Map<Long, PageSubscription> pageSubscriptions,
                                                   final Map<Long, QueueBindingInfo> queueInfos,
                                                   final PagingManager pagingManager) throws Exception {

   PageSubscription subs = pageSubscriptions.get(queueID);
   if (subs == null) {
      QueueBindingInfo queueInfo = queueInfos.get(queueID);

      if (queueInfo != null) {
         SimpleString address = queueInfo.getAddress();
         PagingStore store = pagingManager.getPageStore(address);
         subs = store.getCursorProvider().getSubscription(queueID);
         pageSubscriptions.put(queueID, subs);
      }
   }

   return subs;
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Override
public void removeBinding(final Binding binding) {
   if (binding.isExclusive()) {
      exclusiveBindings.remove(binding);
   } else {
      SimpleString routingName = binding.getRoutingName();

      List<Binding> bindings = routingNameBindingMap.get(routingName);

      if (bindings != null) {
         bindings.remove(binding);

         if (bindings.isEmpty()) {
            routingNameBindingMap.remove(routingName);
         }
      }
   }

   bindingsMap.remove(binding.getID());

   if (logger.isTraceEnabled()) {
      logger.trace("Removing binding " + binding + " from " + this + " bindingTable: " + debugBindings());
   }
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testTypedProperties() throws Exception {
   SimpleString longKey = RandomUtil.randomSimpleString();
   long longValue = RandomUtil.randomLong();
   SimpleString simpleStringKey = RandomUtil.randomSimpleString();
   SimpleString simpleStringValue = RandomUtil.randomSimpleString();
   TypedProperties otherProps = new TypedProperties();
   otherProps.putLongProperty(longKey, longValue);
   otherProps.putSimpleStringProperty(simpleStringKey, simpleStringValue);

   props.putTypedProperties(otherProps);

   long ll = props.getLongProperty(longKey);
   Assert.assertEquals(longValue, ll);
   SimpleString ss = props.getSimpleStringProperty(simpleStringKey);
   Assert.assertEquals(simpleStringValue, ss);
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testByteBufSimpleStringPool() {
   final int capacity = 8;
   final int chars = Integer.toString(capacity).length();
   final SimpleString.ByteBufSimpleStringPool pool = new SimpleString.ByteBufSimpleStringPool(capacity, chars);
   final int bytes = new SimpleString(Integer.toString(capacity)).sizeof();
   final ByteBuf bb = Unpooled.buffer(bytes, bytes);
   for (int i = 0; i < capacity; i++) {
      final SimpleString s = new SimpleString(Integer.toString(i));
      bb.resetWriterIndex();
      SimpleString.writeSimpleString(bb, s);
      bb.resetReaderIndex();
      final SimpleString expectedPooled = pool.getOrCreate(bb);
      bb.resetReaderIndex();
      Assert.assertSame(expectedPooled, pool.getOrCreate(bb));
   }
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testCreateQueueWithNullAddress() throws Exception {
   SimpleString address = RandomUtil.randomSimpleString();
   SimpleString name = address;

   ActiveMQServerControl serverControl = createManagementControl();

   checkNoResource(ObjectNameBuilder.DEFAULT.getQueueObjectName(address, name, RoutingType.ANYCAST));
   serverControl.createQueue(null, name.toString(), "ANYCAST");

   checkResource(ObjectNameBuilder.DEFAULT.getQueueObjectName(name, name, RoutingType.ANYCAST));
   QueueControl queueControl = ManagementControlHelper.createQueueControl(address, name, RoutingType.ANYCAST, mbeanServer);
   Assert.assertEquals(address.toString(), queueControl.getAddress());
   Assert.assertEquals(name.toString(), queueControl.getName());
   Assert.assertNull(queueControl.getFilter());
   Assert.assertEquals(true, queueControl.isDurable());
   Assert.assertEquals(false, queueControl.isTemporary());

   serverControl.destroyQueue(name.toString());

   checkNoResource(ObjectNameBuilder.DEFAULT.getQueueObjectName(address, name, RoutingType.ANYCAST));
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testMulticastMessageRoutingExclusivity() throws Exception {
   conn.connect(defUser, defPass);

   final String addressA = "addressA";
   final String queueA = "queueA";
   final String queueB = "queueB";
   final String queueC = "queueC";

   ActiveMQServer activeMQServer = server.getActiveMQServer();
   ActiveMQServerControl serverControl = server.getActiveMQServer().getActiveMQServerControl();
   serverControl.createAddress(addressA, RoutingType.ANYCAST.toString() + "," + RoutingType.MULTICAST.toString());
   serverControl.createQueue(addressA, queueA, RoutingType.ANYCAST.toString());
   serverControl.createQueue(addressA, queueB, RoutingType.MULTICAST.toString());
   serverControl.createQueue(addressA, queueC, RoutingType.MULTICAST.toString());

   send(conn, addressA, null, "Hello World!", true, RoutingType.MULTICAST);

   assertTrue(Wait.waitFor(() -> activeMQServer.locateQueue(SimpleString.toSimpleString(queueA)).getMessageCount() == 0, 2000, 100));
   assertTrue(Wait.waitFor(() -> activeMQServer.locateQueue(SimpleString.toSimpleString(queueC)).getMessageCount() + activeMQServer.locateQueue(SimpleString.toSimpleString(queueB)).getMessageCount() == 2, 2000, 100));
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testLargeMessage() throws Exception {
   ClientProducer producer = clientSessionTxReceives.createProducer(address);
   ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
   SimpleString rh = new SimpleString("SMID1");

   for (int i = 0; i < 50; i++) {
      ClientMessage message = clientSession.createMessage(true);
      message.setBodyInputStream(createFakeLargeStream(300 * 1024));
      message.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(message);
      clientSession.commit();
   }
   clientSessionTxReceives.start();
   ClientMessage m = consumer.receive(1000);
   Assert.assertNotNull(m);
   m.acknowledge();
   Assert.assertNull(consumer.receiveImmediate());
   clientSessionTxReceives.commit();
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testGetMessagesAdded() throws Exception {
   SimpleString address = RandomUtil.randomSimpleString();
   SimpleString queue = RandomUtil.randomSimpleString();

   session.createQueue(address, queue, null, false);

   QueueControl queueControl = createManagementControl(address, queue);
   Assert.assertEquals(0, getMessagesAdded(queueControl));

   ClientProducer producer = session.createProducer(address);
   producer.send(session.createMessage(false));
   Assert.assertEquals(1, getMessagesAdded(queueControl));
   producer.send(session.createMessage(false));
   Assert.assertEquals(2, getMessagesAdded(queueControl));

   consumeMessages(2, session, queue);

   Assert.assertEquals(2, getMessagesAdded(queueControl));

   session.deleteQueue(queue);
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
private void printMessageProperties(Message message) throws XMLStreamException {
   xmlWriter.writeStartElement(XmlDataConstants.PROPERTIES_PARENT);
   for (SimpleString key : message.getPropertyNames()) {
      Object value = message.getObjectProperty(key);
      xmlWriter.writeEmptyElement(XmlDataConstants.PROPERTIES_CHILD);
      xmlWriter.writeAttribute(XmlDataConstants.PROPERTY_NAME, key.toString());
      xmlWriter.writeAttribute(XmlDataConstants.PROPERTY_VALUE, XmlDataExporterUtil.convertProperty(value));

      // Write the property type as an attribute
      String propertyType = XmlDataExporterUtil.getPropertyType(value);
      if (propertyType != null) {
         xmlWriter.writeAttribute(XmlDataConstants.PROPERTY_TYPE, propertyType);
      }
   }
   xmlWriter.writeEndElement(); // end PROPERTIES_PARENT
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test(timeout = 60000)
public void testMulticastMessageRoutingExclusivityUsingProperty() throws Exception {
   final String addressA = "addressA";
   final String queueA = "queueA";
   final String queueB = "queueB";
   final String queueC = "queueC";

   ActiveMQServerControl serverControl = server.getActiveMQServerControl();
   serverControl.createAddress(addressA, RoutingType.ANYCAST.toString() + "," + RoutingType.MULTICAST.toString());
   serverControl.createQueue(addressA, queueA, RoutingType.ANYCAST.toString());
   serverControl.createQueue(addressA, queueB, RoutingType.MULTICAST.toString());
   serverControl.createQueue(addressA, queueC, RoutingType.MULTICAST.toString());

   sendMessages(addressA, 1, RoutingType.MULTICAST);

   assertEquals(0, server.locateQueue(SimpleString.toSimpleString(queueA)).getMessageCount());
   assertEquals(2, server.locateQueue(SimpleString.toSimpleString(queueC)).getMessageCount() + server.locateQueue(SimpleString.toSimpleString(queueB)).getMessageCount());
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testSystemPropertyBlackWhiteListDefault() throws Exception {
   System.setProperty(ObjectInputStreamWithClassLoader.BLACKLIST_PROPERTY, "*");
   System.setProperty(ObjectInputStreamWithClassLoader.WHITELIST_PROPERTY, "some.other.package");

   String qname = "SerialTestQueue";
   SimpleString qaddr = new SimpleString(qname);
   liveService.createQueue(qaddr, RoutingType.ANYCAST, qaddr, null, true, false);

   try {
      String blackList = null;
      String whiteList = null;
      Object obj = receiveObjectMessage(blackList, whiteList, qname, new TestClass1(), false, false);
      assertTrue("Object is " + obj, obj instanceof JMSException);
      //but String always trusted
      obj = receiveObjectMessage(blackList, whiteList, qname, new String("hello"), false, false);
      assertTrue("java.lang.String always trusted " + obj, "hello".equals(obj));

      //override
      blackList = "some.other.package";
      whiteList = "org.apache.activemq.artemis.tests.integration";
      obj = receiveObjectMessage(blackList, whiteList, qname, new TestClass1(), false, false);
      assertTrue("Object is " + obj, obj instanceof TestClass1);
      //but String always trusted
      obj = receiveObjectMessage(blackList, whiteList, qname, new String("hello"), false, false);
      assertTrue("java.lang.String always trusted " + obj, "hello".equals(obj));
   } finally {
      System.clearProperty(ObjectInputStreamWithClassLoader.BLACKLIST_PROPERTY);
      System.clearProperty(ObjectInputStreamWithClassLoader.WHITELIST_PROPERTY);
   }
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testM() {
   SimpleString s1 = new SimpleString("a.b.c");
   SimpleString s2 = new SimpleString("a.b.x.e");
   SimpleString s3 = new SimpleString("a.b.c.#");
   Address a1 = new AddressImpl(s1);
   Address a2 = new AddressImpl(s2);
   Address w = new AddressImpl(s3);
   Assert.assertTrue(a1.matches(w));
   Assert.assertFalse(a2.matches(w));
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Test
public void testSECURITY_PERMISSION_VIOLATION() throws Exception {
   SimpleString queue = RandomUtil.randomSimpleString();
   SimpleString address = RandomUtil.randomSimpleString();

   // guest can not create queue
   Role role = new Role("roleCanNotCreateQueue", true, true, false, true, false, true, true, true, true, true);
   Set<Role> roles = new HashSet<>();
   roles.add(role);
   server.getSecurityRepository().addMatch(address.toString(), roles);
   ActiveMQJAASSecurityManager securityManager = (ActiveMQJAASSecurityManager) server.getSecurityManager();
   securityManager.getConfiguration().addRole("guest", "roleCanNotCreateQueue");

   SecurityNotificationTest.flush(notifConsumer);

   ServerLocator locator = createInVMNonHALocator();
   ClientSessionFactory sf = createSessionFactory(locator);
   ClientSession guestSession = sf.createSession("guest", "guest", false, true, true, false, 1);

   try {
      guestSession.createQueue(address, queue, true);
      Assert.fail("session creation must fail and a notification of security violation must be sent");
   } catch (Exception e) {
   }

   ClientMessage[] notifications = SecurityNotificationTest.consumeMessages(1, notifConsumer);
   Assert.assertEquals(SECURITY_PERMISSION_VIOLATION.toString(), notifications[0].getObjectProperty(ManagementHelper.HDR_NOTIFICATION_TYPE).toString());
   Assert.assertEquals("guest", notifications[0].getObjectProperty(ManagementHelper.HDR_USER).toString());
   Assert.assertEquals(address.toString(), notifications[0].getObjectProperty(ManagementHelper.HDR_ADDRESS).toString());
   Assert.assertEquals(CheckType.CREATE_DURABLE_QUEUE.toString(), notifications[0].getObjectProperty(ManagementHelper.HDR_CHECK_TYPE).toString());

   guestSession.close();
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
protected void createAddressInfo(final int node,
                                 final String address,
                                 final RoutingType routingType,
                                 final int defaulMaxConsumers,
                                 boolean defaultPurgeOnNoConsumers) throws Exception {
   AddressInfo addressInfo = new AddressInfo(new SimpleString(address));
   addressInfo.addRoutingType(routingType);
   servers[node].addOrUpdateAddressInfo(addressInfo);
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Override
public void createQueue(final String address,
                        final String queueName,
                        final String filterString,
                        final boolean durable,
                        final boolean autoCreated) throws ActiveMQException {
   createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(filterString), durable, autoCreated);
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
@Override
public Object getObjectProperty(String name) {
   try {
      SimpleString key = new SimpleString(name);
      Object property = properties.getProperty(key);
      if (stringPropertyNames.contains(key)) {
         property = property.toString();
      }
      return property;
   } catch (ActiveMQPropertyConversionException ce) {
      throw new MessageFormatRuntimeException(ce.getMessage());
   } catch (RuntimeException e) {
      throw new JMSRuntimeException(e.getMessage());
   }
}
 

import org.apache.activemq.artemis.api.core.SimpleString; //导入依赖的package包/类
private synchronized void removeBinding(final SimpleString clusterName) throws Exception {
   RemoteQueueBinding binding = bindings.remove(clusterName);

   if (binding == null) {
      throw new IllegalStateException("Cannot find binding for queue " + clusterName);
   }

   postOffice.removeBinding(binding.getUniqueName(), null, false);
}