Custom CBOR codec (for metadata)

For default metadata mime-type, using Jackson CBOR codec, produces more allocations (one buffer allocation per value) which are unnecessary. This was the major driving factor to invest in a _simple_ codec that can reduce allocations and be more suitable for ReactiveSocket design (using thread-local pooled buffers).

Although, the codec is intended to be used only for a map type, but it is rather generic to be used for a majority of CBOR data types. If this proves useful, we can extend this for the remaining data types.

Author: NiteshKant

build.gradle

@@ -23,6 +23,12 @@ subprojects {
         testCompile 'org.mockito:mockito-core:1.10.19'
         testRuntime 'org.slf4j:slf4j-simple:1.7.12'
     }
+
+    test {
+        testLogging {
+            showStandardStreams = true
+        }
+    }
 }
 
 // support for snapshot/final releases via versioned branch names like 1.x

reactivesocket-mime-types/src/main/java/io/reactivesocket/mimetypes/KVMetadata.java

@@ -1,8 +1,11 @@
 package io.reactivesocket.mimetypes;
 
+import org.agrona.MutableDirectBuffer;
+
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.Map;
+import java.util.function.Function;
 
 /**
  * A representation of ReactiveSocket metadata as a key-value pair.
@@ -18,7 +21,17 @@ public interface KVMetadata extends Map<String, ByteBuffer> {
      * @param valueEncoding Encoding for the value.
      *
      * @return Value as a string with the passed {@code valueEncoding}
+     * @throws NullPointerException If the key does not exist.
      */
     String getAsString(String key, Charset valueEncoding);
 
+    /**
+     * Creates a new copy of this metadata.
+     *
+     * @param newBufferFactory A factory to create new buffer instances to copy, if required. The argument to the
+     * function is the capacity of the new buffer.
+     *
+     * @return New copy of this metadata.
+     */
+    KVMetadata duplicate(Function<Integer, MutableDirectBuffer> newBufferFactory);
 }

reactivesocket-mime-types/src/main/java/io/reactivesocket/mimetypes/MimeType.java

