import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
/**
 * Set the application properties for a Proton Message but do a check for all properties first if they only contain
 * values that the AMQP 1.0 spec allows.
 *
 * @param msg The Proton message. Must not be null.
 * @param properties The map containing application properties.
 * @throws NullPointerException if the message passed in is null.
 * @throws IllegalArgumentException if the properties contain any value that AMQP 1.0 disallows.
 */
protected static final void setApplicationProperties(final Message msg, final Map<String, ?> properties) {
    if (properties != null) {

        // check the three types not allowed by AMQP 1.0 spec for application properties (list, map and array)
        for (final Map.Entry<String, ?> entry: properties.entrySet()) {
            if (entry.getValue() instanceof List) {
                throw new IllegalArgumentException(String.format("Application property %s can't be a List", entry.getKey()));
            } else if (entry.getValue() instanceof Map) {
                throw new IllegalArgumentException(String.format("Application property %s can't be a Map", entry.getKey()));
            } else if (entry.getValue().getClass().isArray()) {
                throw new IllegalArgumentException(String.format("Application property %s can't be an Array", entry.getKey()));
            }
        }

        final ApplicationProperties applicationProperties = new ApplicationProperties(properties);
        msg.setApplicationProperties(applicationProperties);
    }
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
private static String getRegistrationAssertion(final Message msg, final boolean removeAssertion) {
    Objects.requireNonNull(msg);
    String assertion = null;
    ApplicationProperties properties = msg.getApplicationProperties();
    if (properties != null) {
        Object obj = null;
        if (removeAssertion) {
            obj = properties.getValue().remove(APP_PROPERTY_REGISTRATION_ASSERTION);
        } else {
            obj = properties.getValue().get(APP_PROPERTY_REGISTRATION_ASSERTION);
        }
        if (obj instanceof String) {
            assertion = (String) obj;
        }
    }
    return assertion;
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
private List<String> checkRouter(SyncRequestClient client, String entityType, String attributeName) {
    Map<String, String> properties = new LinkedHashMap<>();
    properties.put("operation", "QUERY");
    properties.put("entityType", entityType);
    Map body = new LinkedHashMap<>();

    body.put("attributeNames", Arrays.asList(attributeName));

    Message message = Proton.message();
    message.setAddress("$management");
    message.setApplicationProperties(new ApplicationProperties(properties));
    message.setBody(new AmqpValue(body));

    try {
        Message response = client.request(message, 10, TimeUnit.SECONDS);
        AmqpValue value = (AmqpValue) response.getBody();
        Map values = (Map) value.getValue();
        List<List<String>> results = (List<List<String>>) values.get("results");
        return results.stream().map(l -> l.get(0)).collect(Collectors.toList());
    } catch (Exception e) {
        log.info("Error requesting router status. Ignoring", e);
        eventLogger.log(RouterCheckFailed, e.getMessage(), Warning, AddressSpace, addressSpaceName);
        return Collections.emptyList();
    }
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Override
public int getSubscriberCount(AmqpClient queueClient, Destination replyQueue, String queue) throws Exception {
    Message requestMessage = Message.Factory.create();
    Map<String, String> appProperties = new HashMap<>();
    appProperties.put(resourceProperty, "queue." + queue);
    appProperties.put(operationProperty, "getConsumerCount");
    requestMessage.setAddress(managementAddress);
    requestMessage.setApplicationProperties(new ApplicationProperties(appProperties));
    requestMessage.setReplyTo(replyQueue.getAddress());
    requestMessage.setBody(new AmqpValue("[]"));

    Future<Integer> sent = queueClient.sendMessages(managementAddress, requestMessage);
    assertThat(sent.get(30, TimeUnit.SECONDS), is(1));
    Logging.log.info("request sent");

    Future<List<Message>> received = queueClient.recvMessages(replyQueue.getAddress(), 1);
    assertThat(received.get(30, TimeUnit.SECONDS).size(), is(1));


    AmqpValue val = (AmqpValue) received.get().get(0).getBody();
    Logging.log.info("answer received: " + val.toString());
    String count = val.getValue().toString().replaceAll("\\[|]|\"", "");

    return Integer.valueOf(count);
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void benchmarkApplicationProperties() throws IOException {
    ApplicationProperties properties = new ApplicationProperties(new HashMap<String, Object>());
    properties.getValue().put("test1", UnsignedByte.valueOf((byte) 128));
    properties.getValue().put("test2", UnsignedShort.valueOf((short) 128));
    properties.getValue().put("test3", UnsignedInteger.valueOf((byte) 128));

    resultSet.start();
    for (int i = 0; i < ITERATIONS; i++) {
        byteBuf.clear();
        encoder.writeObject(properties);
    }
    resultSet.encodesComplete();

    resultSet.start();
    for (int i = 0; i < ITERATIONS; i++) {
        byteBuf.flip();
        decoder.readObject();
    }
    resultSet.decodesComplete();

    time("ApplicationProperties", resultSet);
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
/**
 * Verifies that a Symbol application property is converted to a String [by the JsonObject]
 */
@Test
public void testAMQP_to_JSON_VerifyApplicationPropertySymbol() {
  Map<String, Object> props = new HashMap<>();
  ApplicationProperties appProps = new ApplicationProperties(props);

  String symbolPropKey = "symbolPropKey";
  Symbol symbolPropValue = Symbol.valueOf("symbolPropValue");

  props.put(symbolPropKey, symbolPropValue);

  Message protonMsg = Proton.message();
  protonMsg.setApplicationProperties(appProps);

  JsonObject jsonObject = translator.convertToJsonObject(protonMsg);
  assertNotNull("expected converted msg", jsonObject);
  assertTrue("expected application properties element key to be present",
      jsonObject.containsKey(AmqpConstants.APPLICATION_PROPERTIES));

  JsonObject jsonAppProps = jsonObject.getJsonObject(AmqpConstants.APPLICATION_PROPERTIES);
  assertNotNull("expected application properties element value to be non-null", jsonAppProps);

  assertTrue("expected key to be present", jsonAppProps.containsKey(symbolPropKey));
  assertEquals("expected value to be equal, as a string", symbolPropValue.toString(),
      jsonAppProps.getValue(symbolPropKey));
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private Map<String, Object> getApplicationPropertiesMap() {
   ApplicationProperties appMap = getApplicationProperties();
   Map<String, Object> map = null;

   if (appMap != null) {
      map = appMap.getValue();
   }

   if (map == null) {
      map = new HashMap<>();
      this.applicationProperties = new ApplicationProperties(map);
   }

   return map;
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
private ApplicationProperties getApplicationProperties() {
   parseHeaders();

   if (applicationProperties == null && appLocation >= 0) {
      ByteBuffer buffer = getBuffer().nioBuffer();
      buffer.position(appLocation);
      TLSEncode.getDecoder().setByteBuffer(buffer);
      Object section = TLSEncode.getDecoder().readObject();
      if (section instanceof ApplicationProperties) {
         this.applicationProperties = (ApplicationProperties) section;
      }
      this.appLocation = -1;
      TLSEncode.getDecoder().setByteBuffer(null);
   }

   return applicationProperties;
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
private Message createTypicalQpidJMSMessage() {
   Map<String, Object> applicationProperties = new HashMap<>();
   Map<Symbol, Object> messageAnnotations = new HashMap<>();

   applicationProperties.put("property-1", "string");
   applicationProperties.put("property-2", 512);
   applicationProperties.put("property-3", true);

   messageAnnotations.put(Symbol.valueOf("x-opt-jms-msg-type"), 0);
   messageAnnotations.put(Symbol.valueOf("x-opt-jms-dest"), 0);

   Message message = Proton.message();

   message.setAddress("queue://test-queue");
   message.setDeliveryCount(1);
   message.setApplicationProperties(new ApplicationProperties(applicationProperties));
   message.setMessageAnnotations(new MessageAnnotations(messageAnnotations));
   message.setCreationTime(System.currentTimeMillis());
   message.setContentType("text/plain");
   message.setBody(new AmqpValue("String payload for AMQP message conversion performance testing."));

   return message;
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Test
public void testSimpleConversionStream() throws Exception {
   Map<String, Object> mapprop = createPropertiesMap();
   ApplicationProperties properties = new ApplicationProperties(mapprop);
   MessageImpl message = (MessageImpl) Message.Factory.create();
   message.setApplicationProperties(properties);

   List<Object> objects = new LinkedList<>();
   objects.add(new Integer(10));
   objects.add("10");

   message.setBody(new AmqpSequence(objects));

   AMQPMessage encodedMessage = new AMQPMessage(message);

   ICoreMessage serverMessage = encodedMessage.toCore();

   ServerJMSStreamMessage streamMessage = (ServerJMSStreamMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);

   verifyProperties(streamMessage);

   streamMessage.reset();

   assertEquals(10, streamMessage.readInt());
   assertEquals("10", streamMessage.readString());
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Test
public void testVerySimple() {
   MessageImpl protonMessage = (MessageImpl) Message.Factory.create();
   protonMessage.setHeader( new Header());
   Properties properties = new Properties();
   properties.setTo("someNiceLocal");
   protonMessage.setProperties(properties);
   protonMessage.getHeader().setDeliveryCount(new UnsignedInteger(7));
   protonMessage.getHeader().setDurable(Boolean.TRUE);
   protonMessage.setApplicationProperties(new ApplicationProperties(new HashMap()));

   AMQPMessage decoded = encodeAndDecodeMessage(protonMessage);

   assertEquals(7, decoded.getHeader().getDeliveryCount().intValue());
   assertEquals(true, decoded.getHeader().getDurable());
   assertEquals("someNiceLocal", decoded.getAddress());
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Test
public void testSendProperties() throws Exception {
   ProtonFactoryLoader<MessengerFactory> factoryLoader = new ProtonFactoryLoader(MessengerFactory.class, ImplementationType.PROTON_J);
   ProtonFactoryLoader<MessageFactory> messageLoader = new ProtonFactoryLoader(MessageFactory.class, ImplementationType.PROTON_J);
   MessengerFactory factory = factoryLoader.loadFactory();
   MessageFactory msgFactory = messageLoader.loadFactory();
   Messenger messenger = factory.createMessenger("testSendBytes");
   messenger.start();
   Message msg = msgFactory.createMessage();
   //msg.setAddress("amqp://192.168.1.107:5672/topic://beaconEvents");
   msg.setAddress("amqp://192.168.1.107:5672/beaconEvents");
   Beacon beacon = createBeacon();
   Map<String, Object> beaconProps = beacon.toProperties();
   msg.setApplicationProperties(new ApplicationProperties(beaconProps));
   messenger.put(msg);

   messenger.send();
   messenger.stop();
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Test
public void testGetProperties() throws Exception {
    Map<String, Object> applicationPropertiesMap = new HashMap<>();
    applicationPropertiesMap.put(TEST_PROP_A, TEST_VALUE_STRING_A);
    applicationPropertiesMap.put(TEST_PROP_B, TEST_VALUE_STRING_B);

    Message message2 = Proton.message();
    message2.setApplicationProperties(new ApplicationProperties(applicationPropertiesMap));

    JmsMessageFacade amqpMessageFacade = createReceivedMessageFacade(createMockAmqpConsumer(), message2);

    Set<String> props = amqpMessageFacade.getPropertyNames();
    assertEquals(2, props.size());
    assertTrue(props.contains(TEST_PROP_A));
    assertEquals(TEST_VALUE_STRING_A, amqpMessageFacade.getProperty(TEST_PROP_A));
    assertTrue(props.contains(TEST_PROP_B));
    assertEquals(TEST_VALUE_STRING_B, amqpMessageFacade.getProperty(TEST_PROP_B));
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Test
public void testGetPropertyNames() throws Exception {
    Map<String, Object> applicationPropertiesMap = new HashMap<>();
    applicationPropertiesMap.put(TEST_PROP_A, TEST_VALUE_STRING_A);
    applicationPropertiesMap.put(TEST_PROP_B, TEST_VALUE_STRING_B);

    Message message2 = Proton.message();
    message2.setApplicationProperties(new ApplicationProperties(applicationPropertiesMap));

    AmqpJmsMessageFacade amqpMessageFacade = createReceivedMessageFacade(createMockAmqpConsumer(), message2);

    Set<String> applicationPropertyNames = amqpMessageFacade.getPropertyNames();
    assertEquals(2, applicationPropertyNames.size());
    assertTrue(applicationPropertyNames.contains(TEST_PROP_A));
    assertTrue(applicationPropertyNames.contains(TEST_PROP_B));
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Test
public void testClearProperties() throws Exception {
    Map<String, Object> applicationPropertiesMap = new HashMap<>();
    applicationPropertiesMap.put(TEST_PROP_A, TEST_VALUE_STRING_A);

    Message message = Proton.message();
    message.setApplicationProperties(new ApplicationProperties(applicationPropertiesMap));

    JmsMessageFacade amqpMessageFacade = createReceivedMessageFacade(createMockAmqpConsumer(), message);

    Set<String> props1 = amqpMessageFacade.getPropertyNames();
    assertEquals(1, props1.size());

    amqpMessageFacade.clearProperties();

    Set<String> props2 = amqpMessageFacade.getPropertyNames();
    assertTrue(props2.isEmpty());
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
/**
 * Creates a new <em>Proton</em> message containing a JSON encoded temperature reading.
 *
 * @param messageId the value to set as the message ID.
 * @param action The value to set for the message's <em>action</em> application property.
 * @param tenantId the ID of the tenant the device belongs to.
 * @param deviceId the ID of the device that produced the reading.
 * @param replyTo The reply-to address to set on the message.
 * @return the message containing the reading as a binary payload.
 */
public static Message newRegistrationMessage(final String messageId, final String action, final String tenantId, final String deviceId, final String replyTo) {
    final Message message = ProtonHelper.message();
    message.setMessageId(messageId);
    message.setReplyTo(replyTo);
    final HashMap<String, String> map = new HashMap<>();
    map.put(MessageHelper.APP_PROPERTY_DEVICE_ID, deviceId);
    map.put(MessageHelper.APP_PROPERTY_TENANT_ID, tenantId);
    map.put(MessageHelper.SYS_PROPERTY_SUBJECT, action);
    final ApplicationProperties applicationProperties = new ApplicationProperties(map);
    message.setApplicationProperties(applicationProperties);
    return message;
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
/**
 * Build a Proton message containing the reply to a request for the provided endpoint.
 *
 * @param endpoint The endpoint the reply message will be built for.
 * @param status The status from the service that processed the message.
 * @param correlationId The UUID to correlate the reply with the originally sent message.
 * @param tenantId The tenant for which the message was processed.
 * @param deviceId The device that the message relates to.
 * @param isApplCorrelationId Flag to inidicate if the correlationId has to be available as application property
 *        {@link MessageHelper#ANNOTATION_X_OPT_APP_CORRELATION_ID}.
 * @param payload The payload of the message reply as json object.
 * @return Message The built Proton message. Maybe null. In that case, the message reply will not contain a body.
 */
public static final Message getAmqpReply(final String endpoint, final Integer status, final Object correlationId,
                                         final String tenantId, final String deviceId, final boolean isApplCorrelationId,
                                         final JsonObject payload) {

    final ResourceIdentifier address = ResourceIdentifier.from(endpoint, tenantId, deviceId);
    final Message message = ProtonHelper.message();
    message.setMessageId(UUID.randomUUID().toString());
    message.setCorrelationId(correlationId);
    message.setAddress(address.toString());

    final Map<String, Object> map = new HashMap<>();
    map.put(MessageHelper.APP_PROPERTY_DEVICE_ID, deviceId);
    map.put(MessageHelper.APP_PROPERTY_TENANT_ID, tenantId);
    map.put(MessageHelper.APP_PROPERTY_STATUS, status);
    message.setApplicationProperties(new ApplicationProperties(map));

    if (isApplCorrelationId) {
        Map<Symbol, Object> annotations = new HashMap<>();
        annotations.put(Symbol.valueOf(MessageHelper.ANNOTATION_X_OPT_APP_CORRELATION_ID), true);
        message.setMessageAnnotations(new MessageAnnotations(annotations));
    }

    if (payload != null) {
        message.setContentType("application/json; charset=utf-8");
        message.setBody(new AmqpValue(payload.encode()));
    }
    return message;
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
/**
 * Gets the value of a specific <em>application property</em>.
 * 
 * @param <T> The expected type of the property to retrieve the value of.
 * @param props The application properties to retrieve the value from.
 * @param name The property name.
 * @param type The expected value type.
 * @return The value or {@code null} if the properties do not contain a value of
 *         the expected type for the given name.
 */
@SuppressWarnings("unchecked")
public static <T> T getApplicationProperty(final ApplicationProperties props, final String name, final Class<T> type) {
    if (props == null) {
        return null;
    } else {
        Object value = props.getValue().get(name);
        if (type.isInstance(value)) {
            return (T) value;
        } else {
            return null;
        }
    }
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
@Override
public List<String> getQueueNames(AmqpClient queueClient, Destination replyQueue, String topic) throws Exception {
    Message requestMessage = Message.Factory.create();
    Map<String, String> appProperties = new HashMap<>();
    appProperties.put(resourceProperty, "address." + topic);
    appProperties.put(operationProperty, "getQueueNames");
    requestMessage.setAddress(managementAddress);
    requestMessage.setApplicationProperties(new ApplicationProperties(appProperties));
    requestMessage.setReplyTo(replyQueue.getAddress());
    requestMessage.setBody(new AmqpValue("[]"));

    Future<Integer> sent = queueClient.sendMessages(managementAddress, requestMessage);
    assertThat(sent.get(30, TimeUnit.SECONDS), is(1));
    Logging.log.info("request sent");

    Future<List<Message>> received = queueClient.recvMessages(replyQueue.getAddress(), 1);
    assertThat(received.get(30, TimeUnit.SECONDS).size(), is(1));


    AmqpValue val = (AmqpValue) received.get().get(0).getBody();
    Logging.log.info("answer received: " + val.toString());
    String queues = val.getValue().toString();
    queues = queues.replaceAll("\\[|]|\"", "");


    return Arrays.asList(queues.split(","));
}
 

import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; //导入依赖的package包/类
private Message createOperationMessage(String resource, String operation) {
    Message message = Message.Factory.create();
    Map<String, String> properties = new LinkedHashMap<>();
    properties.put("_AMQ_ResourceName", resource);
    properties.put("_AMQ_OperationName", operation);
    properties.put("JMSReplyTo", replyTo);
    message.setReplyTo(replyTo);
    message.setApplicationProperties(new ApplicationProperties(properties));
    return message;
}