ConsumeQueue { public void destroy() { this.maxPhysicOffset = -1; this.minLogicOffset = 0; this.mappedFileQueue.destroy(); if (isExtReadEnable()) { this.consumeQueueExt.destroy(); } } ConsumeQueue(
        final String topic,
        final int queueId,
        final String storePath,
        final int mappedFileSize,
        final DefaultMessageStore defaultMessageStore); boolean load(); void recover(); long getOffsetInQueueByTime(final long timestamp); void truncateDirtyLogicFiles(long phyOffet); long getLastOffset(); boolean flush(final int flushLeastPages); int deleteExpiredFile(long offset); void correctMinOffset(long phyMinOffset); long getMinOffsetInQueue(); void putMessagePositionInfoWrapper(DispatchRequest request); SelectMappedBufferResult getIndexBuffer(final long startIndex); ConsumeQueueExt.CqExtUnit getExt(final long offset); boolean getExt(final long offset, ConsumeQueueExt.CqExtUnit cqExtUnit); long getMinLogicOffset(); void setMinLogicOffset(long minLogicOffset); long rollNextFile(final long index); String getTopic(); int getQueueId(); long getMaxPhysicOffset(); void setMaxPhysicOffset(long maxPhysicOffset); void destroy(); long getMessageTotalInQueue(); long getMaxOffsetInQueue(); void checkSelf(); boolean isExtAddr(long tagsCode); void deleteExpiredFileByIndex(int startIndex); static final int CQ_STORE_UNIT_SIZE; }
@Test public void testPutMessagePositionInfo_buildCQRepeatedly() throws Exception { DefaultMessageStore messageStore = null; try { messageStore = gen(); int totalMessages = 10; for (int i = 0; i < totalMessages; i++) { putMsg(messageStore); } Thread.sleep(5); ConsumeQueue cq = messageStore.getConsumeQueueTable().get(topic).get(queueId); Method method = cq.getClass().getDeclaredMethod("putMessagePositionInfo", long.class, int.class, long.class, long.class); assertThat(method).isNotNull(); method.setAccessible(true); SelectMappedBufferResult result = messageStore.getCommitLog().getData(0); assertThat(result != null).isTrue(); DispatchRequest dispatchRequest = messageStore.getCommitLog().checkMessageAndReturnSize(result.getByteBuffer(), false, false); assertThat(cq).isNotNull(); Object dispatchResult = method.invoke(cq, dispatchRequest.getCommitLogOffset(), dispatchRequest.getMsgSize(), dispatchRequest.getTagsCode(), dispatchRequest.getConsumeQueueOffset()); assertThat(Boolean.parseBoolean(dispatchResult.toString())).isTrue(); } finally { if (messageStore != null) { messageStore.shutdown(); messageStore.destroy(); } deleteDirectory(storePath); } }