Switched parser logic to ByteBuffer from LittleEndianDataInputStream for more direct and performant access to the originally read replay bytes and easier reusability during parsing

Eliminated double reading of stream and multiple smaller tokenContent stream creation for tokenization and parsing by not preparing separate tokenContent byte chucks and reusing the originally created ByteBuffer
Replaced tokenContent and tokenSize with token limit, which stores the read position limit in the ByteBuffer for the corresponding token
Added buffer limiting to parsing to preserve robustness coming from tokenization
Improved performance of readString methods by replacing the ByteArrayOutputStream and double accessing the ByteBuffer for string ending search and extracting string bytes
Implemented getUnsignedByte util for ByteBuffer

Author: Rogerstalker

data/src/main/java/com/faforever/commons/replay/ReplayDataParser.java

@@ -4,31 +4,31 @@
 import com.faforever.commons.replay.body.ReplayBodyParser;
 import com.faforever.commons.replay.body.ReplayBodyToken;
 import com.faforever.commons.replay.body.ReplayBodyTokenizer;
+import com.faforever.commons.replay.shared.LoadUtils;
 import com.faforever.commons.replay.shared.LuaData;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.io.BaseEncoding;
-import com.google.common.io.LittleEndianDataInputStream;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.compress.compressors.CompressorException;
 import org.apache.commons.compress.compressors.CompressorInputStream;
 import org.apache.commons.compress.compressors.CompressorStreamFactory;
