@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 onReceived(Message message, Context context) { assertThat(message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID)).isEqualTo("NewMsgId"); assertThat(((BytesMessage) message).getBody(byte[].class)).isEqualTo(testBody); context.ack(); } }); ((MessageListenerConcurrently) rocketmqPushConsumer .getMessageListener()).consumeMessage(Collections.singletonList(consumedMsg), null); }
@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 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 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); }
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 attributes(); @Override void resume(); @Override void suspend(); @Override void suspend(long timeout); @Override boolean isSuspended(); @Override PushConsumer attachQueue(final String queueName, final MessageListener listener); @Override PushConsumer attachQueue(String queueName, MessageListener listener, KeyValue attributes); @Override PushConsumer detachQueue(String queueName); @Override void addInterceptor(ConsumerInterceptor interceptor); @Override void removeInterceptor(ConsumerInterceptor interceptor); @Override synchronized void startup(); @Override synchronized void shutdown(); }
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 attributes(); @Override void resume(); @Override void suspend(); @Override void suspend(long timeout); @Override boolean isSuspended(); @Override PushConsumer attachQueue(final String queueName, final MessageListener listener); @Override PushConsumer attachQueue(String queueName, MessageListener listener, KeyValue attributes); @Override PushConsumer detachQueue(String queueName); @Override void addInterceptor(ConsumerInterceptor interceptor); @Override void removeInterceptor(ConsumerInterceptor interceptor); @Override synchronized void startup(); @Override synchronized void shutdown();  }