org.apache.qpid.proton.message.Message.getBody()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(11.3k)|赞(0)|评价(0)|浏览(159)

本文整理了Java中org.apache.qpid.proton.message.Message.getBody()方法的一些代码示例，展示了Message.getBody()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Message.getBody()方法的具体详情如下：
包路径：org.apache.qpid.proton.message.Message
类名称：Message
方法名：getBody

Message.getBody介绍

暂无

代码示例

代码示例来源：origin: vert-x3/vertx-examples

@Override
 public void start() throws Exception {
  ProtonClient client = ProtonClient.create(vertx);

  client.connect("localhost", 5672, res -> {
   if(!res.succeeded()) {
    System.out.println("Connect failed: " + res.cause());
    return;
   }

   ProtonConnection connection = res.result();
   connection.open();

   connection.createReceiver(address).handler((delivery, msg) -> {
    String content = (String) ((AmqpValue)  msg.getBody()).getValue();
    System.out.println("Received message with content: " + content);

    // By default, receivers automatically accept (and settle) the delivery
    // when the handler returns, if no other disposition has been applied.
    // To change this and always manage dispositions yourself, use the
    // setAutoAccept method on the receiver.
   }).open();
  });
 }
}

代码示例来源：origin: vert-x3/vertx-examples

Section body = msg.getBody();
if (body instanceof AmqpValue) {
 String content = (String) ((AmqpValue) body).getValue();

代码示例来源：origin: eclipse/hono

/**
 * Checks if a message's body consists of an AMQP <em>Data</em> section.
 * 
 * @param message The message to check.
 * @return {@code true} if the body consists of a Data section, {@code false} otherwise.
 * @throws NullPointerException If message is {@code null}.
 */
public static boolean hasDataBody(final Message message) {
  Objects.requireNonNull(message);
  return message.getBody() instanceof Data;
}

代码示例来源：origin: Azure/azure-service-bus-java

public static Map getResponseBody(Message responseMessage)
{
  return (Map)((AmqpValue)responseMessage.getBody()).getValue();				
}

代码示例来源：origin: EnMasseProject/enmasse

public synchronized List<String> recvMessages(String address, int numMessages) {
    Queue<Message> queue = queues.get(address);
    if (queue == null) {
      return null;
    }
    List<String> messages = new ArrayList<>();
    while (numMessages > 0) {
      Message message = queue.poll();
      if (message == null) {
        throw new RuntimeException("No more messages, " + numMessages + " remains");
      }
      messages.add((String)((AmqpValue)message.getBody()).getValue());
      numMessages--;
    }
    return messages;
  }
}

代码示例来源：origin: eclipse/hono

private void handleMessage(final String msgType, final Message msg) {
  final Data body = (Data) msg.getBody();
  LOG.debug("Type: [{}] and Message: [{}]", msgType, body != null ? body.getValue().toString() : "");
}

代码示例来源：origin: EnMasseProject/enmasse

private String doOperationWithStringResult(String resource, String operation, Object ... parameters) throws TimeoutException {
  Message response = doOperation(resource, operation, parameters);
  String payload = (String) ((AmqpValue)response.getBody()).getValue();
  JsonArray json = new JsonArray(payload);
  return json.getString(0);
}

代码示例来源：origin: EnMasseProject/enmasse

public long getQueueMessageCount(String queueName) throws TimeoutException {
  log.info("Checking message count for queue {} on broker {}", queueName, syncRequestClient.getRemoteContainer());
  Message response = doAttribute("queue." + queueName, "messageCount");
  String payload = (String) ((AmqpValue)response.getBody()).getValue();
  JsonArray json = new JsonArray(payload);
  return json.getLong(0);
}

代码示例来源：origin: EnMasseProject/enmasse

public String getQueueAddress(String queueName) throws TimeoutException {
  log.info("Checking queue address for queue {} on broker {}", queueName, syncRequestClient.getRemoteContainer());
  Message response = doOperation("queue." + queueName, "getAddress");
  String payload = (String) ((AmqpValue)response.getBody()).getValue();
  JsonArray json = new JsonArray(payload);
  return json.getString(0);
}