-import org.apache.commons.compress.utils.IOUtils;
+import org.apache.commons.io.IOUtils;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.*;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
-
 @SuppressWarnings("unused")
 @Slf4j
 public class ReplayDataParser {
@@ -59,7 +59,7 @@ public class ReplayDataParser {
   @Getter
   private final List<ModeratorEvent> moderatorEvents = new ArrayList<>();
   @Getter
-  private final Map<Integer, Map<Integer, AtomicInteger>> commandsPerMinuteByPlayer = new HashMap<>();
+  private Map<String, Integer> playerIdsByName;
 
   private int ticks;
 
@@ -80,48 +80,49 @@ public ReplayDataParser(Path path, ObjectMapper objectMapper) throws IOException
   }
 
   @VisibleForTesting
-  static String readString(LittleEndianDataInputStream dataStream) throws IOException {
-    ByteArrayOutputStream out = new ByteArrayOutputStream();
-    byte tempByte;
-    while ((tempByte = dataStream.readByte()) != 0) {
-      out.write(tempByte);
+  static String readString(ByteBuffer buffer) {
+    final int offset = buffer.position();
+    while (buffer.get() != 0) {
     }
-    return out.toString(StandardCharsets.UTF_8);
+    final int length = buffer.position() - 1 - offset;
+    byte[] stringBytes = new byte[length];
+    buffer.get(offset, stringBytes);
+    return new String(stringBytes, StandardCharsets.UTF_8);
   }
 
-  private Object parseLua(LittleEndianDataInputStream dataStream) throws IOException {
-    int type = dataStream.readUnsignedByte();
+  private Object parseLua(ByteBuffer buffer) {
+    int type = LoadUtils.getUnsignedByte(buffer);
     switch (type) {
       case LUA_NUMBER:
-        return dataStream.readFloat();
+        return buffer.getFloat();
       case LUA_STRING:
-        return readString(dataStream);
+        return readString(buffer);
       case LUA_NIL:
-        dataStream.skipBytes(1);
+        buffer.get();
         return null;
       case LUA_BOOL: // bool
-        return dataStream.readUnsignedByte() == 0;
+        return LoadUtils.getUnsignedByte(buffer) == 0;
       case LUA_TABLE_START: // lua
         Map<String, Object> result = new HashMap<>();
-        while (peek(dataStream) != LUA_TABLE_END) {
-          Object key = parseLua(dataStream);
+        while (peek(buffer) != LUA_TABLE_END) {
+          Object key = parseLua(buffer);
           if (key instanceof Number) {
             key = ((Number) key).intValue();
           }
-          result.put(String.valueOf(key), parseLua(dataStream));
-          dataStream.mark(1);
+          result.put(String.valueOf(key), parseLua(buffer));
+          buffer.mark();
         }
-        dataStream.skipBytes(1);
+        buffer.get();
         return result;
       default:
         throw new IllegalStateException("Unexpected data type: " + type);
     }
   }
 
-  private int peek(LittleEndianDataInputStream dataStream) throws IOException {
-    dataStream.mark(1);
-    int next = dataStream.readUnsignedByte();
-    dataStream.reset();
+  private int peek(ByteBuffer buffer) {
+    buffer.mark();
+    int next = LoadUtils.getUnsignedByte(buffer);
+    buffer.reset();
     return next;
   }
 
@@ -164,51 +165,51 @@ private byte[] decompress(byte[] data, @NotNull ReplayMetadata metadata) throws
   }
 
   @SuppressWarnings("unchecked")
-  private void parseHeader(LittleEndianDataInputStream dataStream) throws IOException {
-    replayPatchFieldId = readString(dataStream);
-    String arg13 = readString(dataStream); // always \r\n
+  private void parseHeader(ByteBuffer buffer) {
+    replayPatchFieldId = readString(buffer);
+    String arg13 = readString(buffer); // always \r\n
 
-    String[] split = readString(dataStream).split("\\r\\n");
+    String[] split = readString(buffer).split("\\r\\n");
     String replayVersionId = split[0];
     map = split[1];
-    String arg23 = readString((dataStream)); // always \r\n and some unknown character
+    String arg23 = readString((buffer)); // always \r\n and some unknown character
 
-    int sizeModsInBytes = dataStream.readInt();
-    mods = (Map<String, Map<String, ?>>) parseLua(dataStream);
+    int sizeModsInBytes = buffer.getInt();
+    mods = (Map<String, Map<String, ?>>) parseLua(buffer);
 
-    int sizeGameOptionsInBytes = dataStream.readInt();
-    this.gameOptions = ((Map<String, Object>) parseLua(dataStream)).entrySet().stream()
+    int sizeGameOptionsInBytes = buffer.getInt();
+    this.gameOptions = ((Map<String, Object>) parseLua(buffer)).entrySet().stream()
       .filter(entry -> "Options".equals(entry.getKey()))
       .flatMap(entry -> ((Map<String, Object>) entry.getValue()).entrySet().stream())
       .map(entry -> new GameOption(entry.getKey(), entry.getValue()))
       .collect(Collectors.toList());
 
-    int numberOfSources = dataStream.readUnsignedByte();
+    int numberOfSources = LoadUtils.getUnsignedByte(buffer);
 
-    Map<String, Object> playerIdsByName = new HashMap<>();
+    playerIdsByName = new HashMap<>();
     for (int i = 0; i < numberOfSources; i++) {
-      String playerName = readString(dataStream);
-      int playerId = dataStream.readInt();
+      String playerName = readString(buffer);
+      int playerId = buffer.getInt();
       playerIdsByName.put(playerName, playerId);
     }
 
-    boolean cheatsEnabled = dataStream.readUnsignedByte() > 0;
+    boolean cheatsEnabled = LoadUtils.getUnsignedByte(buffer) > 0;
 
-    int numberOfArmies = dataStream.readUnsignedByte();
+    int numberOfArmies = LoadUtils.getUnsignedByte(buffer);
     for (int i = 0; i < numberOfArmies; i++) {
-      int sizePlayerDataInBytes = dataStream.readInt();
-      Map<String, Object> playerData = (Map<String, Object>) parseLua(dataStream);
-      int playerSource = dataStream.readUnsignedByte();
+      int sizePlayerDataInBytes = buffer.getInt();
+      Map<String, Object> playerData = (Map<String, Object>) parseLua(buffer);
+      int playerSource = LoadUtils.getUnsignedByte(buffer);
 
       armies.put(playerSource, playerData);
       playerData.put("commands", new ArrayList<>());
 
       if (playerSource != 255) {
-        dataStream.skipBytes(1);
+        buffer.get();
       }
     }
 
-    randomSeed = dataStream.readInt();
+    randomSeed = buffer.getInt();
   }
 
   private void interpretEvents(List<Event> events) {
@@ -288,19 +289,13 @@ private void interpretEvents(List<Event> events) {
         case Event.IssueCommand(
           Event.CommandUnits commandUnits, Event.CommandData commandData
         ) -> {
-          commandsPerMinuteByPlayer
-            .computeIfAbsent(player, p -> new HashMap<>())
-            .computeIfAbsent(ticks, t -> new AtomicInteger())
-            .incrementAndGet();
+
         }
 
         case Event.IssueFactoryCommand(
           Event.CommandUnits commandUnits, Event.CommandData commandData
         ) -> {
-          commandsPerMinuteByPlayer
-            .computeIfAbsent(player, p -> new HashMap<>())
-            .computeIfAbsent(ticks, t -> new AtomicInteger())
-            .incrementAndGet();
+
         }
 
         case Event.IncreaseCommandCount(int commandId, int delta) -> {
@@ -364,10 +359,13 @@ private void interpretEvents(List<Event> events) {
   }
 
   private void parseGiveResourcesToPlayer(LuaData.Table lua) {
-    if (lua.value().containsKey("Msg") && lua.value().containsKey("From") && lua.value().containsKey("Sender")) {
+    LuaData msg;
+    LuaData from;
+    LuaData sender;
+    if ((msg = lua.value().get("Msg")) != null && (from = lua.value().get("From")) != null && (sender = lua.value().get("Sender")) != null) {
 
       // TODO: use the command source (player value) instead of the values from the callback. The values from the callback can be manipulated
-      if (!(lua.value().get("From") instanceof LuaData.Number(float luaFromArmy))) {
+      if (!(from instanceof LuaData.Number(float luaFromArmy))) {
         return;
       }
 
@@ -376,11 +374,11 @@ private void parseGiveResourcesToPlayer(LuaData.Table lua) {
         return;
       }
 
-      if (!(lua.value().get("Msg") instanceof LuaData.Table(Map<String, LuaData> luaMsg))) {
+      if (!(msg instanceof LuaData.Table(Map<String, LuaData> luaMsg))) {
         return;
       }
 
-      if (!(lua.value().get("Sender") instanceof LuaData.String(String luaSender))) {
+      if (!(sender instanceof LuaData.String(String luaSender))) {
         return;
       }
 
@@ -442,12 +440,18 @@ private Duration tickToTime(int tick) {
   }
 
   private void parse() throws IOException, CompressorException {
-    readReplayData(path);
-    try (LittleEndianDataInputStream dataStream = new LittleEndianDataInputStream(new ByteArrayInputStream(data))) {
-      parseHeader(dataStream);
-      tokens = ReplayBodyTokenizer.tokenize(dataStream);
-    }
-    events = ReplayBodyParser.parseTokens(tokens);
-    interpretEvents(events);
+      readReplayData(path);
+
+      final ByteBuffer buffer = ByteBuffer.wrap(data);
+      buffer.order(ByteOrder.LITTLE_ENDIAN);
+
+      parseHeader(buffer);
+
+      var rewindPosition = buffer.position();
+      tokens = ReplayBodyTokenizer.tokenize(buffer);
+      buffer.position(rewindPosition);
+
+      events = ReplayBodyParser.parseTokens(tokens, buffer);
+      interpretEvents(events);
   }
 }

data/src/main/java/com/faforever/commons/replay/ReplayLoader.java

@@ -10,7 +10,6 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.io.BaseEncoding;
-import com.google.common.io.LittleEndianDataInputStream;
 import org.apache.commons.compress.compressors.CompressorException;
 import org.apache.commons.compress.compressors.CompressorInputStream;
 import org.apache.commons.compress.compressors.CompressorStreamFactory;
@@ -22,6 +21,8 @@
 import java.io.ByteArrayOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -32,29 +33,33 @@
 public class ReplayLoader {
 
   @Contract(pure = true)
-  private static ReplayHeader loadSCFAReplayHeader(LittleEndianDataInputStream stream) throws IOException {
-    return ReplayHeaderParser.parse(stream);
+  private static ReplayHeader loadSCFAReplayHeader(ByteBuffer buffer) {
+    return ReplayHeaderParser.parse(buffer);
   }
 
   @Contract(pure = true)
-  private static @NotNull List<RegisteredEvent> loadSCFAReplayBody(List<Source> sources, LittleEndianDataInputStream stream) throws IOException {
-    List<ReplayBodyToken> bodyTokens = ReplayBodyTokenizer.tokenize(stream);
-    List<Event> bodyEvents = ReplayBodyParser.parseTokens(bodyTokens);
+  private static @NotNull List<RegisteredEvent> loadSCFAReplayBody(List<Source> sources, ByteBuffer buffer) {
+    var rewindPosition = buffer.position();
+    List<ReplayBodyToken> bodyTokens = ReplayBodyTokenizer.tokenize(buffer);
+    buffer.position(rewindPosition);
+
+    List<Event> bodyEvents = ReplayBodyParser.parseTokens(bodyTokens, buffer);
     return ReplaySemantics.registerEvents(sources, bodyEvents);
   }
 
   @Contract(pure = true)
   private static ReplayContainer loadSCFAReplayFromMemory(ReplayMetadata metadata, byte[] scfaReplayBytes) throws IOException {
-    try (LittleEndianDataInputStream stream = new LittleEndianDataInputStream((new ByteArrayInputStream(scfaReplayBytes)))) {
-      ReplayHeader replayHeader = loadSCFAReplayHeader(stream);
-      List<RegisteredEvent> replayBody = loadSCFAReplayBody(replayHeader.sources(), stream);
+    final ByteBuffer buffer = ByteBuffer.wrap(scfaReplayBytes);
+    buffer.order(ByteOrder.LITTLE_ENDIAN);
 
-      if (stream.available() > 0) {
-        throw new EOFException();
-      }
+    ReplayHeader replayHeader = loadSCFAReplayHeader(buffer);
+    List<RegisteredEvent> replayBody = loadSCFAReplayBody(replayHeader.sources(), buffer);
 
-      return new ReplayContainer(metadata, replayHeader, replayBody);
+    if (buffer.position() != buffer.limit()) {
+      throw new EOFException();
     }
+
+    return new ReplayContainer(metadata, replayHeader, replayBody);
   }
 
   public static ReplayContainer loadSCFAReplayFromDisk(Path scfaReplayFile) throws IOException, IllegalArgumentException {