@Test public void testPut() { ConsumeQueueExt consumeQueueExt = genExt(); try { putSth(consumeQueueExt, true, false, unitCount); } finally { consumeQueueExt.destroy(); UtilAll.deleteFile(new File(storePath)); } }
public long put(final CqExtUnit cqExtUnit) { final int retryTimes = 3; try { int size = cqExtUnit.calcUnitSize(); if (size > CqExtUnit.MAX_EXT_UNIT_SIZE) { log.error("Size of cq ext unit is greater than {}, {}", CqExtUnit.MAX_EXT_UNIT_SIZE, cqExtUnit); return 1; } if (this.mappedFileQueue.getMaxOffset() + size > MAX_REAL_OFFSET) { log.warn("Capacity of ext is maximum!{}, {}", this.mappedFileQueue.getMaxOffset(), size); return 1; } if (this.tempContainer == null || this.tempContainer.capacity() < size) { this.tempContainer = ByteBuffer.allocate(size); } for (int i = 0; i < retryTimes; i++) { MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null || mappedFile.isFull()) { mappedFile = this.mappedFileQueue.getLastMappedFile(0); } if (mappedFile == null) { log.error("Create mapped file when save consume queue extend, {}", cqExtUnit); continue; } final int wrotePosition = mappedFile.getWrotePosition(); final int blankSize = this.mappedFileSize - wrotePosition - END_BLANK_DATA_LENGTH; if (size > blankSize) { fullFillToEnd(mappedFile, wrotePosition); log.info("No enough space(need:{}, has:{}) of file {}, so fill to end", size, blankSize, mappedFile.getFileName()); continue; } if (mappedFile.appendMessage(cqExtUnit.write(this.tempContainer), 0, size)) { return decorate(wrotePosition + mappedFile.getFileFromOffset()); } } } catch (Throwable e) { log.error("Save consume queue extend error, " + cqExtUnit, e); } return 1; }
ConsumeQueueExt { public long put(final CqExtUnit cqExtUnit) { final int retryTimes = 3; try { int size = cqExtUnit.calcUnitSize(); if (size > CqExtUnit.MAX_EXT_UNIT_SIZE) { log.error("Size of cq ext unit is greater than {}, {}", CqExtUnit.MAX_EXT_UNIT_SIZE, cqExtUnit); return 1; } if (this.mappedFileQueue.getMaxOffset() + size > MAX_REAL_OFFSET) { log.warn("Capacity of ext is maximum!{}, {}", this.mappedFileQueue.getMaxOffset(), size); return 1; } if (this.tempContainer == null || this.tempContainer.capacity() < size) { this.tempContainer = ByteBuffer.allocate(size); } for (int i = 0; i < retryTimes; i++) { MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null || mappedFile.isFull()) { mappedFile = this.mappedFileQueue.getLastMappedFile(0); } if (mappedFile == null) { log.error("Create mapped file when save consume queue extend, {}", cqExtUnit); continue; } final int wrotePosition = mappedFile.getWrotePosition(); final int blankSize = this.mappedFileSize - wrotePosition - END_BLANK_DATA_LENGTH; if (size > blankSize) { fullFillToEnd(mappedFile, wrotePosition); log.info("No enough space(need:{}, has:{}) of file {}, so fill to end", size, blankSize, mappedFile.getFileName()); continue; } if (mappedFile.appendMessage(cqExtUnit.write(this.tempContainer), 0, size)) { return decorate(wrotePosition + mappedFile.getFileFromOffset()); } } } catch (Throwable e) { log.error("Save consume queue extend error, " + cqExtUnit, e); } return 1; } }
ConsumeQueueExt { public long put(final CqExtUnit cqExtUnit) { final int retryTimes = 3; try { int size = cqExtUnit.calcUnitSize(); if (size > CqExtUnit.MAX_EXT_UNIT_SIZE) { log.error("Size of cq ext unit is greater than {}, {}", CqExtUnit.MAX_EXT_UNIT_SIZE, cqExtUnit); return 1; } if (this.mappedFileQueue.getMaxOffset() + size > MAX_REAL_OFFSET) { log.warn("Capacity of ext is maximum!{}, {}", this.mappedFileQueue.getMaxOffset(), size); return 1; } if (this.tempContainer == null || this.tempContainer.capacity() < size) { this.tempContainer = ByteBuffer.allocate(size); } for (int i = 0; i < retryTimes; i++) { MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null || mappedFile.isFull()) { mappedFile = this.mappedFileQueue.getLastMappedFile(0); } if (mappedFile == null) { log.error("Create mapped file when save consume queue extend, {}", cqExtUnit); continue; } final int wrotePosition = mappedFile.getWrotePosition(); final int blankSize = this.mappedFileSize - wrotePosition - END_BLANK_DATA_LENGTH; if (size > blankSize) { fullFillToEnd(mappedFile, wrotePosition); log.info("No enough space(need:{}, has:{}) of file {}, so fill to end", size, blankSize, mappedFile.getFileName()); continue; } if (mappedFile.appendMessage(cqExtUnit.write(this.tempContainer), 0, size)) { return decorate(wrotePosition + mappedFile.getFileFromOffset()); } } } catch (Throwable e) { log.error("Save consume queue extend error, " + cqExtUnit, e); } return 1; } ConsumeQueueExt(final String topic,
        final int queueId,
        final String storePath,
        final int mappedFileSize,
        final int bitMapLength); }
