### Question:
PullConsumerImpl implements PullConsumer { @Override public Message poll() { MessageExt rmqMsg = localMessageCache.poll(); return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg); } PullConsumerImpl(final String queueName, final KeyValue properties); @Override KeyValue properties(); @Override Message poll(); @Override Message poll(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();  }### Answer:
@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.poll(); assertThat(message.headers().getString(MessageHeader.MESSAGE_ID)).isEqualTo("NewMsgId"); assertThat(((BytesMessage) message).getBody()).isEqualTo(testBody); }
@Test public void testPoll_WithTimeout() { Message message = consumer.poll(); assertThat(message).isNull(); message = consumer.poll(OMS.newKeyValue().put(PropertyKeys.OPERATION_TIMEOUT, 100)); assertThat(message).isNull(); }