@Test public void testAsyncPutHalfMessage() throws Exception { when(messageStore.asyncPutMessage(any(MessageExtBrokerInner.class))) .thenReturn(CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)))); CompletableFuture<PutMessageResult> result = transactionBridge.asyncPutHalfMessage(createMessageBrokerInner()); assertThat(result.get().getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK); }
public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) { return store.asyncPutMessage(parseHalfMessageInner(messageInner)); }
TransactionalMessageBridge { public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) { return store.asyncPutMessage(parseHalfMessageInner(messageInner)); } }
TransactionalMessageBridge { public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) { return store.asyncPutMessage(parseHalfMessageInner(messageInner)); } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); }
TransactionalMessageBridge { public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) { return store.asyncPutMessage(parseHalfMessageInner(messageInner)); } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); long fetchConsumeOffset(MessageQueue mq); Set<MessageQueue> fetchMessageQueues(String topic); void updateConsumeOffset(MessageQueue mq, long offset); PullResult getHalfMessage(int queueId, long offset, int nums); PullResult getOpMessage(int queueId, long offset, int nums); PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner); CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner); boolean putOpMessage(MessageExt messageExt, String opType); PutMessageResult putMessageReturnResult(MessageExtBrokerInner messageInner); boolean putMessage(MessageExtBrokerInner messageInner); MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt); MessageExtBrokerInner renewHalfMessageInner(MessageExt msgExt); MessageExt lookMessageByOffset(final long commitLogOffset); BrokerController getBrokerController(); }
TransactionalMessageBridge { public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) { return store.asyncPutMessage(parseHalfMessageInner(messageInner)); } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); long fetchConsumeOffset(MessageQueue mq); Set<MessageQueue> fetchMessageQueues(String topic); void updateConsumeOffset(MessageQueue mq, long offset); PullResult getHalfMessage(int queueId, long offset, int nums); PullResult getOpMessage(int queueId, long offset, int nums); PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner); CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner); boolean putOpMessage(MessageExt messageExt, String opType); PutMessageResult putMessageReturnResult(MessageExtBrokerInner messageInner); boolean putMessage(MessageExtBrokerInner messageInner); MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt); MessageExtBrokerInner renewHalfMessageInner(MessageExt msgExt); MessageExt lookMessageByOffset(final long commitLogOffset); BrokerController getBrokerController();  }