代码示例来源：origin: EnMasseProject/enmasse

public Set<String> getDivertNames() throws TimeoutException {
  log.info("Retrieving divert names");
  Message response = doOperation("broker", "getDivertNames");
  Set<String> diverts = new LinkedHashSet<>();
  JsonArray payload = new JsonArray((String)((AmqpValue)response.getBody()).getValue());
  for (int i = 0; i < payload.size(); i++) {
    JsonArray inner = payload.getJsonArray(i);
    for (int j = 0; j < inner.size(); j++) {
      diverts.add(inner.getString(j));
    }
  }
  return diverts;
}

代码示例来源：origin: EnMasseProject/enmasse

@Override
public List<String> getQueueNames(AmqpClient queueClient, Destination replyQueue, String topic) throws Exception {
  Message requestMessage = Message.Factory.create();
  Map<String, Object> 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(String.format("Sender failed, expected %d messages", 1), sent.get(30, TimeUnit.SECONDS), is(1));
  log.info("request sent");
  Future<List<Message>> received = queueClient.recvMessages(replyQueue.getAddress(), 1);
  assertThat(String.format("Receiver failed, expected %d messages", 1),
      received.get(30, TimeUnit.SECONDS).size(), is(1));
  AmqpValue val = (AmqpValue) received.get().get(0).getBody();
  log.info("answer received: " + val.toString());
  String queues = val.getValue().toString();
  queues = queues.replaceAll("\\[|]|\"", "");
  return Arrays.asList(queues.split(","));
}

代码示例来源：origin: EnMasseProject/enmasse

private static List<List<String>> collectRouter(SyncRequestClient client, String entityType, List<String> attributeNames) throws Exception {
    Map<String, Object> properties = new LinkedHashMap<>();
    properties.put("operation", "QUERY");
    properties.put("entityType", entityType);
    Map<String, Object> body = new LinkedHashMap<>();

    body.put("attributeNames", attributeNames);

    Message message = Proton.message();
    message.setApplicationProperties(new ApplicationProperties(properties));
    message.setBody(new AmqpValue(body));

    Message response = client.request(message, 10, TimeUnit.SECONDS);
    AmqpValue value = (AmqpValue) response.getBody();
    Map<?,?> values = (Map<?,?>) value.getValue();

    @SuppressWarnings("unchecked")
    List<List<String>> results = (List<List<String>>) values.get("results");
    return results;
  }
}

代码示例来源：origin: EnMasseProject/enmasse

/**
 * Return an AMQP_UNSUBSCRIBE message from the raw AMQP one
 *
 * @param message   raw AMQP message
 * @return  AMQP_UNSUBSCRIBE message
 */
@SuppressWarnings("unchecked")
public static AmqpUnsubscribeMessage from(Message message) {
  if (!message.getSubject().equals(AMQP_SUBJECT)) {
    throw new IllegalArgumentException(String.format("AMQP message subject is no %s", AMQP_SUBJECT));
  }
  Section section = message.getBody();
  if ((section != null) && (section instanceof AmqpValue)) {
    List<String> topics = (List<String>) ((AmqpValue) section).getValue();
    return new AmqpUnsubscribeMessage(AmqpHelper.getClientIdFromPublishAddress((String) message.getCorrelationId()),
                     topics);
  } else {
    throw new IllegalArgumentException("AMQP message wrong body type");
  }
}

代码示例来源：origin: EnMasseProject/enmasse

public Set<String> getQueueNames() throws TimeoutException {
  log.info("Retrieving queue names for broker {}", syncRequestClient.getRemoteContainer());
  Message response = doOperation("broker", "getQueueNames");
  Set<String> queues = new LinkedHashSet<>();
  JsonArray payload = new JsonArray((String)((AmqpValue)response.getBody()).getValue());
  for (int i = 0; i < payload.size(); i++) {
    JsonArray inner = payload.getJsonArray(i);
    for (int j = 0; j < inner.size(); j++) {
      String queueName = inner.getString(j);
      if (!queueName.equals(syncRequestClient.getReplyTo())) {
        queues.add(queueName);
      }
    }
  }
  return queues;
}

