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

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

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

Message.setBody介绍

暂无

代码示例

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

public synchronized void sendMessages(String address, List<String> messages) {
  Queue<Message> queue = queues.computeIfAbsent(address, a -> new ArrayDeque<>());
  for (String data : messages) {
    Message message = Proton.message();
    message.setBody(new AmqpValue(data));
    queue.add(message);
  }
}

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

public Future<Integer> sendMessages(String address, List<String> messages, Predicate<Message> predicate) {
  List<Message> messageList = messages.stream()
      .map(body -> {
        Message message = Message.Factory.create();
        message.setBody(new AmqpValue(body));
        message.setAddress(address);
        return message;
      })
      .collect(Collectors.toList());
  return sendMessages(address, messageList, predicate);
}

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

/**
 * Creates a Message object with the given String contained as an AmqpValue body.
 *
 * @param body
 *          the string to set as an AmqpValue body
 * @return the message
 */
public static Message message(String body) {
 Message value = message();
 value.setBody(new AmqpValue(body));
 return value;
}

代码示例来源：origin: org.apache.qpid/proton-hawtdispatch

public Message createTextMessage(String value) {
  Message msg = Message.Factory.create();
  Section body = new AmqpValue(value);
  msg.setBody(body);
  return msg;
}

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

private static Message createRequestMessageFromValueBody(String operation, Object valueBody, Duration timeout, String associatedLinkName)
{
  Message requestMessage = Message.Factory.create();
  requestMessage.setBody(new AmqpValue(valueBody));
  HashMap applicationPropertiesMap = new HashMap();
  applicationPropertiesMap.put(ClientConstants.REQUEST_RESPONSE_OPERATION_NAME, operation);
  applicationPropertiesMap.put(ClientConstants.REQUEST_RESPONSE_TIMEOUT, timeout.toMillis());
  if(!StringUtil.isNullOrEmpty(associatedLinkName))
  {
    applicationPropertiesMap.put(ClientConstants.REQUEST_RESPONSE_ASSOCIATED_LINK_NAME, associatedLinkName);
  }
  requestMessage.setApplicationProperties(new ApplicationProperties(applicationPropertiesMap));
  return requestMessage;
}

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

/**
* Sets a described type into the body of an outgoing Message, throws
* an exception if this is an incoming message instance.
*
* @param described the described type value to store in the Message body.
* @throws IllegalStateException if the message is read only.
*/
public void setDescribedType(DescribedType described) throws IllegalStateException {
 checkReadOnly();
 AmqpValue body = new AmqpValue(described);
 getWrappedMessage().setBody(body);
}

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

/**
* Sets a String value into the body of an outgoing Message, throws
* an exception if this is an incoming message instance.
*
* @param value the String value to store in the Message body.
* @throws IllegalStateException if the message is read only.
*/
public void setText(String value) throws IllegalStateException {
 checkReadOnly();
 AmqpValue body = new AmqpValue(value);
 getWrappedMessage().setBody(body);
}

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

/**
 * Return a raw AMQP message
 *
 * @return
 */
public Message toAmqp() {
  Message message = ProtonHelper.message();
  message.setSubject(AMQP_SUBJECT);
  message.setCorrelationId(String.format(AmqpHelper.AMQP_CLIENT_PUBLISH_ADDRESS_TEMPLATE, this.clientId));
  message.setBody(new AmqpValue(this.topics));
  return message;
}

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

private Message doRequestResponse(long timeout, TimeUnit timeUnit, Message message, Object ... parameters) throws TimeoutException {
  JsonArray params = new JsonArray();
  for (Object param : parameters) {
    if (param == null) {
      params.addNull();
    } else {
      params.add(param);
    }
  }
  message.setBody(new AmqpValue(Json.encode(params)));
  return syncRequestClient.request(message, timeout, timeUnit);
}

代码示例来源：origin: org.apache.qpid/proton-hawtdispatch

public Message createBinaryMessage(byte value[], int offset, int len) {
    Message msg = Message.Factory.create();
    Data body = new Data(new Binary(value, offset,len));
    msg.setBody(body);
    return msg;
  }
}

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

/**
* Sets a byte array value into the body of an outgoing Message, throws
* an exception if this is an incoming message instance.
*
* @param bytes the byte array value to store in the Message body.
* @throws IllegalStateException if the message is read only.
*/
public void setBytes(byte[] bytes) throws IllegalStateException {
 checkReadOnly();
 Data body = new Data(new Binary(bytes));
 getWrappedMessage().setBody(body);
}

代码示例来源：origin: com.ibm.mqlight/mqlight-api

@Override
public <T> boolean sendJson(String topic, String json,
    Map<String, Object> properties, SendOptions sendOptions,
    CompletionListener<T> listener, T context)
