naming change for clarity

Author: tmontgomery

src/main/java/io/reactivesocket/DuplexConnection.java

@@ -23,8 +23,8 @@
 public interface DuplexConnection {
     // TODO should we call this 'Connection'? 'SocketConnection'? 'ReactiveSocketConnection'?
 
-	Publisher<Message> getInput();
+	Publisher<Frame> getInput();
 
-	Publisher<Void> write(Publisher<Message> o);
+	Publisher<Void> write(Publisher<Frame> o);
 
 }

src/main/java/io/reactivesocket/Message.java renamed to src/main/java/io/reactivesocket/Frame.java

@@ -18,22 +18,22 @@
 import java.nio.ByteBuffer;
 
 /**
- * Represents a Message sent over a {@link DuplexConnection}.
+ * Represents a Frame sent over a {@link DuplexConnection}.
  * <p>
  * This provides encoding, decoding and field accessors.
  */
-public class Message {
-
+public class Frame
+{
     // TODO: make thread local to demonstrate idea
-    private static final ThreadLocal<FrameHandler> FRAME_HANDLER = ThreadLocal.withInitial(FrameHandler::new);
+    private static final ThreadLocal<FrameFlyweight> FRAME_HANDLER = ThreadLocal.withInitial(FrameFlyweight::new);
 
-    private Message() {
+    private Frame() {
     }
 
     // not final so we can reuse this object
     private ByteBuffer b;
     private int streamId;
-    private MessageType type;
+    private FrameType type;
     private String message;
 
     public ByteBuffer getBytes() {
@@ -54,7 +54,7 @@ public int getStreamId() {
         return streamId;
     }
 
-    public MessageType getMessageType() {
+    public FrameType getMessageType() {
         if (type == null) {
             decode();
         }
@@ -79,8 +79,8 @@ public void wrap(ByteBuffer b) {
      * @param b
      * @return
      */
-    public static Message from(ByteBuffer b) {
-        Message f = new Message();
+    public static Frame from(ByteBuffer b) {
+        Frame f = new Frame();
         f.b = b;
         return f;
     }
@@ -92,7 +92,7 @@ public static Message from(ByteBuffer b) {
      * @param type
      * @param message
      */
-    public void wrap(int streamId, MessageType type, String message) {
+    public void wrap(int streamId, FrameType type, String message) {
         this.streamId = streamId;
         this.type = type;
         this.message = message;
@@ -107,60 +107,35 @@ public void wrap(int streamId, MessageType type, String message) {
      * @param message
      * @return
      */
-    public static Message from(int streamId, MessageType type, String message) {
-        Message f = new Message();
+    public static Frame from(int streamId, FrameType type, String message) {
+        Frame f = new Frame();
         f.b = getBytes(streamId, type, message);
         f.streamId = streamId;
         f.type = type;
         f.message = message;
         return f;
     }
 
-    private static ByteBuffer getBytes(int messageId, MessageType type, String message) {
-        // TODO replace with binary
-        /**
-         * This is NOT how we want it for real. Just representing the idea for discussion.
-         */
-//        String s = "[" + type.getMessageId() + "]" + getIdString(messageId) + message;
-        // TODO stop allocating ... use flywheels
-//        return ByteBuffer.wrap(s.getBytes());
-
-        final FrameHandler frameHandler = FRAME_HANDLER.get();
+    private static ByteBuffer getBytes(int messageId, FrameType type, String message) {
+        final FrameFlyweight frameFlyweight = 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()));
+        final ByteBuffer buffer = ByteBuffer.allocate(FrameFlyweight.frameLength(message.length()));
 
-        frameHandler.encode(buffer, messageId, type, message.getBytes());
+        frameFlyweight.encode(buffer, messageId, type, message.getBytes());
         return buffer;
     }
 
-    private static String getIdString(int id) {
-        return "[" + id + "]|";
-    }
-
     private void decode() {
-        // TODO replace with binary
-        /**
-         * 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());
-
-        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
+        final FrameFlyweight frameFlyweight = FRAME_HANDLER.get();
+
+        frameFlyweight.decode(b);
+        this.type = frameFlyweight.messageType();
+        this.streamId = (int) frameFlyweight.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);
+        final byte[] data = new byte[frameFlyweight.dataLength()];
+        frameFlyweight.getDataBytes(data);
         this.message = new String(data);
     }
 
@@ -173,6 +148,6 @@ public String toString() {
                 e.printStackTrace();
             }
         }
-        return "Message => ID: " + streamId + " Type: " + type + " Data: " + message;
+        return "Frame => ID: " + streamId + " Type: " + type + " Data: " + message;
     }
 }

src/main/java/io/reactivesocket/FrameHandler.java renamed to src/main/java/io/reactivesocket/FrameFlyweight.java

@@ -26,7 +26,7 @@
  * Per connection frame handling.
  * Holds codecs and DirectBuffers for wrapping
  */
-public class FrameHandler
+public class FrameFlyweight
 {
     /**
      * Not the real frame layout, just an iteration on the ASCII version
@@ -46,7 +46,7 @@ public class FrameHandler
 
     // set by decode
     private long streamId;
-    private MessageType messageType;
+    private FrameType frameType;
     private byte version;
     private int dataLength;
 
@@ -55,7 +55,7 @@ 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)
+    public void encode(final ByteBuffer byteBuffer, final long streamId, final FrameType type, final byte[] data)
     {
         frameBuffer.wrap(byteBuffer);
         frameBuffer.putByte(VERSION_FIELD_OFFSET, CURRENT_VERSION);
@@ -74,7 +74,7 @@ public void decode(final ByteBuffer 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));
+        frameType = FrameType.from(frameBuffer.getInt(TYPE_FIELD_OFFSET, ByteOrder.BIG_ENDIAN));
         dataLength = frameBuffer.getInt(DATA_LENGTH_OFFSET, ByteOrder.BIG_ENDIAN);
     }
 
@@ -88,9 +88,9 @@ public byte version()
         return version;
     }
 
-    public MessageType messageType()
+    public FrameType messageType()
     {
-        return messageType;
+        return frameType;
     }
 
     public long streamId()

src/main/java/io/reactivesocket/MessageType.java renamed to src/main/java/io/reactivesocket/FrameType.java

@@ -16,10 +16,10 @@
 package io.reactivesocket;
 
 /**
- * Types of {@link Message} that can be sent.
+ * Types of {@link Frame} that can be sent.
  */
-public enum MessageType {
-
+public enum FrameType
+{
      SETUP(0x01),
      // Messages from Requestor
      REQUEST_RESPONSE(0x11),
@@ -33,35 +33,35 @@ public enum MessageType {
      COMPLETE(0x23),
      ERROR(0x24);
 
-    private static MessageType[] typesById;
+    private static FrameType[] typesById;
 
     /**
      * Index types by id for indexed lookup. 
      */
     static {
         int max = 0;
-        for (MessageType t : values()) {
+        for (FrameType t : values()) {
             if (t.id > max) {
                 max = t.id;
             }
         }
-        typesById = new MessageType[max + 1];
-        for (MessageType t : values()) {
+        typesById = new FrameType[max + 1];
+        for (FrameType t : values()) {
             typesById[t.id] = t;
         }
     }
 
     private final int id;
 
-    private MessageType(int id) {
+    FrameType(int id) {
         this.id = id;
     }
 
     public int getMessageId() {
         return id;
     }
 
-    public static MessageType from(int id) {
+    public static FrameType from(int id) {
         return typesById[id];
     }
 }

src/main/java/io/reactivesocket/ReactiveSocketClientProtocol.java

@@ -49,13 +49,13 @@ public class ReactiveSocketClientProtocol {
      * a Map each containing a single Subject for lookup+delivery is better
      * for doing demux ... currently chosen multicast+filter
      */
-    private final Observable<Message> multicastedInputStream;
+    private final Observable<Frame> multicastedInputStream;
     private int streamCount = 0;// 0 is reserved for setup, all normal messages are >= 1
 
     private ReactiveSocketClientProtocol(DuplexConnection connection) {
         this.connection = connection;
         // multicast input stream to any requestor (with backpressure support via RxJava .publish() operator)
-        ConnectableObservable<Message> published = toObservable(connection.getInput()).publish();
+        ConnectableObservable<Frame> published = toObservable(connection.getInput()).publish();
         multicastedInputStream = published;
         // start listening ... ignoring returned subscription as we don't control connection lifecycle here
         published.connect();
@@ -72,7 +72,7 @@ public static ReactiveSocketClientProtocol create(DuplexConnection connection) {
      * @return
      */
     public Publisher<String> requestResponse(String payload) {
-        return startStream(Message.from(nextStreamId(), MessageType.REQUEST_RESPONSE, payload));
+        return startStream(Frame.from(nextStreamId(), FrameType.REQUEST_RESPONSE, payload));
     }
 
     /**
@@ -83,7 +83,7 @@ public Publisher<String> requestResponse(String payload) {
      * @return
      */
     public Publisher<String> requestStream(String payload) {
-        return startStream(Message.from(nextStreamId(), MessageType.REQUEST_STREAM, payload));
+        return startStream(Frame.from(nextStreamId(), FrameType.REQUEST_STREAM, payload));
     }
 
     /**
@@ -97,7 +97,7 @@ public Publisher<String> requestStream(String payload) {
      * @return
      */
     public Publisher<Void> fireAndForget(String payload) {
-        return connection.write(toPublisher(just(Message.from(nextStreamId(), MessageType.FIRE_AND_FORGET, payload))));
+        return connection.write(toPublisher(just(Frame.from(nextStreamId(), FrameType.FIRE_AND_FORGET, payload))));
     }
 
     /**
@@ -108,14 +108,14 @@ public Publisher<Void> fireAndForget(String payload) {
      * @return
      */
     public Publisher<String> requestSubscription(String payload) {
-        return startStream(Message.from(nextStreamId(), MessageType.REQUEST_SUBSCRIPTION, payload));
+        return startStream(Frame.from(nextStreamId(), FrameType.REQUEST_SUBSCRIPTION, payload));
     }
 
-    private Publisher<String> startStream(Message requestMessage) {
+    private Publisher<String> startStream(Frame requestFrame) {
         return toPublisher(Observable.create(child -> {
 
             // TODO replace this with a UnicastSubject without the overhead of multicast support
-            PublishSubject<Observable<Message>> writer = PublishSubject.create();
+            PublishSubject<Observable<Frame>> writer = PublishSubject.create();
             Observable<Void> written = toObservable(connection.write(toPublisher(Observable.merge(writer))));
 
             child.setProducer(new Producer() {
@@ -128,43 +128,43 @@ public void request(long n) {
                         start();
                     }
                     // send REQUEST_N over network for streaming request types
-                    if (requestMessage.getMessageType() == MessageType.REQUEST_STREAM || requestMessage.getMessageType() == MessageType.REQUEST_SUBSCRIPTION) {
-                        writer.onNext(just(Message.from(requestMessage.getStreamId(), MessageType.REQUEST_N, String.valueOf(n))));
+                    if (requestFrame.getMessageType() == FrameType.REQUEST_STREAM || requestFrame.getMessageType() == FrameType.REQUEST_SUBSCRIPTION) {
+                        writer.onNext(just(Frame.from(requestFrame.getStreamId(), FrameType.REQUEST_N, String.valueOf(n))));
                     }
                 }
 
                 private void start() {
                     // wire up the response handler before emitting request
-                    Observable<Message> input = multicastedInputStream
-                            .filter(m -> m.getStreamId() == requestMessage.getStreamId());
+                    Observable<Frame> input = multicastedInputStream
+                            .filter(m -> m.getStreamId() == requestFrame.getStreamId());
 
                     // combine input and output so errors and unsubscription are composed, then subscribe
                     rx.Subscription subscription = Observable
-                            .merge(input, written.cast(Message.class))
-                            .takeUntil(m -> (m.getMessageType() == MessageType.COMPLETE
-                                    || m.getMessageType() == MessageType.ERROR))
+                            .merge(input, written.cast(Frame.class))
+                            .takeUntil(m -> (m.getMessageType() == FrameType.COMPLETE
+                                    || m.getMessageType() == FrameType.ERROR))
                             .flatMap(m -> {
                         // convert ERROR/COMPLETE messages into terminal events
-                        if (m.getMessageType() == MessageType.ERROR) {
+                        if (m.getMessageType() == FrameType.ERROR) {
                             return error(new Exception(m.getMessage()));
-                        } else if (m.getMessageType() == MessageType.COMPLETE) {
+                        } else if (m.getMessageType() == FrameType.COMPLETE) {
                             return empty();// unsubscribe handled in takeUntil above
-                        } else if (m.getMessageType() == MessageType.NEXT) {
+                        } else if (m.getMessageType() == FrameType.NEXT) {
                             return just(m.getMessage());
                         } else {
-                            return error(new Exception("Unexpected MessageType: " + m.getMessageType()));
+                            return error(new Exception("Unexpected FrameType: " + m.getMessageType()));
                         }
                     }).subscribe(Subscribers.from(child));// only propagate Observer methods, backpressure is via Producer above
 
                     // if the child unsubscribes, we need to send a CANCEL message if we're not terminated
                     child.add(Subscriptions.create(() -> {
-                        writer.onNext(just(Message.from(requestMessage.getStreamId(), MessageType.CANCEL, "")));
+                        writer.onNext(just(Frame.from(requestFrame.getStreamId(), FrameType.CANCEL, "")));
                         // after sending the CANCEL we then tear down this stream
                         subscription.unsubscribe();
                     }));
 
                     // send the request to start everything
-                    writer.onNext(just(requestMessage));
+                    writer.onNext(just(requestFrame));
                 }
 
             });