Merge pull request #963 from rabbitmq/refactor-codec-api

Add internal AMQP message format codec

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/Codec.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2025 Broadcom. All Rights Reserved.
+// Copyright (c) 2020-2026 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
@@ -14,7 +14,9 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream;
 
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import io.netty.buffer.ByteBuf;
+import java.io.IOException;
+import java.io.OutputStream;
 
 /**
  * Codec to encode and decode messages.
@@ -27,28 +29,16 @@ public interface Codec {
 
   EncodedMessage encode(Message message);
 
-  Message decode(byte[] data);
+  Message decode(ByteBuf buf, int length);
 
   MessageBuilder messageBuilder();
 
-  class EncodedMessage {
+  interface EncodedMessage {
 
-    private final int size;
-    private final byte[] data;
+    void writeTo(OutputStream outputStream) throws IOException;
 
-    @SuppressFBWarnings("EI_EXPOSE_REP2")
-    public EncodedMessage(int size, byte[] data) {
-      this.size = size;
-      this.data = data;
-    }
+    void writeTo(ByteBuf buf);
 
-    @SuppressFBWarnings("EI_EXPOSE_REP")
-    public byte[] getData() {
-      return data;
-    }
-
-    public int getSize() {
-      return size;
-    }
+    int getSize();
   }
 }

src/main/java/com/rabbitmq/stream/amqp/Symbol.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2025 Broadcom. All Rights Reserved.
+// Copyright (c) 2020-2026 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
@@ -14,9 +14,12 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream.amqp;
 
-public class Symbol {
+import java.util.concurrent.ConcurrentHashMap;
 
-  // FIXME maintain a cache
+public final class Symbol {
+
+  private static final int CACHE_MAX_SIZE = 2048;
+  private static final ConcurrentHashMap<String, Symbol> CACHE = new ConcurrentHashMap<>(64);
 
   private final String value;
 
@@ -25,11 +28,40 @@ private Symbol(String value) {
   }
 
   public static Symbol valueOf(String value) {
-    return new Symbol(value);
+    if (value == null) {
+      return null;
+    }
+    Symbol symbol = CACHE.get(value);
+    if (symbol == null) {
+      symbol = new Symbol(value);
+      if (CACHE.size() < CACHE_MAX_SIZE) {
+        Symbol existing = CACHE.putIfAbsent(value, symbol);
+        if (existing != null) {
+          symbol = existing;
+        }
+      }
+    }
+    return symbol;
   }
 
   @Override
   public String toString() {
     return value;
   }
+
+  @Override
+  public int hashCode() {
+    return value.hashCode();
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    return value.equals(((Symbol) o).value);
+  }
 }