代码示例来源：origin: EnMasseProject/enmasse

public Set<String> getConnectorNames() throws TimeoutException {
    log.info("Retrieving conector names for broker {}", syncRequestClient.getRemoteContainer());
    Message response = doOperation("broker", "getConnectorServices");

    Set<String> connectors = new LinkedHashSet<>();
    JsonArray payload = new JsonArray((String)((AmqpValue)response.getBody()).getValue());
    for (int i = 0; i < payload.size(); i++) {
      JsonArray inner = payload.getJsonArray(i);
      for (int j = 0; j < inner.size(); j++) {
        String connector = inner.getString(j);
        if (!connector.equals("amqp-connector")) {
          connectors.add(connector);
        }
      }
    }
    return connectors;
  }
}

代码示例来源：origin: io.vertx/vertx-amqp-bridge

private void doJSON_to_AMQP_VerifyStringBodyTestImpl(boolean setBodyType) {
 String testContent = "myTestContent";
 JsonObject jsonObject = new JsonObject();
 jsonObject.put(AmqpConstants.BODY, testContent);
 if(setBodyType){
  jsonObject.put(AmqpConstants.BODY_TYPE, AmqpConstants.BODY_TYPE_VALUE);
 }
 Message protonMsg = translator.convertToAmqpMessage(jsonObject);
 assertNotNull("Expected converted msg", protonMsg);
 Section body = protonMsg.getBody();
 assertTrue("Unexpected body type", body instanceof AmqpValue);
 assertEquals("Unexpected message body value", testContent, ((AmqpValue) body).getValue());
}

代码示例来源：origin: io.vertx/vertx-amqp-bridge

@Test
public void testJSON_to_AMQP_VerifyDataBody() {
 String testContent = "myTestContent";
 JsonObject jsonObject = new JsonObject();
 jsonObject.put(AmqpConstants.BODY, testContent.getBytes(StandardCharsets.UTF_8));
 jsonObject.put(AmqpConstants.BODY_TYPE, AmqpConstants.BODY_TYPE_DATA);
 Message protonMsg = translator.convertToAmqpMessage(jsonObject);
 assertNotNull("Expected converted msg", protonMsg);
 Section body = protonMsg.getBody();
 assertTrue("Unexpected body type", body instanceof Data);
 assertNotNull("Unexpected body content", body);
 assertEquals("Unexpected message body value", new Binary(testContent.getBytes(StandardCharsets.UTF_8)),
   ((Data) body).getValue());
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-io-amqp

@Test
 public void encodeDecodeLargeMessage() throws Exception {
  Message message = Message.Factory.create();
  message.setAddress("address");
  message.setSubject("subject");
  String body = Joiner.on("").join(Collections.nCopies(32 * 1024 * 1024, " "));
  message.setBody(new AmqpValue(body));

  AmqpMessageCoder coder = AmqpMessageCoder.of();

  Message clone = CoderUtils.clone(coder, message);

  clone.getBody().toString().equals(message.getBody().toString());
 }
}

代码示例来源：origin: apache/activemq-artemis

private void validateMessage(byte[] expectedPayload, int msgNum, AmqpMessage message) {
 assertNotNull("failed at " + msgNum, message);
 Section body = message.getWrappedMessage().getBody();
 assertNotNull("No message body for msg " + msgNum, body);
 assertTrue("Unexpected message body type for msg " + body.getClass(), body instanceof Data);
 assertEquals("Unexpected body content for msg", new Binary(expectedPayload, 0, expectedPayload.length), ((Data) body).getValue());
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-io-amqp

@Test
public void encodeDecode() throws Exception {
 Message message = Message.Factory.create();
 message.setBody(new AmqpValue("body"));
 message.setAddress("address");
 message.setSubject("test");
 AmqpMessageCoder coder = AmqpMessageCoder.of();
 Message clone = CoderUtils.clone(coder, message);
 assertEquals("AmqpValue{body}", clone.getBody().toString());
 assertEquals("address", clone.getAddress());
 assertEquals("test", clone.getSubject());
}

内容来源于网络，如有侵权，请联系作者删除！

Message

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Message类方法

    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com