2 * Copyright (c) 2024 PANTHEON.tech, s.r.o. and others. All rights reserved.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
16 package io.atomix.storage.journal;
18 import static com.google.common.base.Verify.verify;
19 import static java.util.Objects.requireNonNull;
21 import io.netty.buffer.ByteBuf;
22 import io.netty.buffer.Unpooled;
23 import org.eclipse.jdt.annotation.Nullable;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
27 final class JournalSegmentReader {
28 private static final Logger LOG = LoggerFactory.getLogger(JournalSegmentReader.class);
30 private final JournalSegment segment;
31 private final FileReader fileReader;
32 private final int maxSegmentSize;
33 private final int maxEntrySize;
37 JournalSegmentReader(final JournalSegment segment, final FileReader fileReader, final int maxEntrySize) {
38 this.segment = requireNonNull(segment);
39 this.fileReader = requireNonNull(fileReader);
40 maxSegmentSize = segment.file().maxSize();
41 this.maxEntrySize = maxEntrySize;
45 * Return the current position.
47 * @return current position.
54 * Set the file position.
56 * @param position new position
58 void setPosition(final int position) {
59 verify(position >= JournalSegmentDescriptor.BYTES && position < maxSegmentSize,
60 "Invalid position %s", position);
61 this.position = position;
62 fileReader.invalidateCache();
66 * Invalidate any cache that is present, so that the next read is coherent with the backing file.
68 void invalidateCache() {
69 fileReader.invalidateCache();
73 * Reads the next binary data block
75 * @param index entry index
76 * @return The binary data, or {@code null}
78 @Nullable ByteBuf readBytes(final long index) {
79 // Check if there is enough in the buffer remaining
80 final int remaining = maxSegmentSize - position - SegmentEntry.HEADER_BYTES;
82 // Not enough space in the segment, there can never be another entry
86 // Calculate maximum entry length not exceeding file size nor maxEntrySize
87 final var maxLength = Math.min(remaining, maxEntrySize);
88 final var buffer = fileReader.read(position, maxLength + SegmentEntry.HEADER_BYTES);
90 // Read the entry length
91 final var length = buffer.getInt(0);
92 if (length < 1 || length > maxLength) {
93 // Invalid length, make sure next read re-tries
98 // Read the entry checksum
99 final int checksum = buffer.getInt(Integer.BYTES);
101 // Slice off the entry's bytes
102 final var entryBuffer = buffer.slice(SegmentEntry.HEADER_BYTES, length);
103 // If the stored checksum does not equal the computed checksum, do not proceed further
104 final var computed = SegmentEntry.computeChecksum(entryBuffer);
105 if (checksum != computed) {
106 LOG.warn("Expected checksum {}, computed {}", Integer.toHexString(checksum), Integer.toHexString(computed));
112 position += SegmentEntry.HEADER_BYTES + length;
115 return Unpooled.buffer(length).writeBytes(entryBuffer.rewind());
122 segment.closeReader(this);