ConsumeQueueExt { public long put(final CqExtUnit cqExtUnit) { final int retryTimes = 3; try { int size = cqExtUnit.calcUnitSize(); if (size > CqExtUnit.MAX_EXT_UNIT_SIZE) { log.error("Size of cq ext unit is greater than {}, {}", CqExtUnit.MAX_EXT_UNIT_SIZE, cqExtUnit); return 1; } if (this.mappedFileQueue.getMaxOffset() + size > MAX_REAL_OFFSET) { log.warn("Capacity of ext is maximum!{}, {}", this.mappedFileQueue.getMaxOffset(), size); return 1; } if (this.tempContainer == null || this.tempContainer.capacity() < size) { this.tempContainer = ByteBuffer.allocate(size); } for (int i = 0; i < retryTimes; i++) { MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null || mappedFile.isFull()) { mappedFile = this.mappedFileQueue.getLastMappedFile(0); } if (mappedFile == null) { log.error("Create mapped file when save consume queue extend, {}", cqExtUnit); continue; } final int wrotePosition = mappedFile.getWrotePosition(); final int blankSize = this.mappedFileSize - wrotePosition - END_BLANK_DATA_LENGTH; if (size > blankSize) { fullFillToEnd(mappedFile, wrotePosition); log.info("No enough space(need:{}, has:{}) of file {}, so fill to end", size, blankSize, mappedFile.getFileName()); continue; } if (mappedFile.appendMessage(cqExtUnit.write(this.tempContainer), 0, size)) { return decorate(wrotePosition + mappedFile.getFileFromOffset()); } } } catch (Throwable e) { log.error("Save consume queue extend error, " + cqExtUnit, e); } return 1; } ConsumeQueueExt(final String topic,
        final int queueId,
        final String storePath,
        final int mappedFileSize,
        final int bitMapLength); static boolean isExtAddr(final long address); long unDecorate(final long address); long decorate(final long offset); CqExtUnit get(final long address); boolean get(final long address, final CqExtUnit cqExtUnit); long put(final CqExtUnit cqExtUnit); boolean load(); void checkSelf(); void recover(); void truncateByMinAddress(final long minAddress); void truncateByMaxAddress(final long maxAddress); boolean flush(final int flushLeastPages); void destroy(); long getMaxAddress(); long getMinAddress(); }
ConsumeQueueExt { public long put(final CqExtUnit cqExtUnit) { final int retryTimes = 3; try { int size = cqExtUnit.calcUnitSize(); if (size > CqExtUnit.MAX_EXT_UNIT_SIZE) { log.error("Size of cq ext unit is greater than {}, {}", CqExtUnit.MAX_EXT_UNIT_SIZE, cqExtUnit); return 1; } if (this.mappedFileQueue.getMaxOffset() + size > MAX_REAL_OFFSET) { log.warn("Capacity of ext is maximum!{}, {}", this.mappedFileQueue.getMaxOffset(), size); return 1; } if (this.tempContainer == null || this.tempContainer.capacity() < size) { this.tempContainer = ByteBuffer.allocate(size); } for (int i = 0; i < retryTimes; i++) { MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null || mappedFile.isFull()) { mappedFile = this.mappedFileQueue.getLastMappedFile(0); } if (mappedFile == null) { log.error("Create mapped file when save consume queue extend, {}", cqExtUnit); continue; } final int wrotePosition = mappedFile.getWrotePosition(); final int blankSize = this.mappedFileSize - wrotePosition - END_BLANK_DATA_LENGTH; if (size > blankSize) { fullFillToEnd(mappedFile, wrotePosition); log.info("No enough space(need:{}, has:{}) of file {}, so fill to end", size, blankSize, mappedFile.getFileName()); continue; } if (mappedFile.appendMessage(cqExtUnit.write(this.tempContainer), 0, size)) { return decorate(wrotePosition + mappedFile.getFileFromOffset()); } } } catch (Throwable e) { log.error("Save consume queue extend error, " + cqExtUnit, e); } return 1; } ConsumeQueueExt(final String topic,
        final int queueId,
        final String storePath,
        final int mappedFileSize,
        final int bitMapLength); static boolean isExtAddr(final long address); long unDecorate(final long address); long decorate(final long offset); CqExtUnit get(final long address); boolean get(final long address, final CqExtUnit cqExtUnit); long put(final CqExtUnit cqExtUnit); boolean load(); void checkSelf(); void recover(); void truncateByMinAddress(final long minAddress); void truncateByMaxAddress(final long maxAddress); boolean flush(final int flushLeastPages); void destroy(); long getMaxAddress(); long getMinAddress(); static final int END_BLANK_DATA_LENGTH; static final long MAX_ADDR; static final long MAX_REAL_OFFSET; }