@Test public void testPutMessage() { when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult (PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK))); Boolean success = transactionBridge.putMessage(createMessageBrokerInner()); assertThat(success).isEqualTo(true); }
public boolean putMessage(MessageExtBrokerInner messageInner) { PutMessageResult putMessageResult = store.putMessage(messageInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { return true; } else { LOGGER.error("Put message failed, topic: {}, queueId: {}, msgId: {}", messageInner.getTopic(), messageInner.getQueueId(), messageInner.getMsgId()); return false; } }
TransactionalMessageBridge { public boolean putMessage(MessageExtBrokerInner messageInner) { PutMessageResult putMessageResult = store.putMessage(messageInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { return true; } else { LOGGER.error("Put message failed, topic: {}, queueId: {}, msgId: {}", messageInner.getTopic(), messageInner.getQueueId(), messageInner.getMsgId()); return false; } } }
TransactionalMessageBridge { public boolean putMessage(MessageExtBrokerInner messageInner) { PutMessageResult putMessageResult = store.putMessage(messageInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { return true; } else { LOGGER.error("Put message failed, topic: {}, queueId: {}, msgId: {}", messageInner.getTopic(), messageInner.getQueueId(), messageInner.getMsgId()); return false; } } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); }
TransactionalMessageBridge { public boolean putMessage(MessageExtBrokerInner messageInner) { PutMessageResult putMessageResult = store.putMessage(messageInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { return true; } else { LOGGER.error("Put message failed, topic: {}, queueId: {}, msgId: {}", messageInner.getTopic(), messageInner.getQueueId(), messageInner.getMsgId()); return false; } } 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 boolean putMessage(MessageExtBrokerInner messageInner) { PutMessageResult putMessageResult = store.putMessage(messageInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { return true; } else { LOGGER.error("Put message failed, topic: {}, queueId: {}, msgId: {}", messageInner.getTopic(), messageInner.getQueueId(), messageInner.getMsgId()); return false; } } 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();  }