(Protocol): ProtocolWriter and ProtocolReader contract changes

- ProtocolWriter no longer implements Flushable
- BytesWriter instances can be reused for writing multiple requests.
- Internal optimizations in BytesWriter to minimize allocations
- More unit tests for BytesWriter (+2 squashed commit)
- ProtocolResponseReader can now read the data after data-enriched responses
- Add the new "SCOPE_DATA" scope which is entered after the values of a
data-enriched response is read and the ProtocolResponseReader is at the
begging of the attached data bytes.

Author: georgi-neykov-hub

protocol/src/main/java/com/pcloud/networking/protocol/BytesReader.java

@@ -17,10 +17,12 @@
 package com.pcloud.networking.protocol;
 
 import com.pcloud.utils.IOUtils;
+import okio.BufferedSink;
 import okio.BufferedSource;
 import okio.Okio;
 
 import java.io.IOException;
+import java.io.OutputStream;
 import java.net.ProtocolException;
 import java.util.ArrayDeque;
 import java.util.Arrays;
@@ -30,11 +32,15 @@
 import static com.pcloud.utils.IOUtils.closeQuietly;
 
 /**
- * Reads bytes from source
+ * A reader for pCloud's binary data protocol
  * <p>
- * An implementation of {@linkplain ProtocolResponseReader} which is able to read bytes from a {@linkplain BufferedSource}
+ * An implementation of {@linkplain ProtocolResponseReader} which can read data
+ * encoded in pCloud's binary protocol from a stream of bytes in pull-based fashion
+ * with minimal overhead.
  * <p>
- * Generally used to read the bytes of a network response into serialized data
+ *
+ * The implementation can be reused for reading multiple responses from a byte source,
+ * as long as they are read completely.
  *
  * @see ProtocolReader
  * @see ProtocolResponseReader
@@ -121,18 +127,17 @@ public long beginResponse() throws IOException {
             lastStringId = 0;
             return pullNumber(RESPONSE_LENGTH_BYTESIZE);
         }
-
-        throw new SerializationException("Trying to start reading a response, which has already been started.");
+        throw new IllegalStateException("An already started response or data after it has not been fully read.");
     }
 
     @Override
-    public long endResponse() throws IOException {
+    public boolean endResponse() throws IOException {
         if (currentScope != SCOPE_RESPONSE) {
             if (currentScope == SCOPE_NONE) {
-                throw new SerializationException("Trying to end a response " +
+                throw new IllegalStateException("Trying to end a response " +
                         "but none is being read, first call beginResponse()");
             } else {
-                throw new SerializationException("Trying to end a response, " +
+                throw new IllegalStateException("Trying to end a response, " +
                         "but current scope is " + scopeName(currentScope));
             }
         }
@@ -146,11 +151,12 @@ public long endResponse() throws IOException {
         popScope();
         stringCache = null;
 
-        if (dataLength == UNKNOWN_SIZE) {
-            dataLength = 0;
+        boolean dataAvailable = dataLength != UNKNOWN_SIZE;
+        if (dataAvailable) {
+            pushScope(SCOPE_DATA);
         }
 
-        return dataLength;
+        return dataAvailable;
     }
 
     @Override
@@ -274,9 +280,30 @@ public boolean hasNext() throws IOException {
 
     @Override
     public long dataContentLength() {
+        if (currentScope == SCOPE_NONE) {
+            throw new IllegalStateException("No response or data after it are being read.");
+        }
         return dataLength;
     }
 
+    @Override
+    public void readData(OutputStream outputStream) throws IOException {
+        readData(Okio.buffer(Okio.sink(outputStream)));
+    }
+
+    @Override
+    public void readData(BufferedSink sink) throws IOException {
+        if (currentScope != SCOPE_DATA) {
+            throw new IllegalStateException("Cannot read data," +
+                    " either the response is not read fully or no data is following.");
+        }
+
+        long length = dataLength;
+        sink.write(bufferedSource, length);
+        dataLength = UNKNOWN_SIZE;
+        popScope();
+    }
+
     @Override
     public ProtocolResponseReader newPeekingReader() {
         BytesReader reader = new PeekingByteReader();
@@ -297,10 +324,9 @@ public ProtocolResponseReader newPeekingReader() {
     @Override
     public void skipValue() throws IOException {
 
-        if (currentScope == SCOPE_NONE) {
+        if (currentScope == SCOPE_NONE || currentScope == SCOPE_DATA) {
             throw new IllegalStateException("Trying to skipValue, but currentScope is " +
-                                                    currentScope +
-                                                    ". You must call beginResponse() first.");
+                    scopeName(currentScope) + ".");
         }
 
         final int type = peekType();
@@ -509,5 +535,15 @@ private PeekingByteReader() {
         public ProtocolResponseReader newPeekingReader() {
             throw new IllegalStateException("Cannot call newPeekingReader(), this reader is already non-consuming.");
         }
+
+        @Override
+        public void readData(OutputStream outputStream) throws IOException {
+            throw new UnsupportedOperationException("Data cannot be peeked.");
+        }
+
+        @Override
+        public void readData(BufferedSink sink) throws IOException {
+            throw new UnsupportedOperationException("Data cannot be peeked.");
+        }
     }
 }

protocol/src/main/java/com/pcloud/networking/protocol/BytesWriter.java

@@ -50,8 +50,8 @@ public class BytesWriter implements ProtocolRequestWriter {
     private static final int MAX_METHOD_LENGTH = 127;
     private static final int BITWISE_SHIFT_SIX = 6;
 
-    private BufferedSink sink;
-    private Buffer paramsBuffer;
+    private final BufferedSink sink;
+    private final Buffer paramsBuffer;
     private DataSource dataSource;
 
     private String methodName;
@@ -63,7 +63,6 @@ public class BytesWriter implements ProtocolRequestWriter {
     /**
      * Create a {@linkplain BytesReader} instance
      * <p>
-     *
      * @param sink a {@linkplain BufferedSink} to house the bytes
      * @throws IllegalArgumentException on a null {@linkplain BufferedSink} argument
      */
