Resuable Payload (#209)

__Problem__

`PayloadImpl` caches `data` and `metadata` buffers. This means once those buffers are read, they can not be reused as the pointer of `ByteBuffer` has moved to the end of buffer.
This restricts the usage of `PayloadImpl` instance with something like:

```java
socket.requestResponse(new PayloadImpl("Hello")).retry(2);
```

For any retry in the above code, data read from the payload instance will be empty.

__Modification__

- store buffer `position()` on creation and reset them on every `get()`.
- Additional constructor to override this behavior and create a payload for single use.
- `PayloadImpl` defaults to reusable payload.

__Result__

Possible to retry a request without having a custom `Payload` implementation.

Author: NiteshKant

reactivesocket-core/src/main/java/io/reactivesocket/util/PayloadImpl.java

@@ -23,13 +23,23 @@
 import java.nio.charset.Charset;
 
 /**
- * An implementation of {@link Payload}
+ * An implementation of {@link Payload}. This implementation is <b>not</b> thread-safe, and hence any method can not be
+ * invoked concurrently.
+ *
+ * <h2>Reusability</h2>
+ *
+ * By default, an instance is reusable, i.e. everytime {@link #getData()} or {@link #getMetadata()} is invoked, the
+ * position of the returned buffer is reset to the start of data in the buffer. For creating an instance for single-use,
+ * {@link #PayloadImpl(ByteBuffer, ByteBuffer, boolean)} must be used.
  */
 public class PayloadImpl implements Payload {
 
     public static final Payload EMPTY = new PayloadImpl(Frame.NULL_BYTEBUFFER, Frame.NULL_BYTEBUFFER);
 
     private final ByteBuffer data;
+    private final int dataStartPosition;
+    private final int metadataStartPosition;
+    private final boolean reusable;
     private final ByteBuffer metadata;
 
     public PayloadImpl(String data) {
@@ -61,17 +71,39 @@ public PayloadImpl(ByteBuffer data) {
     }
 
     public PayloadImpl(ByteBuffer data, ByteBuffer metadata) {
+        this(data, metadata, true);
+    }
+
+    /**
+     * New instance where every invocation of {@link #getMetadata()} and {@link #getData()} will reset the position of
+     * the buffer to the position when it is created, if and only if, {@code reusable} is set to {@code true}.
+     *
+     * @param data Buffer for data.
+     * @param metadata Buffer for metadata.
+     * @param reusable {@code true} if the buffer position is to be reset on every invocation of {@link #getData()} and
+     * {@link #getMetadata()}.
+     */
+    public PayloadImpl(ByteBuffer data, ByteBuffer metadata, boolean reusable) {
         this.data = data;
+        this.reusable = reusable;
         this.metadata = null == metadata ? Frame.NULL_BYTEBUFFER : metadata;
+        dataStartPosition = reusable ? this.data.position() : 0;
+        metadataStartPosition = reusable ? this.metadata.position() : 0;
     }
 
     @Override
     public ByteBuffer getData() {
+        if (reusable) {
+            data.position(dataStartPosition);
+        }
         return data;
     }
 
     @Override
     public ByteBuffer getMetadata() {
+        if (reusable) {
+            metadata.position(metadataStartPosition);
+        }
         return metadata;
     }
 

reactivesocket-core/src/test/java/io/reactivesocket/util/PayloadImplTest.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2016 Netflix, Inc.
+ * <p>
+ *  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
+ *  <p>
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  <p>
+ *  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.util;
+
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+
+import static java.nio.ByteBuffer.wrap;
+import static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+
+public class PayloadImplTest {
+
+    public static final String DATA_VAL = "data";
+    public static final String METADATA_VAL = "Metadata";
+
+    @Test
+    public void testReuse() throws Exception {
+        PayloadImpl p = new PayloadImpl(DATA_VAL, METADATA_VAL);
+        assertDataAndMetadata(p);
+        assertDataAndMetadata(p);
+    }
+    @Test
+    public void testSingleUse() throws Exception {
+        PayloadImpl p = new PayloadImpl(wrap(DATA_VAL.getBytes()), wrap(METADATA_VAL.getBytes()), false);
+        assertDataAndMetadata(p);
+        assertThat("Unexpected data length", p.getData().remaining(), is(0));
+        assertThat("Unexpected metadata length", p.getMetadata().remaining(), is(0));
+    }
+
+    @Test
+    public void testReuseWithExternalMark() throws Exception {
+        PayloadImpl p = new PayloadImpl(DATA_VAL, METADATA_VAL);
+        assertDataAndMetadata(p);
+        p.getData().position(2).mark();
+        assertDataAndMetadata(p);
+    }
+
+    public void assertDataAndMetadata(PayloadImpl p) {
+        assertThat("Unexpected data.", readValue(p.getData()), equalTo(DATA_VAL));
+        assertThat("Unexpected metadata.", readValue(p.getMetadata()), equalTo(METADATA_VAL));
+    }
+
+    public String readValue(ByteBuffer buffer) {
+        byte[] bytes = new byte[buffer.remaining()];
+        buffer.get(bytes);
+        return new String(bytes);
+    }
+}