multichannel works - still a race condition - I think(!) it's because all threads from both peer runs in a single java engine. See test for details

Author: thsc42

src/net/sharksystem/hub/hubside/HubGenericImpl.java

@@ -155,10 +155,17 @@ void connectionCreated(CharSequence sourcePeerID, CharSequence targetPeerID, Str
                 StreamPairLink dataLink =
                         new StreamPairLink(dataSessionRequest.connection, sourcePeerID, connection, targetPeerID, false);
 
+                /*
                 if(dataSessionRequest.connectionPreparer != null)
                     dataSessionRequest.connectionPreparer.prepare(dataSessionRequest.connection);
 
                 if(connectionPreparer != null) connectionPreparer.prepare(connection);
+                 */
+
+                // tell peers e2e is established
+                Log.writeLog(this, "send ready byte to each peer");
+                dataSessionRequest.connection.getOutputStream().write(ConnectionPreparer.readyByte);
+                connection.getOutputStream().write(ConnectionPreparer.readyByte);
 
                 // now go
                 dataLink.start();

src/net/sharksystem/hub/peerside/SharedChannelConnectorPeerSide.java

@@ -239,7 +239,26 @@ protected void silenceEnded() { }
 
     @Override
     protected void dataSessionStarted(CharSequence targetPeerID, StreamPair streamPair) {
-        Log.writeLog(this, this.toString(), "data session started to peer: " + targetPeerID);
+        Log.writeLog(this, this.toString(), "data session started to peer " + targetPeerID);
+
+        Log.writeLog(this, this.toString(), "wait for ready byte " + targetPeerID);
+
+        byte b = 0;
+        while(b != ConnectionPreparer.readyByte) {
+            Log.writeLog(this, this.toString(), "no ready byte yet");
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+                // ignore
+            }
+            try {
+                b = (byte) streamPair.getInputStream().read();
+            } catch (IOException e) {
+                Log.writeLog(this, this.toString(), "connection gone before usage, other peer: "
+                        + targetPeerID);
+            }
+        }
+        Log.writeLog(this, this.toString(), "got ready byte from hub - notify data session can begin");
 
         // tell listener
         if(this.listener != null) {

tests/net/sharksystem/hub/HubConnectorTester.java

@@ -73,8 +73,13 @@ public void notifyPeerConnected(CharSequence targetPeerID, StreamPair streamPair
                 System.out.println(peerID + " writes pure bytes: " + this.messageA);
                 streamPair.getOutputStream().write(testBytes);
                 byte[] receivedBytes = new byte[10];
-                System.out.println(peerID + " available: " + streamPair.getInputStream().available());
-                //Thread.sleep(100); // give other process a moment
+                int available = streamPair.getInputStream().available();
+                if(available < 1) {
+                    //System.out.println(peerID + " try to go in read block");
+                    System.out.println(peerID + " sleep");
+                    //streamPair.getInputStream().read();
+                    Thread.sleep(100); // give other process a moment
+                }
                 System.out.println(peerID + " available #2: " + streamPair.getInputStream().available());
                 streamPair.getInputStream().read(receivedBytes);
                 System.out.println(peerID + " available #3: " + streamPair.getInputStream().available());
@@ -89,12 +94,15 @@ public void notifyPeerConnected(CharSequence targetPeerID, StreamPair streamPair
             byte[] outBytes = baos.toByteArray();
             ASAPSerialization.writeByteArray(outBytes, streamPair.getOutputStream());
 
+            System.out.println(peerID + " sleep - other threads need to get a chance. Is that still a bug? ");
+            Thread.sleep(100); // give other process a moment
+
             // read
             System.out.println(peerID + " reading..");
             if(PeerIDHelper.sameID(peerID, ALICE_ID)) {
                 int i = 42; // debug break
             }
-            Thread.sleep(1000);
+            //Thread.sleep(1000);
             byte[] inBytes = ASAPSerialization.readByteArray(streamPair.getInputStream());
             ByteArrayInputStream bais = new ByteArrayInputStream(inBytes);
             String receivedMessage = ASAPSerialization.readCharSequenceParameter(bais);
@@ -118,6 +126,9 @@ public void notifyPeerConnected(CharSequence targetPeerID, StreamPair streamPair
             outBytes = baos.toByteArray();
             ASAPSerialization.writeByteArray(outBytes, streamPair.getOutputStream());
 
+            System.out.println(peerID + " sleep - other threads need to get a chance. Is that still a bug? ");
+            Thread.sleep(100); // give other process a moment
+
             System.out.println(this.peerID + " reading.. ");
             inBytes = ASAPSerialization.readByteArray(streamPair.getInputStream());
             bais = new ByteArrayInputStream(inBytes);