### Question:
PushConsumerImpl implements PushConsumer { @Override public PushConsumer attachQueue(final String queueName, final MessageListener listener) { this.subscribeTable.put(queueName, listener); try { this.rocketmqPushConsumer.subscribe(queueName, "*"); } catch (MQClientException e) { throw new OMSRuntimeException("-1", String.format("RocketMQ push consumer can't attach to %s.", queueName)); } return this; } PushConsumerImpl(final KeyValue properties); @Override KeyValue properties(); @Override void resume(); @Override void suspend(); @Override boolean isSuspended(); @Override PushConsumer attachQueue(final String queueName, final MessageListener listener); @Override synchronized void startup(); @Override synchronized void shutdown();  }### Answer:
@Test public void testConsumeMessage() { 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("HELLO_QUEUE"); consumer.attachQueue("HELLO_QUEUE", new MessageListener() { @Override public void onMessage(final Message message, final ReceivedMessageContext context) { assertThat(message.headers().getString(MessageHeader.MESSAGE_ID)).isEqualTo("NewMsgId"); assertThat(((BytesMessage) message).getBody()).isEqualTo(testBody); context.ack(); } }); ((MessageListenerConcurrently) rocketmqPushConsumer .getMessageListener()).consumeMessage(Collections.singletonList(consumedMsg), null); }