@Test public void testSubmitConsumeRequest() throws Exception { byte[] body = new byte[] {'1', '2', '3'}; MessageExt consumedMsg = new MessageExt(); consumedMsg.setMsgId("NewMsgId"); consumedMsg.setBody(body); consumedMsg.putUserProperty(NonStandardKeys.MESSAGE_DESTINATION, "TOPIC"); consumedMsg.setTopic("HELLO_QUEUE"); when(consumeRequest.getMessageExt()).thenReturn(consumedMsg); localMessageCache.submitConsumeRequest(consumeRequest); assertThat(localMessageCache.poll()).isEqualTo(consumedMsg); }
void submitConsumeRequest(ConsumeRequest consumeRequest) { try { consumeRequestCache.put(consumeRequest); } catch (InterruptedException ignore) { } }
LocalMessageCache implements ServiceLifecycle { void submitConsumeRequest(ConsumeRequest consumeRequest) { try { consumeRequestCache.put(consumeRequest); } catch (InterruptedException ignore) { } } }
LocalMessageCache implements ServiceLifecycle { void submitConsumeRequest(ConsumeRequest consumeRequest) { try { consumeRequestCache.put(consumeRequest); } catch (InterruptedException ignore) { } } LocalMessageCache(final DefaultMQPullConsumer rocketmqPullConsumer, final ClientConfig clientConfig); }
LocalMessageCache implements ServiceLifecycle { void submitConsumeRequest(ConsumeRequest consumeRequest) { try { consumeRequestCache.put(consumeRequest); } catch (InterruptedException ignore) { } } LocalMessageCache(final DefaultMQPullConsumer rocketmqPullConsumer, final ClientConfig clientConfig); @Override void startup(); @Override void shutdown(); }
LocalMessageCache implements ServiceLifecycle { void submitConsumeRequest(ConsumeRequest consumeRequest) { try { consumeRequestCache.put(consumeRequest); } catch (InterruptedException ignore) { } } LocalMessageCache(final DefaultMQPullConsumer rocketmqPullConsumer, final ClientConfig clientConfig); @Override void startup(); @Override void shutdown();  }