@@ -42,14 +42,15 @@ default <T> T decodeMetadata(Frame toDecode, Class<T> clazz) {
     /**
      * Decodes the passed buffer to the specified {@code clazz}.
      *
+     * @param <T> Type of the class to which the buffer will be decoded.
+     *
      * @param toDecode buffer to be decoded.
      * @param clazz Class to which the buffer will be decoded.
-     *
-     * @param <T> Type of the class to which the buffer will be decoded.
+     * @param offset Offset in the buffer.
      *
      * @return Instance of the class post decode.
      */
-    <T> T decodeMetadata(DirectBuffer toDecode, Class<T> clazz);
+    <T> T decodeMetadata(DirectBuffer toDecode, Class<T> clazz, int offset);
 
     /**
      * Encodes passed metadata to a buffer.
@@ -76,12 +77,12 @@ default <T> T decodeMetadata(Frame toDecode, Class<T> clazz) {
     /**
      * Encodes passed metadata to the passed buffer.
      *
+     * @param <T> Type of the object to encode.
      * @param buffer Encodes the metadata to this buffer.
      * @param toEncode Metadata to encode.
-     *
-     * @param <T> Type of the object to encode.
+     * @param offset Offset in the buffer to start writing.
      */
-    <T> void encodeMetadataTo(MutableDirectBuffer buffer, T toEncode);
+    <T> void encodeMetadataTo(MutableDirectBuffer buffer, T toEncode, int offset);
 
     /**
      * Encodes passed metadata to the passed buffer.
@@ -122,14 +123,15 @@ default <T> T decodeData(Frame toDecode, Class<T> clazz) {
     /**
      * Decodes the passed buffer to the specified {@code clazz}.
      *
+     * @param <T> Type of the class to which the buffer will be decoded.
+     *
      * @param toDecode buffer to be decoded.
      * @param clazz Class to which the buffer will be decoded.
-     *
-     * @param <T> Type of the class to which the buffer will be decoded.
+     * @param offset Offset in the buffer.
      *
      * @return Instance of the class post decode.
      */
-    <T> T decodeData(DirectBuffer toDecode, Class<T> clazz);
+    <T> T decodeData(DirectBuffer toDecode, Class<T> clazz, int offset);
 
     /**
      * Encodes passed data to a buffer.
@@ -156,12 +158,12 @@ default <T> T decodeData(Frame toDecode, Class<T> clazz) {
     /**
      * Encodes passed data to the passed buffer.
      *
+     * @param <T> Type of the object to encode.
      * @param buffer Encodes the data to this buffer.
      * @param toEncode Data to encode.
-     *
-     * @param <T> Type of the object to encode.
+     * @param offset Offset in the buffer to start writing.
      */
-    <T> void encodeDataTo(MutableDirectBuffer buffer, T toEncode);
+    <T> void encodeDataTo(MutableDirectBuffer buffer, T toEncode, int offset);
 
     /**
      * Encodes passed data to the passed buffer.

reactivesocket-mime-types/src/main/java/io/reactivesocket/mimetypes/MimeTypeFactory.java

@@ -2,8 +2,8 @@
 
 import io.reactivesocket.ConnectionSetupPayload;
 import io.reactivesocket.mimetypes.internal.Codec;
-import io.reactivesocket.mimetypes.internal.ReactiveSocketDefaultMetadataCodec;
 import io.reactivesocket.mimetypes.internal.cbor.CborCodec;
+import io.reactivesocket.mimetypes.internal.cbor.ReactiveSocketDefaultMetadataCodec;
 import io.reactivesocket.mimetypes.internal.json.JsonCodec;
 import org.agrona.DirectBuffer;
 import org.agrona.MutableDirectBuffer;
@@ -122,8 +122,8 @@ public <T> T decodeMetadata(ByteBuffer toDecode, Class<T> clazz) {
         }
 
         @Override
-        public <T> T decodeMetadata(DirectBuffer toDecode, Class<T> clazz) {
-            return metaCodec.decode(toDecode, clazz);
+        public <T> T decodeMetadata(DirectBuffer toDecode, Class<T> clazz, int offset) {
+            return metaCodec.decode(toDecode, offset, clazz);
         }
 
         @Override
@@ -137,8 +137,8 @@ public <T> DirectBuffer encodeMetadataDirect(T toEncode) {
         }
 
         @Override
-        public <T> void encodeMetadataTo(MutableDirectBuffer buffer, T toEncode) {
-            metaCodec.encodeTo(buffer, toEncode);
+        public <T> void encodeMetadataTo(MutableDirectBuffer buffer, T toEncode, int offset) {
+            metaCodec.encodeTo(buffer, toEncode, offset);
         }
 
         @Override
@@ -152,8 +152,8 @@ public <T> T decodeData(ByteBuffer toDecode, Class<T> clazz) {
         }
 
         @Override
-        public <T> T decodeData(DirectBuffer toDecode, Class<T> clazz) {
-            return dataCodec.decode(toDecode, clazz);
+        public <T> T decodeData(DirectBuffer toDecode, Class<T> clazz, int offset) {
+            return dataCodec.decode(toDecode, offset, clazz);
         }
 
         @Override
@@ -167,8 +167,8 @@ public <T> DirectBuffer encodeDataDirect(T toEncode) {
         }
 
         @Override
-        public <T> void encodeDataTo(MutableDirectBuffer buffer, T toEncode) {
-            dataCodec.encodeTo(buffer, toEncode);
+        public <T> void encodeDataTo(MutableDirectBuffer buffer, T toEncode, int offset) {
+            dataCodec.encodeTo(buffer, toEncode, offset);
         }
 
         @Override

reactivesocket-mime-types/src/main/java/io/reactivesocket/mimetypes/internal/AbstractJacksonCodec.java

@@ -64,9 +64,9 @@ public <T> T decode(ByteBuffer buffer, Class<T> tClass) {
     }
 
     @Override
-    public <T> T decode(DirectBuffer buffer, Class<T> tClass) {
+    public <T> T decode(DirectBuffer buffer, int offset, Class<T> tClass) {
         DirectBufferInputStream stream = directInWrappers.get();
-        stream.wrap(buffer);
+        stream.wrap(buffer, offset, buffer.capacity());
         return _decode(stream, tClass);
     }
 
@@ -88,9 +88,9 @@ public <T> void encodeTo(ByteBuffer buffer, T toEncode) {
     }
 
     @Override
-    public <T> void encodeTo(MutableDirectBuffer buffer, T toEncode) {
+    public <T> void encodeTo(MutableDirectBuffer buffer, T toEncode, int offset) {
         MutableDirectBufferOutputStream stream = directOutWrappers.get();
-        stream.wrap(buffer);
+        stream.wrap(buffer, offset, buffer.capacity());
         _encodeTo(stream, toEncode);
     }
 

reactivesocket-mime-types/src/main/java/io/reactivesocket/mimetypes/internal/Codec.java

@@ -9,13 +9,13 @@ public interface Codec {
 
     <T> T decode(ByteBuffer buffer, Class<T> tClass);
 
-    <T> T decode(DirectBuffer buffer, Class<T> tClass);
+    <T> T decode(DirectBuffer buffer, int offset, Class<T> tClass);
 
     <T> ByteBuffer encode(T toEncode);
 
     <T> DirectBuffer encodeDirect(T toEncode);
 
     <T> void encodeTo(ByteBuffer buffer, T toEncode);
 
-    <T> void encodeTo(MutableDirectBuffer buffer, T toEncode);
+    <T> void encodeTo(MutableDirectBuffer buffer, T toEncode, int offset);
 }

reactivesocket-mime-types/src/main/java/io/reactivesocket/mimetypes/internal/KVMetadataImpl.java

@@ -1,13 +1,15 @@
 package io.reactivesocket.mimetypes.internal;
 
 import io.reactivesocket.mimetypes.KVMetadata;
+import org.agrona.MutableDirectBuffer;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 
 public class KVMetadataImpl implements KVMetadata {
 
@@ -37,14 +39,20 @@ public String getAsString(String key, Charset valueEncoding) {
         ByteBuffer toReturn = get(key);
 
         if (null != toReturn) {
-            byte[] dst = new byte[toReturn.limit() - toReturn.position()];
+            byte[] dst = new byte[toReturn.remaining()];
             toReturn.get(dst);
             return new String(dst, valueEncoding);
         }
 
         return null;
     }
 
+    @Override
+    public KVMetadata duplicate(Function<Integer, MutableDirectBuffer> newBufferFactory) {
+        Map<String, ByteBuffer> copy = new HashMap<>(store);
+        return new KVMetadataImpl(copy);
+    }
+
     @Override
     public int size() {
         return store.size();

reactivesocket-mime-types/src/main/java/io/reactivesocket/mimetypes/internal/MalformedInputException.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.reactivesocket.mimetypes.internal;
+
+public class MalformedInputException extends RuntimeException {
+
+    private static final long serialVersionUID = 3130502874275862715L;
+
+    public MalformedInputException(String message) {
+        super(message);
+    }
+
+    public MalformedInputException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public MalformedInputException(Throwable cause) {
+        super(cause);
+    }
+
+    public MalformedInputException(String message, Throwable cause, boolean enableSuppression,
+                                   boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+}