@Test public void testRenewImmunityHalfMessageInner() { MessageExt messageExt = createMessageBrokerInner(); final String offset = "123456789"; MessageExtBrokerInner msgInner = transactionBridge.renewImmunityHalfMessageInner(messageExt); MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET,offset); assertThat(msgInner).isNotNull(); Map<String,String> properties = msgInner.getProperties(); assertThat(properties).isNotNull(); String resOffset = properties.get(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET); assertThat(resOffset).isEqualTo(offset); }
public MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt) { MessageExtBrokerInner msgInner = renewHalfMessageInner(msgExt); String queueOffsetFromPrepare = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET); if (null != queueOffsetFromPrepare) { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(queueOffsetFromPrepare)); } else { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(msgExt.getQueueOffset())); } msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); return msgInner; }
TransactionalMessageBridge { public MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt) { MessageExtBrokerInner msgInner = renewHalfMessageInner(msgExt); String queueOffsetFromPrepare = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET); if (null != queueOffsetFromPrepare) { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(queueOffsetFromPrepare)); } else { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(msgExt.getQueueOffset())); } msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); return msgInner; } }
TransactionalMessageBridge { public MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt) { MessageExtBrokerInner msgInner = renewHalfMessageInner(msgExt); String queueOffsetFromPrepare = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET); if (null != queueOffsetFromPrepare) { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(queueOffsetFromPrepare)); } else { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(msgExt.getQueueOffset())); } msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); return msgInner; } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); }
TransactionalMessageBridge { public MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt) { MessageExtBrokerInner msgInner = renewHalfMessageInner(msgExt); String queueOffsetFromPrepare = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET); if (null != queueOffsetFromPrepare) { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(queueOffsetFromPrepare)); } else { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(msgExt.getQueueOffset())); } msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); return msgInner; } 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 MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt) { MessageExtBrokerInner msgInner = renewHalfMessageInner(msgExt); String queueOffsetFromPrepare = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET); if (null != queueOffsetFromPrepare) { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(queueOffsetFromPrepare)); } else { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET, String.valueOf(msgExt.getQueueOffset())); } msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); return msgInner; } 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();  }