exploratory binary encoding.

Author: tmontgomery

src/main/java/io/reactivesocket/FrameHandler.java

@@ -0,0 +1,111 @@
+/**
+ * Copyright 2015 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;
+
+import uk.co.real_logic.agrona.BitUtil;
+import uk.co.real_logic.agrona.MutableDirectBuffer;
+import uk.co.real_logic.agrona.concurrent.UnsafeBuffer;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+/**
+ * Per connection frame handling.
+ * Holds codecs and DirectBuffers for wrapping
+ */
+public class FrameHandler
+{
+    /**
+     * Not the real frame layout, just an iteration on the ASCII version
+     */
+    private static final int VERSION_FIELD_OFFSET = 0;
+    private static final int STREAM_ID_FIELD_OFFSET = VERSION_FIELD_OFFSET + BitUtil.SIZE_OF_INT;
+    private static final int TYPE_FIELD_OFFSET = STREAM_ID_FIELD_OFFSET + BitUtil.SIZE_OF_LONG;
+    private static final int DATA_LENGTH_OFFSET = TYPE_FIELD_OFFSET + BitUtil.SIZE_OF_INT;
+    private static final int DATA_OFFSET = DATA_LENGTH_OFFSET + BitUtil.SIZE_OF_INT;
+
+    private static final byte CURRENT_VERSION = 0;
+
+    private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
+
+    // single threaded assumed
+    private final MutableDirectBuffer frameBuffer = new UnsafeBuffer(EMPTY_BUFFER);
+
+    // set by decode
+    private long streamId;
+    private MessageType messageType;
+    private byte version;
+    private int dataLength;
+
+    public static int frameLength(final int dataLength)
+    {
+        return DATA_OFFSET + dataLength;
+    }
+
+    public void encode(final ByteBuffer byteBuffer, final long streamId, final MessageType type, final byte[] data)
+    {
+        frameBuffer.wrap(byteBuffer);
+        frameBuffer.putByte(VERSION_FIELD_OFFSET, CURRENT_VERSION);
+        frameBuffer.putLong(STREAM_ID_FIELD_OFFSET, streamId, ByteOrder.BIG_ENDIAN);
+        frameBuffer.putInt(TYPE_FIELD_OFFSET, type.getMessageId(), ByteOrder.BIG_ENDIAN);
+        frameBuffer.putInt(DATA_LENGTH_OFFSET, data.length, ByteOrder.BIG_ENDIAN);
+        frameBuffer.putBytes(DATA_OFFSET, data);
+    }
+
+    /**
+     * populate streamId, type, dataLength, etc.
+     */
+    public void decode(final ByteBuffer byteBuffer)
+    {
+        frameBuffer.wrap(byteBuffer);
+
+        version = frameBuffer.getByte(VERSION_FIELD_OFFSET);
+        streamId = frameBuffer.getLong(STREAM_ID_FIELD_OFFSET, ByteOrder.BIG_ENDIAN);
+        messageType = MessageType.from(frameBuffer.getInt(TYPE_FIELD_OFFSET, ByteOrder.BIG_ENDIAN));
+        dataLength = frameBuffer.getInt(DATA_LENGTH_OFFSET, ByteOrder.BIG_ENDIAN);
+    }
+
+    public ByteBuffer buffer()
+    {
+        return frameBuffer.byteBuffer();
+    }
+
+    public byte version()
+    {
+        return version;
+    }
+
+    public MessageType messageType()
+    {
+        return messageType;
+    }
+
+    public long streamId()
+    {
+        return streamId;
+    }
+
+    public int dataLength()
+    {
+        return dataLength;
+    }
+
+    public void getDataBytes(final byte[] array)
+    {
+        frameBuffer.getBytes(DATA_OFFSET, array);
+    }
+
+}

src/main/java/io/reactivesocket/Message.java

@@ -24,6 +24,9 @@
  */
 public class Message {
 
+    // TODO: make thread local to demonstrate idea
+    private static final ThreadLocal<FrameHandler> FRAME_HANDLER = ThreadLocal.withInitial(FrameHandler::new);
+
     private Message() {
     }
 
@@ -118,9 +121,17 @@ private static ByteBuffer getBytes(int messageId, MessageType type, String messa
         /**
          * This is NOT how we want it for real. Just representing the idea for discussion.
          */
-        String s = "[" + type.getMessageId() + "]" + getIdString(messageId) + message;
+//        String s = "[" + type.getMessageId() + "]" + getIdString(messageId) + message;
         // TODO stop allocating ... use flywheels
-        return ByteBuffer.wrap(s.getBytes());
+//        return ByteBuffer.wrap(s.getBytes());
+
+        final FrameHandler frameHandler = FRAME_HANDLER.get();
+
+        // TODO: allocation side effect of how this works currently with the rest of the machinery.
+        final ByteBuffer buffer = ByteBuffer.allocate(FrameHandler.frameLength(message.length()));
+
+        frameHandler.encode(buffer, messageId, type, message.getBytes());
+        return buffer;
     }
 
     private static String getIdString(int id) {
@@ -132,14 +143,25 @@ private void decode() {
         /**
          * This is NOT how we want it for real. Just representing the idea for discussion.
          */
-        byte[] copy = new byte[b.limit()];
-        b.get(copy);
-        String data = new String(copy);
-        int separator = data.indexOf('|');
-        String prefix = data.substring(0, separator);
-        this.type = MessageType.from(Integer.parseInt(prefix.substring(1, data.indexOf(']'))));
-        this.streamId = Integer.parseInt(prefix.substring(prefix.lastIndexOf("[") + 1, prefix.length() - 1));
-        this.message = data.substring(separator + 1, data.length());
+//        byte[] copy = new byte[b.limit()];
+//        b.get(copy);
+//        String data = new String(copy);
+//        int separator = data.indexOf('|');
+//        String prefix = data.substring(0, separator);
+//        this.type = MessageType.from(Integer.parseInt(prefix.substring(1, data.indexOf(']'))));
+//        this.streamId = Integer.parseInt(prefix.substring(prefix.lastIndexOf("[") + 1, prefix.length() - 1));
+//        this.message = data.substring(separator + 1, data.length());
+
+        final FrameHandler frameHandler = FRAME_HANDLER.get();
+
+        frameHandler.decode(b);
+        this.type = frameHandler.messageType();
+        this.streamId = (int)frameHandler.streamId();  // TODO: temp cast to only touch as little as possible
+
+        // TODO: temp allocation to touch as little as possible
+        final byte[] data = new byte[frameHandler.dataLength()];
+        frameHandler.getDataBytes(data);
+        this.message = new String(data);
     }
 
     @Override

src/main/java/io/reactivesocket/MessageType.java

@@ -20,7 +20,7 @@
  */
 public enum MessageType {
 
-    SETUP(0x01),
+     SETUP(0x01),
      // Messages from Requestor
      REQUEST_RESPONSE(0x11),
      FIRE_AND_FORGET(0x12),