Version upgrade initialized.

.github/workflows/build.yml

@@ -16,7 +16,7 @@ jobs:
 
     services:
       throttr:
-        image: ghcr.io/throttr/throttr:4.0.17-debug-${{ matrix.size }}
+        image: ghcr.io/throttr/throttr:5.0.8-debug-${{ matrix.size }}-AMD64-metrics-enabled
         ports:
           - 9000:9000
 

pom.xml

@@ -1,12 +1,11 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
-                             https://maven.apache.org/xsd/maven-4.0.0.xsd">
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>cl.throttr</groupId>
     <artifactId>sdk</artifactId>
-    <version>4.1.1</version>
+    <version>5.0.0</version>
     <packaging>jar</packaging>
 
     <name>Throttr SDK for Java</name>
@@ -20,13 +19,6 @@
         </license>
     </licenses>
 
-    <properties>
-        <maven.compiler.source>21</maven.compiler.source>
-        <maven.compiler.target>21</maven.compiler.target>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <jacoco.agent.argLine />
-    </properties>
-
     <developers>
         <developer>
             <id>zen0x7</id>
@@ -49,12 +41,11 @@
         </repository>
     </distributionManagement>
 
-    <repositories>
-        <repository>
-            <id>central</id>
-            <url>https://repo.maven.apache.org/maven2</url>
-        </repository>
-    </repositories>
+    <properties>
+        <maven.compiler.source>21</maven.compiler.source>
+        <maven.compiler.target>21</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
 
     <dependencies>
         <dependency>
@@ -68,7 +59,6 @@
             <groupId>javax.annotation</groupId>
             <artifactId>javax.annotation-api</artifactId>
             <version>1.3.2</version>
-            <scope>compile</scope>
         </dependency>
 
         <dependency>
@@ -79,56 +69,40 @@
         </dependency>
 
         <dependency>
-            <groupId>org.jacoco</groupId>
-            <artifactId>org.jacoco.agent</artifactId>
-            <version>0.8.13</version>
-            <classifier>runtime</classifier>
-            <scope>test</scope>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+            <version>4.1.122.Final</version>
         </dependency>
     </dependencies>
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>3.5.3</version>
+            </plugin>
+
             <plugin>
                 <groupId>org.jacoco</groupId>
                 <artifactId>jacoco-maven-plugin</artifactId>
                 <version>0.8.13</version>
                 <executions>
                     <execution>
-                        <id>instrument</id>
+                        <id>prepare-agent</id>
                         <goals>
-                            <goal>instrument</goal>
+                            <goal>prepare-agent</goal>
                         </goals>
                     </execution>
-                    <execution>
-                        <id>restore-classes</id>
-                        <goals>
-                            <goal>restore-instrumented-classes</goal>
-                        </goals>
-                        <phase>prepare-package</phase>
-                    </execution>
                     <execution>
                         <id>report</id>
                         <phase>verify</phase>
                         <goals>
                             <goal>report</goal>
                         </goals>
-                        <configuration>
-                            <dataFile>${project.basedir}/jacoco.exec</dataFile>
-                        </configuration>
                     </execution>
                 </executions>
             </plugin>
-
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-surefire-plugin</artifactId>
-                <version>3.5.3</version>
-                <configuration>
-                    <reuseForks>false</reuseForks>
-                    <forkCount>1</forkCount>
-                </configuration>
-            </plugin>
         </plugins>
     </build>
-</project>
+</project>

src/main/java/cl/throttr/ByteBufAccumulator.java

