From e28d980c61c78bcf8341c2f2bbd82ecc166deb6c Mon Sep 17 00:00:00 2001
From: Rob Austin <rob.austin@boundedbuffer.co.uk>
Date: Wed, 4 Mar 2015 10:28:23 +0000
Subject: [PATCH] added number based tests - still work in progress

---
 .../map/StatelessWiredConnector.java          | 46 +++++++++++--------
 .../WiredStatelessClientTcpConnectionHub.java | 33 ++++++-------
 .../net/openhft/chronicle/wire/TestWire.java  |  3 +-
 3 files changed, 45 insertions(+), 37 deletions(-)

diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
index c1f6042c4..810770b35 100644
--- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
@@ -153,8 +153,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
     public void onRead(SocketChannel socketChannel, SelectionKey key) throws IOException {
         this.key = key;
 
-        if (!handshingComplete)
-            outWire.bytes().writeByte(localIdentifier);
+       // if (!handshingComplete)
+        //    outWire.bytes().writeByte(localIdentifier);
 
         final int len = readSocket(socketChannel);
         assert inWire.bytes().remaining() < 999;
@@ -184,16 +184,16 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
                 return;
             }
 
-            if (!handshingComplete) {
-                System.out.println("doHandshaking");
+            // if (!handshingComplete) {
+            System.out.println("doHandshaking");
 
-                onHandShaking();
-                System.out.println("end-doHandshaking");
-            } else {
-                System.out.println("onEvent");
-                onEvent();
-                System.out.println("end-onEvent");
-            }
+            //  onHandShaking();
+            //    System.out.println("end-doHandshaking");
+            // } else {
+            System.out.println("onEvent");
+            onEvent();
+            System.out.println("end-onEvent");
+            // }
 
             System.out.println("limit=>" + outWire.bytes().limit());
             assert inWire.bytes().remaining() < 999;
@@ -279,7 +279,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
     }
 
     void onHandShaking() {
-        remoteIdentifier = inWire.read(() -> "IDENTIFIER").int8();
+        //   remoteIdentifier = inWire.read(() -> "IDENTIFIER").int8();
         handshingComplete = true;
     }
 
@@ -287,21 +287,29 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
     Work onEvent() {
 
         // it is assumed by this point that the buffer has all the bytes in it for this message
+
+        System.out.println("StatelessWiredConnector - trans=" + Bytes.toHex(inWire.bytes()));
         long transactionId = inWire.read(() -> "TRANSACTION_ID").int64();
         timestamp = inWire.read(() -> "TIME_STAMP").int64();
         channelId = inWire.read(() -> "CHANNEL_ID").int16();
         inWire.read(() -> "METHOD_NAME").text(methodName);
 
         // for the length
+        long markStart = outWire.bytes().position();
         outWire.bytes().skip(4);
 
-        // write the transaction id
-        outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
+        try {
+            // write the transaction id
+            outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
 
-        if ("PUT".contentEquals(methodName))
+            if ("PUT".contentEquals(methodName))
                 return put();
-        else
-            throw new IllegalStateException("unsupported event=" + methodName);
+            else
+                throw new IllegalStateException("unsupported event=" + methodName);
+        } finally {
+            int len = (int) (outWire.bytes().position() - markStart);
+            outWire.bytes().writeInt(markStart, len);
+        }
 
     }
 
@@ -436,8 +444,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
             // todo remove
             e.printStackTrace();
 
-            // move back to the start
-            outWire.bytes().position(2);
+            // move back to just after the size
+            outWire.bytes().position(4);
 
             return sendException(outWire, e);
         } finally {
diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
index 4954ec7ea..823ce7b67 100644
--- a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
+++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
@@ -39,7 +39,6 @@ import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 
 import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
-import static net.openhft.chronicle.map.AbstractChannelReplicator.openSocketChannel;
 import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.APPLICATION_VERSION;
 
 /**
@@ -186,7 +185,7 @@ public class WiredStatelessClientTcpConnectionHub {
         }
         clientChannel = result;
 
-        ByteBuffer byteBuffer = ByteBuffer.allocateDirect(128);
+       /* ByteBuffer byteBuffer = ByteBuffer.allocateDirect(128);
         Bytes<ByteBuffer> bufferBytes = Bytes.wrap(byteBuffer);
         TextWire wire = new TextWire(bufferBytes);
         wire.bytes().skip(4);
@@ -221,7 +220,7 @@ public class WiredStatelessClientTcpConnectionHub {
                 //
             }
 
-        }
+        }*/
 
     }
 
