@Test public void testPoll() { final byte[] testBody = new byte[] {'a', 'b'}; MessageExt consumedMsg = new MessageExt(); consumedMsg.setMsgId("NewMsgId"); consumedMsg.setBody(testBody); consumedMsg.putUserProperty(NonStandardKeys.MESSAGE_DESTINATION, "TOPIC"); consumedMsg.setTopic(queueName); when(localMessageCache.poll()).thenReturn(consumedMsg); Message message = consumer.receive(); assertThat(message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID)).isEqualTo("NewMsgId"); assertThat(((BytesMessage) message).getBody(byte[].class)).isEqualTo(testBody); }
@Override public Message receive() { MessageExt rmqMsg = localMessageCache.poll(); return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg); }
PullConsumerImpl implements PullConsumer { @Override public Message receive() { MessageExt rmqMsg = localMessageCache.poll(); return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg); } }
PullConsumerImpl implements PullConsumer { @Override public Message receive() { MessageExt rmqMsg = localMessageCache.poll(); return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg); } PullConsumerImpl(final KeyValue properties); }
PullConsumerImpl implements PullConsumer { @Override public Message receive() { MessageExt rmqMsg = localMessageCache.poll(); return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg); } PullConsumerImpl(final KeyValue properties); @Override KeyValue attributes(); @Override PullConsumer attachQueue(String queueName); @Override PullConsumer attachQueue(String queueName, KeyValue attributes); @Override PullConsumer detachQueue(String queueName); @Override Message receive(); @Override Message receive(final KeyValue properties); @Override void ack(final String messageId); @Override void ack(final String messageId, final KeyValue properties); @Override synchronized void startup(); @Override synchronized void shutdown(); }
PullConsumerImpl implements PullConsumer { @Override public Message receive() { MessageExt rmqMsg = localMessageCache.poll(); return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg); } PullConsumerImpl(final KeyValue properties); @Override KeyValue attributes(); @Override PullConsumer attachQueue(String queueName); @Override PullConsumer attachQueue(String queueName, KeyValue attributes); @Override PullConsumer detachQueue(String queueName); @Override Message receive(); @Override Message receive(final KeyValue properties); @Override void ack(final String messageId); @Override void ack(final String messageId, final KeyValue properties); @Override synchronized void startup(); @Override synchronized void shutdown();  }