@@ -0,0 +1,141 @@
+// Copyright (C) 2025 Ian Torres
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+package cl.throttr;
+
+import cl.throttr.enums.ValueSize;
+import cl.throttr.parsers.*;
+import cl.throttr.requests.PendingRequest;
+import cl.throttr.utils.Binary;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+import java.util.Map;
+import java.util.Queue;
+import java.util.function.Consumer;
+
+public class ByteBufAccumulator extends SimpleChannelInboundHandler<ByteBuf> {
+    private final Queue<PendingRequest> pending;
+    private final Map<String, Consumer<String>> subscriptions;
+    private final ValueSize size;
+    private ByteBuf buffer;
+
+    private final Map<Integer, ResponseParser> parsers;
+
+    public ByteBufAccumulator(Queue<PendingRequest> pending, Map<String, Consumer<String>> subscriptions, ValueSize size) {
+        this.pending = pending;
+        this.subscriptions = subscriptions;
+        this.size = size;
+        this.parsers = Map.ofEntries(
+                Map.entry(0x01, new StatusParser()),
+                Map.entry(0x02, new QueryParser(size)),
+                Map.entry(0x03, new StatusParser()),
+                Map.entry(0x04, new StatusParser()),
+                Map.entry(0x05, new StatusParser()),
+                Map.entry(0x06, new GetParser(size)),
+                Map.entry(0x07, new ListParser(size)),
+                Map.entry(0x08, new InfoParser()),
+                Map.entry(0x09, new StatParser()),
+                Map.entry(0x10, new StatsParser()),
+                Map.entry(0x11, new StatusParser()), // SUBSCRIBE
+                Map.entry(0x12, new StatusParser()), // UNSUBSCRIBE
+                Map.entry(0x13, new StatusParser()), // PUBLISH
+                Map.entry(0x14, new ConnectionsParser()),
+                Map.entry(0x15, new ConnectionParser()),
+                Map.entry(0x16, new ChannelsParser()),
+                Map.entry(0x17, new ChannelParser()),
+                Map.entry(0x18, new WhoamiParser())
+        );
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf incoming) {
+        if (buffer == null) {
+            buffer = incoming.alloc().buffer();
+        }
+        buffer.writeBytes(incoming);
+
+        while (true) {
+            if (buffer.readableBytes() < 1) return;
+
+            buffer.markReaderIndex();
+            int type = buffer.getUnsignedByte(buffer.readerIndex());
+
+            if (type == 0x19) {
+                if (!handleChannelMessage()) return;
+                continue;
+            }
+
+            if (!handlePendingRequest()) return;
+        }
+    }
+
+    private boolean handleChannelMessage() {
+        int readerIndex = buffer.readerIndex();
+
+        if (buffer.readableBytes() < 1 + size.getValue()) return false;
+
+        int channelSize = Byte.toUnsignedInt(buffer.getByte(readerIndex + 1));
+        int headerSize = 1 + 1 + size.getValue() + channelSize;
+
+        if (buffer.readableBytes() < headerSize) return false;
+
+        long payloadLength = Binary.read(buffer, readerIndex + 2, size);
+        if (buffer.readableBytes() < headerSize + payloadLength) return false;
+
+        byte[] channelBytes = new byte[channelSize];
+        buffer.getBytes(readerIndex + 2 + size.getValue(), channelBytes);
+        String channel = new String(channelBytes);
+
+        byte[] payloadBytes = new byte[(int) payloadLength];
+        buffer.getBytes(readerIndex + headerSize, payloadBytes);
+        String payload = new String(payloadBytes);
+
+        buffer.readerIndex(readerIndex + headerSize + (int) payloadLength);
+
+        Consumer<String> callback = subscriptions.get(channel);
+        if (callback != null) {
+            callback.accept(payload);
+        }
+
+        return true;
+    }
+
+    private boolean handlePendingRequest() {
+        PendingRequest pendingRequest = pending.peek();
+        if (pendingRequest == null) {
+            buffer.resetReaderIndex();
+            return false;
+        }
+
+        int expectedType = pendingRequest.type();
+        ResponseParser parser = parsers.get(expectedType);
+        ReadResult result = parser.tryParse(buffer);
+        if (result == null) {
+            buffer.resetReaderIndex();
+            return false;
+        }
+
+        buffer.skipBytes(result.consumed());
+        pending.poll().future().complete(result.value());
+        return true;
+    }
+
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) {
+        if (buffer != null) buffer.release();
+    }
+}