@@ -328,7 +327,7 @@ public class WiredStatelessClientTcpConnectionHub {
 
     @SuppressWarnings("WeakerAccess")
     @NotNull
-      String clientVersion() {
+    String clientVersion() {
         return BuildVersion.version();
     }
 
@@ -336,7 +335,7 @@ public class WiredStatelessClientTcpConnectionHub {
     /**
      * sends data to the server via TCP/IP
      */
-      void writeSocket() {
+    void writeSocket() {
 
         assert outBytesLock().isHeldByCurrentThread();
         assert !inBytesLock().isHeldByCurrentThread();
@@ -430,11 +429,13 @@ public class WiredStatelessClientTcpConnectionHub {
                 final int remainingBytes0 = messageSize - SIZE_OF_SIZE;
                 readSocket(remainingBytes0, timeoutTime);
 
-                 intWire.bytes().skip(4);
+                intWire.bytes().skip(4);
                 intWire.bytes().limit(messageSize);
-                String text = intWire.read(() -> "TRANSACTION_ID").text();
-                System.out.println(text);
-                long transactionId0 =0;
+                System.out.println("trans="+Bytes.toHex(intWire.bytes()));
+                long  trans = intWire.read(() -> "TRANSACTION_ID").int64();
+                System.out.println(trans);
+                long transactionId0 = 0;
+
                 System.out.println("transactionId0=" + transactionId0);
                 // check the transaction id is reasonable
           /*      assert transactionId0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR :
@@ -633,8 +634,8 @@ public class WiredStatelessClientTcpConnectionHub {
 
     @SuppressWarnings("SameParameterValue")
     @Nullable
-      <O extends Marshallable> O fetchObject(final Class<O> tClass,
-                                             @NotNull final String methodName, short channelID) {
+    <O extends Marshallable> O fetchObject(final Class<O> tClass,
+                                           @NotNull final String methodName, short channelID) {
         final long startTime = System.currentTimeMillis();
         long transactionId;
 
@@ -663,7 +664,7 @@ public class WiredStatelessClientTcpConnectionHub {
 
     @SuppressWarnings("SameParameterValue")
     @Nullable
-      String proxyReturnString(@NotNull final String messageId, short channelID) {
+    String proxyReturnString(@NotNull final String messageId, short channelID) {
         final long startTime = System.currentTimeMillis();
         long transactionId;
 
@@ -712,11 +713,11 @@ public class WiredStatelessClientTcpConnectionHub {
     /**
      * mark the location of the outWire size
      */
-      void markSize() {
+    void markSize() {
 
-          assert outBytesLock().isHeldByCurrentThread();
+        assert outBytesLock().isHeldByCurrentThread();
 
-          // this is where the size will go
+        // this is where the size will go
         sizeMark = outWire.bytes().position();
 
 
@@ -724,7 +725,7 @@ public class WiredStatelessClientTcpConnectionHub {
         outWire.bytes().skip(4);
     }
 
-      void startTime(long startTime) {
+    void startTime(long startTime) {
         this.startTime = startTime;
     }
 }
diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/wire/TestWire.java
index 849861290..ceed64b65 100644
--- a/src/test/java/net/openhft/chronicle/wire/TestWire.java
+++ b/src/test/java/net/openhft/chronicle/wire/TestWire.java
@@ -26,7 +26,6 @@ import net.openhft.chronicle.map.WiredChronicleMapStatelessClientBuilder;
 import net.openhft.lang.io.ByteBufferBytes;
 import net.openhft.lang.io.Bytes;
 import org.jetbrains.annotations.NotNull;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import java.io.IOException;
@@ -62,7 +61,7 @@ public class TestWire {
 
     private ReplicationHub hubA;
 
-    @Ignore
+
     @Test
     public void test() throws IOException {