throws StoppedException {
  final String methodName = "sendJson";
  logger.entry(this, methodName, topic, json, properties, sendOptions, listener, context);
  org.apache.qpid.proton.message.Message protonMsg = Proton.message();
  protonMsg.setBody(new AmqpValue(json));
  protonMsg.setContentType("application/json");
  final boolean result = send(topic, protonMsg, properties, sendOptions == null ? defaultSendOptions : sendOptions, listener, context);
  logger.exit(this, methodName, result);
  return result;
}

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

/**
   * Verifies that the endpoint forwards a request message via the event bus.
   */
  @Test
  public void testProcessMessageSendsRequestViaEventBus() {

    final Message msg = ProtonHelper.message();
    msg.setMessageId("4711");
    msg.setSubject(RegistrationConstants.ACTION_ASSERT);
    msg.setBody(new AmqpValue(new JsonObject().put("temp", 15).encode()));
    MessageHelper.annotate(msg, resource);
    endpoint.processRequest(msg, resource, Constants.PRINCIPAL_ANONYMOUS);

    verify(eventBus).send(eq(RegistrationConstants.EVENT_BUS_ADDRESS_REGISTRATION_IN), any(JsonObject.class), any(DeliveryOptions.class));
  }
}

代码示例来源：origin: Azure/azure-event-hubs-java

private int getSize(final EventDataImpl eventData, final boolean isFirst) {

    final Message amqpMessage = this.partitionKey != null ? eventData.toAmqpMessage(this.partitionKey) : eventData.toAmqpMessage();
    int eventSize = amqpMessage.encode(this.eventBytes, 0, maxMessageSize); // actual encoded bytes size
    eventSize += 16; // data section overhead

    if (isFirst) {
      amqpMessage.setBody(null);
      amqpMessage.setApplicationProperties(null);
      amqpMessage.setProperties(null);
      amqpMessage.setDeliveryAnnotations(null);

      eventSize += amqpMessage.encode(this.eventBytes, 0, maxMessageSize);
    }

    return eventSize;
  }
}

代码示例来源：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: org.apache.beam/beam-sdks-java-io-amqp

@Test
public void encodeDecodeTooMuchLargerMessage() throws Exception {
 thrown.expect(CoderException.class);
 Message message = Message.Factory.create();
 message.setAddress("address");
 message.setSubject("subject");
 String body = Joiner.on("").join(Collections.nCopies(64 * 1024 * 1024, " "));
 message.setBody(new AmqpValue(body));
 AmqpMessageCoder coder = AmqpMessageCoder.of();
 byte[] encoded = CoderUtils.encodeToByteArray(coder, message);
}

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

/**
 * Verifies that a message containing a non Data section body
 * does not pass the filter.
 */
@Test
public void testVerifyFailsForNonDataSectionBody() {
  // GIVEN a message with an unsupported subject
  final Message msg = givenAValidMessageWithoutBody(CredentialsConstants.CredentialsAction.get);
  msg.setBody(new AmqpValue(BILLIE_HASHED_PASSWORD.encode()));
  msg.setContentType("application/json");
  // WHEN receiving the message via a link with any tenant
  final boolean filterResult = CredentialsMessageFilter.verify(target, msg);
  // THEN message validation fails
  assertFalse(filterResult);
}

代码示例来源：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());
}

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

/**
 * Verifies that a valid message passes the filter.
 */
@Test
public void testVerifySucceedsForValidGetAction() {
  // GIVEN a credentials message for user billie
  final Message msg = givenAValidMessageWithoutBody(CredentialsConstants.CredentialsAction.get);
  msg.setBody(new Data(new Binary(BILLIE_HASHED_PASSWORD.toBuffer().getBytes())));
  msg.setContentType("application/json");
  // WHEN receiving the message via a link with any tenant
  final boolean filterResult = CredentialsMessageFilter.verify(target, msg);
  // THEN message validation succeeds
  assertTrue(filterResult);
}

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

@Test
public void testAMQP_to_JSON_VerifyBodyWithAmqpValueString() {
 String testContent = "myTestContent";
 Message protonMsg = Proton.message();
 protonMsg.setBody(new AmqpValue(testContent));
 JsonObject jsonObject = translator.convertToJsonObject(protonMsg);
 assertNotNull("expected converted msg", jsonObject);
 assertTrue("expected body element key to be present", jsonObject.containsKey(AmqpConstants.BODY));
 assertNotNull("expected body element value to be non-null", jsonObject.getValue(AmqpConstants.BODY));
 assertEquals("body value not as expected", testContent, jsonObject.getValue(AmqpConstants.BODY));
 assertTrue("expected body_type element key to be present", jsonObject.containsKey(AmqpConstants.BODY_TYPE));
 assertEquals("unexpected body_type value", AmqpConstants.BODY_TYPE_VALUE,
   jsonObject.getValue(AmqpConstants.BODY_TYPE));
}

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

Message

相关文章

    查看更多
    微信公众号

    热门标签

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

    最新文章

    更多

    Message类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com