@@ -106,50 +105,56 @@ public ProtocolRequestWriter endRequest() throws IOException {
         checkWriteValueFinished();
 
         if (methodName == null) {
-            throw new IllegalArgumentException("Cannot end request, writeMethodName() has not been called.");
+            throw new IllegalArgumentException("Cannot end request, " +
+                    "writeMethodName() has not been called.");
         }
 
-
-        Buffer metadataBuffer = new Buffer();
-
         int methodNameLength = (int) Utf8.size(methodName);
         if (methodNameLength > MAX_METHOD_LENGTH) {
-            throw new SerializationException("Method name cannot be larger than 127 bytes.");
+            throw new SerializationException(
+                    "Invalid method name '%s', " +
+                            "value cannot be larger than 127 bytes in UTF-8 encoding.", methodName);
         }
 
         final long dataSourceLength = dataSource != null ? dataSource.contentLength() : 0;
         if (dataSourceLength < 0L) {
-            throw new SerializationException("Unknown or invalid DataSource content length '" +
-                                                     dataSourceLength +
-                                                     "'.");
+            throw new SerializationException(
+                    "Unknown or invalid DataSource content length '%d'.", dataSourceLength);
         }
 
         boolean hasData = dataSourceLength > 0;
-        if (hasData) {
-            methodNameLength = methodNameLength | (1 << REQUEST_BINARY_DATA_FLAG_POSITION);
-        }
-        metadataBuffer.writeByte(methodNameLength);
 
-        if (hasData) {
-            metadataBuffer.writeLongLe(dataSourceLength);
-        }
-
-        metadataBuffer.writeUtf8(methodName);
-        metadataBuffer.writeByte(parameterCount);
-
-        final long requestSize = metadataBuffer.size() + paramsBuffer.size();
+        final long requestSize =
+                (hasData ? 8 : 0) + // Data length, if any (8 bytes)
+                        2 + // Method name length + data flag (1 byte) + Parameter count (1 byte)
+                        methodNameLength +
+                        +paramsBuffer.size();
         if (requestSize > REQUEST_SIZE_LIMIT_BYTES) {
             throw new SerializationException("The maximum allowed request size is 65535 bytes," +
                                                      " current is " + requestSize + " bytes.");
         }
 
-        sink.writeShortLe((int) requestSize);
-        sink.write(metadataBuffer, metadataBuffer.size());
+        if (hasData) {
+            methodNameLength = methodNameLength | (1 << REQUEST_BINARY_DATA_FLAG_POSITION);
+        }
+
+        sink.writeShortLe((int) requestSize); // Request size, 2 bytes
+        sink.writeByte(methodNameLength);
+        if (hasData) {
+            sink.writeLongLe(dataSourceLength); // Size of data after response, 8 bytes
+        }
+        sink.writeUtf8(methodName);
+        sink.writeByte(parameterCount);
         sink.write(this.paramsBuffer, paramsBuffer.size());
         if (hasData) {
             dataSource.writeTo(sink);
         }
 
+        sink.emit();
+        dataSource = null;
+        methodName = null;
+        requestStarted = false;
+        parameterCount = 0;
         return this;
     }
 
@@ -297,16 +302,6 @@ public void close() {
         dataSource = null;
     }
 
-    /**
-     * Flush the {@linkplain BufferedSink}
-     *
-     * @throws IOException on failed IO operations
-     */
-    @Override
-    public void flush() throws IOException {
-        sink.flush();
-    }
-
     private void checkRequestStarted() {
         if (!requestStarted) {
             throw new IllegalStateException("Call beginRequest() before calling this method.");

protocol/src/main/java/com/pcloud/networking/protocol/ProtocolWriter.java

@@ -25,7 +25,7 @@
  * @see BytesWriter
  * @see ValueWriter
  */
-public interface ProtocolWriter extends AutoCloseable, Closeable, Flushable {
+public interface ProtocolWriter extends AutoCloseable, Closeable {
 
     /**
      * Write a key for a data pair
@@ -111,11 +111,4 @@ public interface ProtocolWriter extends AutoCloseable, Closeable, Flushable {
      */
     @Override
     void close();
-
-    /**
-     *  Flush the data source
-     *  @throws IOException on failed IO operations
-     */
-    @Override
-    void flush() throws IOException;
 }