feat: implement region file reassembly and improve bundle format
- Add fastutil and lz4-java dependencies for efficient data structures and compression - Implement VarInt/VarLong serialization for more compact bundle headers - Redesign bundle header format to include timestamps and use variable-length integers - Replace GZIP-based compression with region file reassembly for better compression of chunk data - Add RegionFile and RegionFileReassembler classes for handling Minecraft region files - Implement configurable compression types (GZIP, ZLIB, LZ4, NONE) for region files - Add timestamp filtering and duplicate handling to BundleReader - Improve parallel processing capabilities with configurable worker threads - Add JUnit 5 testing support and configure shadowJar minimization - Update file extensions from .jpack to .jbp.zst for clarity - Enhance error handling and logging throughout the codebase - Add comprehensive region file manipulation including chunk-level recompression - Implement memory-efficient region file parsing using Netty ByteBuf - Add support for dictionary-based Zstd compression - Update backup worker to use new bundle format - Various performance optimizations and bug fixes
This commit is contained in:
1 parent
28280bc162
commit
d48b11b58b
14 files changed
+544
-83
No files matched your search
@@ -11,8 +11,20 @@ dependencies {
|
||||
compileOnly 'io.netty:netty-all:4.2.6.Final'
|
||||
api "io.ib67.kiwi:lang:1.2.3-SNAPSHOT"
|
||||
api 'io.netty:netty-all:4.2.6.Final'
|
||||
api 'it.unimi.dsi:fastutil:8.5.15'
|
||||
api 'org.lz4:lz4-java:1.8.0'
|
||||
compileOnly 'org.lz4:lz4-java:1.8.0'
|
||||
testImplementation 'org.junit.jupiter:junit-jupiter:5.13.4'
|
||||
testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
|
||||
}
|
||||
|
||||
shadowJar {
|
||||
minimize()
|
||||
}
|
||||
|
||||
test {
|
||||
useJUnitPlatform()
|
||||
}
|
||||
|
||||
jar {
|
||||
manifest {
|
||||
|
||||
@@ -1,10 +1,19 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
/**
|
||||
* Header Format:
|
||||
* u16 MAGIC u64 time varint attribute varlong length varint length of name
|
||||
* @param attribute
|
||||
* @param time
|
||||
* @param length
|
||||
* @param name
|
||||
*/
|
||||
public record BundleEntry(
|
||||
short attribute,
|
||||
long fileLen,
|
||||
long time,
|
||||
long length,
|
||||
String name
|
||||
) {
|
||||
public static short ATTR_GUNZIP = 1;
|
||||
public static short ATTR_REASSEMBLE = 1;
|
||||
public static short MAGIC = (short) 0xA2E3;
|
||||
}
|
||||
@@ -1,38 +1,90 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import com.github.luben.zstd.Zstd;
|
||||
import com.github.luben.zstd.ZstdInputStreamNoFinalizer;
|
||||
import io.ib67.sfcraft.bundler.region.RegionFile;
|
||||
import io.ib67.sfcraft.bundler.region.RegionFileReassembler;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import lombok.Builder;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.RandomAccessFile;
|
||||
import java.nio.channels.Channels;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.StandardOpenOption;
|
||||
import java.util.HashMap;
|
||||
|
||||
public class BundleReader {
|
||||
public static void extract(Path bundlePath, Path destination) throws IOException {
|
||||
var buffer = new byte[4096];
|
||||
try (var in = new EntryInputStream(
|
||||
new ZstdInputStreamNoFinalizer(Files.newInputStream(bundlePath)))) {
|
||||
protected final long maxTime;
|
||||
protected final long minTime;
|
||||
private final byte[] buffer = new byte[4096];
|
||||
protected final boolean skipOldEntry;
|
||||
protected final ByteBufAllocator allocator;
|
||||
protected final byte[] dict;
|
||||
|
||||
@Builder
|
||||
public BundleReader(long maxTime, long minTime, boolean skipOldEntry, ByteBufAllocator allocator, byte[] dict) {
|
||||
this.maxTime = maxTime == 0 ? Long.MAX_VALUE : maxTime;
|
||||
this.minTime = minTime;
|
||||
this.skipOldEntry = skipOldEntry;
|
||||
this.allocator = allocator == null ? ByteBufAllocator.DEFAULT : allocator;
|
||||
this.dict = dict;
|
||||
}
|
||||
|
||||
public void extract(Path bundle, Path destination) throws IOException {
|
||||
var readMaps = new HashMap<String, Long>();
|
||||
try (var zstd = new ZstdInputStreamNoFinalizer(Files.newInputStream(bundle, StandardOpenOption.READ));
|
||||
var in = new EntryInputStream(zstd)) {
|
||||
if (dict != null) zstd.setDict(dict);
|
||||
while (in.available() > 0) {
|
||||
var entry = in.readEntry();
|
||||
var len = entry.fileLen();
|
||||
var dstPath = destination.resolve(Path.of(entry.name()));
|
||||
if (Files.notExists(dstPath.getParent())) {
|
||||
Files.createDirectories(dstPath.getParent());
|
||||
var time = entry.time();
|
||||
if (time < minTime || time > maxTime) {
|
||||
System.err.println("Skipping entry " + entry.name() + " (too old or too new)");
|
||||
continue;
|
||||
}
|
||||
try (var out = Files.newOutputStream(dstPath,
|
||||
StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
|
||||
var read = 0;
|
||||
var toRead = len;
|
||||
while (toRead > 0) {
|
||||
read = in.read(buffer, 0, (int) Math.min(buffer.length, toRead));
|
||||
toRead -= read;
|
||||
out.write(buffer, 0, read);
|
||||
}
|
||||
var lastSeen = readMaps.get(entry.name());
|
||||
if (lastSeen != null && skipOldEntry && entry.time() < lastSeen) {
|
||||
System.err.println("Skipping entry " + entry.name() + " (older than )");
|
||||
continue;
|
||||
}
|
||||
readMaps.put(entry.name(), time);
|
||||
|
||||
var path = destination.resolve(entry.name()); // todo prevent path underflow
|
||||
var parentDir = path.getParent();
|
||||
if (Files.notExists(parentDir)) Files.createDirectories(parentDir);
|
||||
if ((entry.attribute() & BundleEntry.ATTR_REASSEMBLE) != 0) {
|
||||
var entryData = allocator.buffer();
|
||||
var toRead = entry.length();
|
||||
var buffer = this.buffer;
|
||||
while (toRead > 0) {
|
||||
var read = in.read(buffer, 0, (int) Math.min(buffer.length, toRead));
|
||||
toRead -= read;
|
||||
entryData.writeBytes(buffer, 0, read);
|
||||
}
|
||||
try (var reassembler = new RegionFileReassembler(entryData)) {
|
||||
var result = reassembler.writeReassembled(allocator, RegionFile.CompressType.GZIP);
|
||||
try (var fc = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
|
||||
result.readBytes(fc, 0, result.readableBytes());
|
||||
} finally {
|
||||
result.release();
|
||||
}
|
||||
}
|
||||
} else {
|
||||
writePlain(path, entry, in);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private void writePlain(Path path, BundleEntry entry, EntryInputStream in) throws IOException {
|
||||
var buffer = this.buffer;
|
||||
try (var fc = Files.newOutputStream(path, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
|
||||
var toRead = entry.length();
|
||||
while (toRead > 0) {
|
||||
var read = in.read(buffer, 0, (int) Math.min(buffer.length, toRead));
|
||||
toRead -= read;
|
||||
fc.write(buffer, 0, read);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,8 +3,12 @@ package io.ib67.sfcraft.bundler;
|
||||
import com.github.luben.zstd.ZstdDictCompress;
|
||||
import com.github.luben.zstd.ZstdOutputStreamNoFinalizer;
|
||||
import io.ib67.kiwi.routine.Uni;
|
||||
import io.ib67.sfcraft.bundler.region.RegionFile;
|
||||
import io.ib67.sfcraft.bundler.region.RegionFileReassembler;
|
||||
import io.netty.buffer.AdaptiveByteBufAllocator;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import io.netty.buffer.ByteBufOutputStream;
|
||||
|
||||
import lombok.Builder;
|
||||
import lombok.SneakyThrows;
|
||||
|
||||
@@ -14,12 +18,11 @@ import java.nio.file.Path;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.function.UnaryOperator;
|
||||
import java.util.zip.GZIPInputStream;
|
||||
|
||||
public class BundleWriter implements Closeable {
|
||||
public static Comparator<Path> SORT = BundleWriter::preferRegion;
|
||||
protected static final ByteBufAllocator ALLOC = new AdaptiveByteBufAllocator();
|
||||
protected final boolean allowGunzip;
|
||||
protected final ByteBufAllocator allocator;
|
||||
protected final boolean allowReassemble;
|
||||
protected final int compressionLevel;
|
||||
protected final Path relativeRoot;
|
||||
protected final EntryOutputStream outputStream;
|
||||
@@ -33,16 +36,17 @@ public class BundleWriter implements Closeable {
|
||||
int maxWorkers,
|
||||
Path relativeRoot,
|
||||
OutputStream outputStream,
|
||||
byte[] dictionary
|
||||
byte[] dictionary, ByteBufAllocator allocator
|
||||
) {
|
||||
this.allowGunzip = allowGunzip;
|
||||
this.allowReassemble = allowGunzip;
|
||||
this.compressionLevel = compressionLevel;
|
||||
this.relativeRoot = relativeRoot;;
|
||||
this.relativeRoot = relativeRoot;
|
||||
this.allocator = allocator == null ? ByteBufAllocator.DEFAULT : allocator;
|
||||
var zstdOut = new ZstdOutputStreamNoFinalizer(outputStream, compressionLevel);
|
||||
if (dictionary != null) {
|
||||
zstdOut.setDict(new ZstdDictCompress(dictionary, compressionLevel));
|
||||
}
|
||||
if(maxWorkers > 1) zstdOut.setWorkers(maxWorkers);
|
||||
if (maxWorkers > 1) zstdOut.setWorkers(maxWorkers);
|
||||
this.outputStream = new EntryOutputStream(zstdOut);
|
||||
this.buffer = new byte[4096];
|
||||
}
|
||||
@@ -71,7 +75,7 @@ public class BundleWriter implements Closeable {
|
||||
var futures = new ArrayList<CompletableFuture<Path>>();
|
||||
for (int i = 0; i < parted.size(); i++) {
|
||||
var pathList = parted.get(i);
|
||||
var output = parentOfBundle.resolve("Worker_" + System.currentTimeMillis() + "_" + i + ".jpack");
|
||||
var output = parentOfBundle.resolve("Worker_" + System.currentTimeMillis() + "_" + i + ".jbp.zst");
|
||||
var future = CompletableFuture.supplyAsync(() -> createBundleWorker(pathList, output, config), executor);
|
||||
futures.add(future);
|
||||
}
|
||||
@@ -138,38 +142,45 @@ public class BundleWriter implements Closeable {
|
||||
}
|
||||
|
||||
public void write(Path path, boolean suggestGunzip) throws IOException {
|
||||
var size = Files.size(path);
|
||||
if (suggestGunzip) {
|
||||
boolean isGzip;
|
||||
try (var raf = new RandomAccessFile(path.toFile(), "r")) {
|
||||
isGzip = raf.read() == 0x8b && raf.read() == 0x1f;
|
||||
}
|
||||
if (isGzip) {
|
||||
var buffer = ALLOC.buffer((int) size * 2);
|
||||
try (var gunzip = new GZIPInputStream(new FileInputStream(path.toFile()))) {
|
||||
while (buffer.writeBytes(gunzip, 4096) != -1) ;
|
||||
outputStream.writeEntryHeader( // bug
|
||||
relativeRoot.relativize(path).toString(),
|
||||
BundleEntry.ATTR_GUNZIP,
|
||||
size
|
||||
);
|
||||
buffer.readBytes(outputStream, buffer.readableBytes());
|
||||
} catch (IOException e) {
|
||||
writePlain(path);
|
||||
} finally {
|
||||
buffer.release();
|
||||
}
|
||||
}
|
||||
//todo size limit
|
||||
writeReassembleInMem(path);
|
||||
return;
|
||||
}
|
||||
|
||||
writePlain(path);
|
||||
}
|
||||
|
||||
@SneakyThrows
|
||||
private void writeReassembleInMem(Path path) {
|
||||
if (!"region".equals(path.getParent().toString()) && !path.toString().endsWith(".mca")) {
|
||||
System.out.println("Mismatch "+path+", parent: "+path.getParent());
|
||||
writePlain(path);
|
||||
return;
|
||||
}
|
||||
System.out.println("Reassembling "+path);
|
||||
var uncompressed = allocator.buffer();
|
||||
var rPath = relativeRoot.toAbsolutePath().relativize(path.toAbsolutePath());
|
||||
try (var out = new EntryOutputStream(new ByteBufOutputStream(uncompressed));
|
||||
var rf = new RegionFileReassembler(path)) {
|
||||
var result = rf.writeReassembled(allocator, RegionFile.CompressType.NONE);
|
||||
var length = result.readableBytes();
|
||||
out.writeEntryHeader(rPath.toString(), BundleEntry.ATTR_REASSEMBLE, length, System.currentTimeMillis());
|
||||
result.readBytes(out, length);
|
||||
result.release();
|
||||
uncompressed.readBytes(outputStream, uncompressed.readableBytes());
|
||||
} catch (IOException e) {
|
||||
System.err.println("Failed to parse region " + path + ": " + e);
|
||||
writePlain(path);
|
||||
} finally {
|
||||
uncompressed.release();
|
||||
}
|
||||
}
|
||||
|
||||
private void writePlain(Path path) throws IOException {
|
||||
var rPath = relativeRoot.toAbsolutePath().relativize(path.toAbsolutePath());
|
||||
var buffer = this.buffer;
|
||||
var outputStream = this.outputStream;
|
||||
outputStream.writeEntryHeader(rPath.toString(), (short) 0, Files.size(path));
|
||||
outputStream.writeEntryHeader(rPath.toString(), (short) 0, Files.size(path), 0);
|
||||
var read = 0;
|
||||
try (var fs = Files.newInputStream(path)) {
|
||||
while ((read = fs.read(buffer)) > 0) {
|
||||
|
||||
@@ -9,8 +9,6 @@ import java.io.InputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
public class EntryInputStream extends FilterInputStream {
|
||||
private final ByteBuffer buffer;
|
||||
|
||||
/**
|
||||
* Creates a {@code FilterInputStream}
|
||||
* by assigning the argument {@code in}
|
||||
@@ -22,19 +20,18 @@ public class EntryInputStream extends FilterInputStream {
|
||||
*/
|
||||
protected EntryInputStream(InputStream in) {
|
||||
super(in);
|
||||
this.buffer = ByteBuffer.allocate(4 + 8 + 4);
|
||||
}
|
||||
|
||||
public BundleEntry readEntry() throws IOException {
|
||||
buffer.clear();
|
||||
in.read(buffer.array(), 0, 4+8+4);
|
||||
var magic = buffer.getShort();
|
||||
if (magic != BundleEntry.MAGIC) throw new IOException("Invalid magic number");
|
||||
var attribute = buffer.getShort();
|
||||
var fileLen = buffer.getLong();
|
||||
var lenName = buffer.getInt();
|
||||
var magic = (short) (in.read() << 8 | in.read()); // u16
|
||||
if (magic != BundleEntry.MAGIC) throw new IOException("Invalid magic number: "+Long.toHexString(magic));
|
||||
var time = (long) in.read() << 56 | (long) in.read() << 48 | (long) in.read() << 40 | (long) in.read() << 32 |
|
||||
(long) in.read() << 24 | in.read() << 16 | in.read() << 8 | in.read();
|
||||
var attribute = VarInts.readVarInt(in);
|
||||
var segmentLen = VarInts.readVarLong(in);
|
||||
var lenName = VarInts.readVarInt(in);
|
||||
var name = new String(in.readNBytes(lenName)).intern();
|
||||
return new BundleEntry(attribute, fileLen, name);
|
||||
return new BundleEntry((short) attribute, time, segmentLen, name);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
@@ -10,7 +11,6 @@ import java.nio.ByteBuffer;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
public class EntryOutputStream extends FilterOutputStream {
|
||||
protected final ByteBuffer buffer;
|
||||
/**
|
||||
* Creates an output stream filter built on top of the specified
|
||||
* underlying output stream.
|
||||
@@ -22,17 +22,26 @@ public class EntryOutputStream extends FilterOutputStream {
|
||||
*/
|
||||
public EntryOutputStream(OutputStream out) {
|
||||
super(out);
|
||||
this.buffer = ByteBuffer.allocate(4+8+4);
|
||||
}
|
||||
|
||||
public void writeEntryHeader(String path, short attribute, long fileLen) throws IOException {
|
||||
public void writeEntryHeader(String path, short attribute, long segmentLen, long time) throws IOException {
|
||||
var name = path.getBytes(StandardCharsets.UTF_8);
|
||||
buffer.clear();
|
||||
buffer.putShort(BundleEntry.MAGIC);
|
||||
buffer.putShort(attribute);
|
||||
buffer.putLong(fileLen);
|
||||
buffer.putInt(name.length);
|
||||
out.write(buffer.array());
|
||||
out.write(BundleEntry.MAGIC >> 8);
|
||||
out.write(BundleEntry.MAGIC & 0xFF); // u16 be
|
||||
var timeArr = new byte[] { // u64 be
|
||||
(byte) (time >> 56),
|
||||
(byte) (time >> 48),
|
||||
(byte) (time >> 40),
|
||||
(byte) (time >> 32),
|
||||
(byte) (time >> 24),
|
||||
(byte) (time >> 16),
|
||||
(byte) (time >> 8),
|
||||
(byte) time
|
||||
};
|
||||
out.write(timeArr);
|
||||
VarInts.writeVarInt(out, attribute);
|
||||
VarInts.writeVarLong(out, segmentLen);
|
||||
VarInts.writeVarInt(out, name.length);
|
||||
out.write(name);
|
||||
}
|
||||
|
||||
|
||||
@@ -17,12 +17,12 @@ public class Main {
|
||||
@SneakyThrows
|
||||
public static void main(String[] args) {
|
||||
var opts = ArgOpts.builder()
|
||||
.args(args).description("Tools for .jpack format bundle").programName("jpack").build();
|
||||
.args(args).description("Tools for .jbp.zst format bundle").programName("jpack").build();
|
||||
var in = opts.string("in", "Input. Can be directory or jpack bundle", null);
|
||||
var out = opts.string("out", "Output. Can be jpack bundle or directory", null);
|
||||
var help = opts.bool("help", false);
|
||||
var allowGunzip = opts.bool("gunzip",
|
||||
"Should we recompress NBT files for maximum compression rate (depends on compress)", true);
|
||||
var allowGunzip = opts.bool("reasm",
|
||||
"Should we reassemble NBT files for maximum compression rate", true);
|
||||
var parallel = opts.bool("parallel", "Use parallelized bundler. This conflicts with compress options", false);
|
||||
var parallelThreshold = opts.integer("threshold", "The threshold of parallel bundling", 512);
|
||||
var parallelCompression = opts.integer("parallel-compression", "The worker threads used for parallel compression. Do not confuse this with --parallel, which focus on bundling.\n" +
|
||||
@@ -48,16 +48,16 @@ public class Main {
|
||||
System.out.println("The output is already exists. Please move it to other places");
|
||||
return;
|
||||
}
|
||||
if (in.endsWith(".jpack")) {
|
||||
BundleReader.extract(inPath, outPath);
|
||||
} else if (out.endsWith(".jpack")) {
|
||||
if (!Files.isDirectory(inPath))
|
||||
throw new IllegalArgumentException("Your input should be a world directory.");
|
||||
if (in.endsWith(".jbp.zst")) {
|
||||
BundleReader.builder().build().extract(inPath, outPath);
|
||||
} else if (out.endsWith(".jbp.zst")) {
|
||||
UnaryOperator<BundleWriter.BundleWriterBuilder> cfg = it -> it
|
||||
.allowGunzip(allowGunzip)
|
||||
.relativeRoot(outPath.toAbsolutePath().getParent())
|
||||
.relativeRoot(inPath)
|
||||
.dictionary(dict)
|
||||
.compressionLevel(compressionLevel);
|
||||
if (!Files.isDirectory(inPath))
|
||||
throw new IllegalArgumentException("Your input should be a world directory.");
|
||||
List<Path> paths;
|
||||
try (var _paths = Files.walk(inPath)) {
|
||||
paths = _paths.toList();
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import lombok.SneakyThrows;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
|
||||
public class VarInts {
|
||||
public static void writeVarInt(OutputStream buf, int value) throws IOException {
|
||||
while (true) {
|
||||
if ((value & ~0x7F) == 0) {
|
||||
buf.write(value);
|
||||
return;
|
||||
}
|
||||
buf.write((value & 0x7F) | 0x80);
|
||||
value >>>= 7;
|
||||
}
|
||||
}
|
||||
|
||||
public static void writeVarLong(OutputStream buf, long value) throws IOException {
|
||||
while (true) {
|
||||
if ((value & ~0x7FL) == 0) {
|
||||
buf.write((int) value);
|
||||
return;
|
||||
}
|
||||
buf.write((int) ((value & 0x7F) | 0x80));
|
||||
value >>>= 7;
|
||||
}
|
||||
}
|
||||
|
||||
@SneakyThrows
|
||||
public static int readVarInt(InputStream buf) {
|
||||
int value = 0;
|
||||
int position = 0;
|
||||
byte currentByte;
|
||||
|
||||
while (true) {
|
||||
currentByte = (byte) buf.read();
|
||||
value |= (currentByte & 0x7F) << position;
|
||||
|
||||
if ((currentByte & 0x80) == 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
position += 7;
|
||||
|
||||
if (position >= 32) {
|
||||
throw new RuntimeException("VarInt is too big");
|
||||
}
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
@SneakyThrows
|
||||
public static long readVarLong(InputStream buf) {
|
||||
long value = 0;
|
||||
int position = 0;
|
||||
byte currentByte;
|
||||
|
||||
while (true) {
|
||||
currentByte = (byte) buf.read();
|
||||
value |= (long) (currentByte & 0x7F) << position;
|
||||
|
||||
if ((currentByte & 0x80) == 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
position += 7;
|
||||
|
||||
if (position >= 64) {
|
||||
throw new RuntimeException("VarLong is too big");
|
||||
}
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package io.ib67.sfcraft.bundler.region;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import io.netty.buffer.ByteBufOutputStream;
|
||||
import it.unimi.dsi.fastutil.shorts.Short2ObjectAVLTreeMap;
|
||||
import it.unimi.dsi.fastutil.shorts.Short2ObjectSortedMap;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import net.jpountz.lz4.LZ4BlockOutputStream;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.util.zip.DeflaterOutputStream;
|
||||
import java.util.zip.GZIPOutputStream;
|
||||
|
||||
public class RegionFile implements Closeable {
|
||||
|
||||
@RequiredArgsConstructor
|
||||
public enum CompressType {
|
||||
GZIP(1), ZLIB(2), NONE(3), LZ4(4);
|
||||
public final int index;
|
||||
|
||||
public static CompressType findByIndex(int i) {
|
||||
return switch (i) {
|
||||
case 1 -> GZIP;
|
||||
case 2 -> ZLIB;
|
||||
case 3 -> NONE;
|
||||
case 4 -> LZ4;
|
||||
default -> throw new IllegalArgumentException("Unknown compression type: " + i);
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
protected final Short2ObjectSortedMap<ByteBuf> regionData = new Short2ObjectAVLTreeMap<>();
|
||||
protected final ByteBufAllocator allocator;
|
||||
protected final ByteBuf timestamp;
|
||||
|
||||
public RegionFile(ByteBufAllocator allocator) {
|
||||
this.allocator = allocator;
|
||||
timestamp = allocator.buffer(4096);
|
||||
}
|
||||
|
||||
private short getChunkKey(int x, int z) {
|
||||
return (short) ((z & 0b111111) << 6 | (x & 0b111111));
|
||||
}
|
||||
|
||||
// this retains the input bytebuf.
|
||||
public void writeChunk(int chunkX, int chunkZ, CompressType type, ByteBuf data) throws IOException {
|
||||
if (chunkX > 32 || chunkZ > 32 || chunkX < 0 || chunkZ < 0)
|
||||
throw new IllegalArgumentException("Invalid chunkX+chunkZ: " + chunkX + " " + chunkZ);
|
||||
var key = getChunkKey(chunkX, chunkZ);
|
||||
var buf = allocator.buffer(5 + data.readableBytes());
|
||||
if (type == CompressType.NONE) {
|
||||
buf.writeInt(data.readableBytes());
|
||||
buf.writeByte(type.index);
|
||||
buf.writeBytes(data);
|
||||
} else {
|
||||
try (var out = findCompressor(type, new ByteBufOutputStream(buf))) {
|
||||
while (data.readableBytes() > 0) data.readBytes(out, data.readableBytes());
|
||||
var head = allocator.buffer(5);
|
||||
head.writeInt(buf.readableBytes());
|
||||
head.writeByte(type.index);
|
||||
var composite = allocator.compositeBuffer();
|
||||
composite.addComponent(true, head);
|
||||
composite.addComponent(true, buf);
|
||||
} catch (IOException e) {
|
||||
buf.release();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
regionData.put(key, buf);
|
||||
}
|
||||
|
||||
private OutputStream findCompressor(CompressType type, OutputStream out) throws IOException {
|
||||
return switch (type) {
|
||||
case LZ4 -> new LZ4BlockOutputStream(out);
|
||||
case GZIP -> new GZIPOutputStream(out);
|
||||
case ZLIB -> new DeflaterOutputStream(out);
|
||||
default -> throw new IllegalArgumentException("Unknown compression type: " + type);
|
||||
};
|
||||
}
|
||||
|
||||
public void updateChunkTime(int chunkX, int chunkZ, int timestamp) {
|
||||
if (chunkX > 32 || chunkZ > 32 || chunkX < 0 || chunkZ < 0)
|
||||
throw new IllegalArgumentException("Invalid chunkX+chunkZ: " + chunkX + " " + chunkZ);
|
||||
var index = (chunkX + 32 * chunkZ) * 4;
|
||||
this.timestamp.setInt(index, timestamp);
|
||||
}
|
||||
|
||||
public ByteBuf writeOutput() {
|
||||
var sect = allocator.buffer(8192);
|
||||
var compositeByteBuf = allocator.compositeBuffer(2 + regionData.size());
|
||||
var beginSector = 2;
|
||||
compositeByteBuf.addComponent(true, sect);
|
||||
timestamp.retain();
|
||||
compositeByteBuf.addComponent(true, timestamp);
|
||||
for (var entry : regionData.short2ObjectEntrySet()) {
|
||||
var chunkKey = entry.getShortKey();
|
||||
var x = chunkKey & 0b111111;
|
||||
var z = (chunkKey >>> 6) & 0b111111;
|
||||
var chunk = entry.getValue();
|
||||
var size = chunk.readableBytes();
|
||||
var sizeSectors = Math.ceilDiv(size, 4096);
|
||||
if (sizeSectors > 255) {
|
||||
throw new IllegalArgumentException("MCC not supported.");
|
||||
}
|
||||
var indexEntry = Integer.reverseBytes(beginSector) << 1 | (sizeSectors & 0xFF);
|
||||
// sect = int[1024]
|
||||
sect.setInt((x + 32 * z) * 4, indexEntry);
|
||||
chunk.retain(); // retain a local reference
|
||||
compositeByteBuf.addComponent(true, chunk);
|
||||
beginSector += sizeSectors;
|
||||
}
|
||||
return compositeByteBuf;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
timestamp.release();
|
||||
regionData.values().forEach(ByteBuf::release);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package io.ib67.sfcraft.bundler.region;
|
||||
|
||||
import io.netty.buffer.*;
|
||||
import it.unimi.dsi.fastutil.longs.LongArrayList;
|
||||
import it.unimi.dsi.fastutil.longs.LongList;
|
||||
import net.jpountz.lz4.LZ4BlockInputStream;
|
||||
|
||||
import java.io.*;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.zip.DeflaterInputStream;
|
||||
import java.util.zip.GZIPInputStream;
|
||||
|
||||
public class RegionFileReassembler implements Closeable {
|
||||
protected final ByteBuf source;
|
||||
protected final byte[] sectorBuffer = new byte[4096];
|
||||
|
||||
public RegionFileReassembler(Path file) throws IOException {
|
||||
if (Files.size(file) < 8192) throw new IOException("The region file " + file + " has been truncated");
|
||||
source = Unpooled.wrappedBuffer(Files.readAllBytes(file));
|
||||
}
|
||||
|
||||
public RegionFileReassembler(ByteBuf source) {
|
||||
this.source = source;
|
||||
source.retain();
|
||||
}
|
||||
|
||||
// the lower 32bit is the entry content
|
||||
// upper 32bit is the chunkX <<< 6 | chunkZ (12 bits in total)
|
||||
public LongList readAllValidChunks() {
|
||||
var result = new LongArrayList();
|
||||
source.readerIndex(0);
|
||||
for (int i = 0; i < 1024; i++) {
|
||||
var entry = source.readInt();
|
||||
if (entry == 0) continue;
|
||||
var x = i % 32;
|
||||
var z = Math.max(0, (i - x) / 32);
|
||||
result.add(((long) ((x << 6) | z) << 32) | entry);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return kind of compression.
|
||||
*/
|
||||
public int readChunk(int entry, ByteBuf buf) {
|
||||
var offset = readOffset(entry);
|
||||
source.readerIndex(offset * 4096);
|
||||
// sector info
|
||||
var sectorLen = source.readInt();
|
||||
var compressType = source.readByte();
|
||||
if (compressType < 0) {
|
||||
// mcc todo
|
||||
}
|
||||
buf.writeBytes(source, sectorLen-1);
|
||||
return compressType;
|
||||
}
|
||||
|
||||
public ByteBuf writeReassembled(ByteBufAllocator alloc, RegionFile.CompressType newCompressType) throws IOException {
|
||||
var chunks = readAllValidChunks();
|
||||
var compressed = alloc.buffer();
|
||||
var uncompressed = alloc.buffer();
|
||||
try (var regionFile = new RegionFile(alloc)) {
|
||||
// read times
|
||||
source.readerIndex(4096);
|
||||
source.readBytes(sectorBuffer);
|
||||
regionFile.timestamp.clear();
|
||||
regionFile.timestamp.writeBytes(sectorBuffer);
|
||||
for (int i = 0; i < chunks.size(); i++) {
|
||||
compressed.clear();
|
||||
uncompressed.clear();
|
||||
var key = chunks.getLong(i);
|
||||
var chunkXZ = readChunkXZ(key);
|
||||
var entry = readEntry(key);
|
||||
short x = (short) (chunkXZ & 0x3F);
|
||||
short z = (short) ((chunkXZ >>> 6) & 0x3F);
|
||||
var compressType = readChunk(entry, compressed);
|
||||
try (var decompressor = getDecompressorByType(compressType, compressed)) {
|
||||
while (uncompressed.writeBytes(decompressor, 4096) != -1) ;
|
||||
regionFile.writeChunk(x, z, newCompressType, uncompressed);
|
||||
}
|
||||
}
|
||||
return regionFile.writeOutput();
|
||||
} finally {
|
||||
uncompressed.release();
|
||||
compressed.release();
|
||||
}
|
||||
}
|
||||
|
||||
private InputStream getDecompressorByType(int compressionType, ByteBuf in) throws IOException {
|
||||
return switch (compressionType) {
|
||||
case 1 -> new GZIPInputStream(new ByteBufInputStream(in));
|
||||
case 2 -> new DeflaterInputStream(new ByteBufInputStream(in));
|
||||
case 3 -> new ByteBufInputStream(in);
|
||||
case 4 -> new LZ4BlockInputStream(new ByteBufInputStream(in));
|
||||
// user defined compression algorithm, no going to support.
|
||||
// (or mcc)
|
||||
default -> throw new IOException("Unsupported compression type. " + compressionType);
|
||||
};
|
||||
}
|
||||
|
||||
public static short readChunkXZ(long packedEntry) {
|
||||
packedEntry = packedEntry >> 32;
|
||||
return (short) (packedEntry & 0xFFF);
|
||||
}
|
||||
|
||||
public static int readEntry(long packedEntry) {
|
||||
return (int) (packedEntry & 0xFFFFFFFFL);
|
||||
}
|
||||
|
||||
public static int readOffset(int headerEntry) {
|
||||
return (headerEntry >> 8);
|
||||
}
|
||||
|
||||
public static int readSize(int headerSize) {
|
||||
return headerSize & 0xFF;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
source.release();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package io.ib67.bundler;
|
||||
|
||||
import io.ib67.sfcraft.bundler.BundleWriter;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Path;
|
||||
import java.util.List;
|
||||
|
||||
public class TestBundleWriter {
|
||||
@Test
|
||||
public void testWriteBundle() throws IOException {
|
||||
var file = Path.of("/home/icybear/IdeaProjects/justbackup/bundler/src/test/resources/region/r.9.-5.mca");
|
||||
BundleWriter.createBundle(Path.of("/tmp/test.out"),
|
||||
List.of(file),
|
||||
c -> c.allowGunzip(true).relativeRoot(file.getParent()));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package io.ib67.bundler;
|
||||
|
||||
import io.ib67.sfcraft.bundler.region.RegionFile;
|
||||
import io.ib67.sfcraft.bundler.region.RegionFileReassembler;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import io.netty.buffer.Unpooled;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
public class TestRegionReassembler {
|
||||
@Test
|
||||
public void testReassembler() throws IOException {
|
||||
byte[] bytes;
|
||||
try(var in = TestRegionReassembler.class.getClassLoader().getResourceAsStream("region/r.9.-5.mca")){
|
||||
bytes = in.readAllBytes();
|
||||
}
|
||||
try(var reasm = new RegionFileReassembler(Unpooled.wrappedBuffer(bytes))) {
|
||||
var result = reasm.writeReassembled(ByteBufAllocator.DEFAULT, RegionFile.CompressType.NONE);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user