transient message exchange works; no infinite routing

Author: thsc42

src/main/java/net/sharksystem/asap/engine/ASAPEngine.java

@@ -412,12 +412,18 @@ public void handleASAPAssimilate(ASAP_AssimilationPDU_1_0 asapAssimilationPDU, A
         MessagesContainer messagesContainer = null;
         ASAPInMemoTransientMessages transientMessages = null;
 
-        if(eraSender != ASAP.TRANSIENT_ERA) {
-            incomingChunk = this.getIncomingChunk(asapAssimilationPDU);
-            messagesContainer = incomingChunk;
-        } else {
-            transientMessages = new ASAPInMemoTransientMessages(asapAssimilationPDU);
-            messagesContainer = transientMessages;
+        try {
+            if(eraSender != ASAP.TRANSIENT_ERA) {
+                    incomingChunk = this.getIncomingChunk(encounteredPeer, asapAssimilationPDU);
+                    messagesContainer = incomingChunk;
+            } else {
+                transientMessages = new ASAPInMemoTransientMessages(asapAssimilationPDU);
+                messagesContainer = transientMessages;
+            }
+        }
+        catch(ASAPException e) {
+            asapAssimilationPDU.takeDataFromStream();
+            throw e;
         }
 
         // put messages into container - incoming chunk or transient message container
@@ -476,7 +482,7 @@ public void handleASAPAssimilate(ASAP_AssimilationPDU_1_0 asapAssimilationPDU, A
         }
     }
 
-    private ASAPInternalChunk getIncomingChunk(ASAP_AssimilationPDU_1_0 asapAssimilationPDU)
+    private ASAPInternalChunk getIncomingChunk(String encounteredPeer, ASAP_AssimilationPDU_1_0 asapAssimilationPDU)
             throws IOException, ASAPException {
 
         String uri = asapAssimilationPDU.getChannelUri();
@@ -486,50 +492,53 @@ private ASAPInternalChunk getIncomingChunk(ASAP_AssimilationPDU_1_0 asapAssimila
         ASAPChunkStorage incomingChunkStorage = incomingStorage.getChunkStorage();
         Log.writeLog(this, this.toString(), "got incoming chunk storage for senderE2E: " + senderE2E);
 
-        try {
             // get local target for data to come
+        if (!incomingChunkStorage.existsChunk(uri, eraSender)) {
             ASAPInternalChunk localChunk = null;
+            // is there a local chunk - to clone recipients from?
+            if (this.channelExists(uri)) {
+                localChunk = this.getStorage().getChunk(uri, this.getEra());
+            } else {
+                Log.writeLog(this, this.toString(), "asked to set up new channel: (uri/senderE2E): "
+                        + uri + " | " + senderE2E);
+                // this channel is new to local peer - am I allowed to create it?
+                if (!this.securityAdministrator.allowedToCreateChannel(asapAssimilationPDU)) {
+                    Log.writeLog(this, this.toString(),
+                            ".. not allowed .. TODO not yet implemented .. always set up");
 
-            if (!incomingChunkStorage.existsChunk(uri, eraSender)) {
-                // is there a local chunk - to clone recipients from?
-                if (this.channelExists(uri)) {
-                    localChunk = this.getStorage().getChunk(uri, this.getEra());
+                    //allowedAssimilation = false; // TODO
                 } else {
-                    Log.writeLog(this, this.toString(), "asked to set up new channel: (uri/senderE2E): "
-                            + uri + " | " + senderE2E);
-                    // this channel is new to local peer - am I allowed to create it?
-                    if (!this.securityAdministrator.allowedToCreateChannel(asapAssimilationPDU)) {
-                        Log.writeLog(this, this.toString(),
-                                ".. not allowed .. TODO not yet implemented .. always set up");
-
-                        //allowedAssimilation = false; // TODO
-                    } else {
-                        Log.writeLog(this, this.toString(), "allowed. Set it up.");
-                        this.createChannel(uri);
-                    }
+                    Log.writeLog(this, this.toString(), "allowed. Set it up.");
+                    this.createChannel(uri);
                 }
-            } else {
-                Log.writeLog(this, this.toString(), "received chunk that already exists - did nothing: "
-                        + senderE2E + " | " + eraSender + " | " + uri);
-
-                // read assimilation message payload to oblivion!
-                asapAssimilationPDU.takeDataFromStream();
             }
-
             ASAPInternalChunk incomingChunk = incomingStorage.createNewChunk(uri, eraSender);
-            //ASAPInternalChunk incomingChunk = incomingChunkStorage.getChunk(uri, eraSender);
 
             if (localChunk != null) {
                 Log.writeLog(this, this.toString(), "copy local meta data into newly created incoming chunk");
                 incomingChunk.copyMetaData(this.getChannel(uri));
             }
-
             return incomingChunk;
+        } else {
+            // already exists. Add message if sender == originator and era last era of this channel
+            if(incomingStorage.getEra() == eraSender // era is current era
+                    && PeerIDHelper.sameID(encounteredPeer, senderE2E) // E2E sender == P2P sender
+                    ) {
+                Log.writeLog(this, this.toString(),
+                        "received chunk exists but sender is originator and current era: "
+                        + senderE2E + " | " + eraSender + " | " + uri);
 
-        } catch (IOException | ASAPException e) {
-            Log.writeLogErr(this, this.toString(),
-                    "exception (give up, keep streams untouched): " + e.getLocalizedMessage());
-            throw e;
+                // finish routing...
+                if(incomingChunkStorage.getChunk(uri, eraSender).getASAPHopList().size() > 1) {
+                    throw new ASAPException("chunk was already routed to me - met originator - will no overwrite chunk");
+                }
+
+                return incomingChunkStorage.getChunk(uri, eraSender);
+            } else {
+                // read assimilation message payload to oblivion!
+                throw new ASAPException("chunk that already exists; not current era or not originator: "
+                        + senderE2E + " | " + eraSender + " | " + uri);
+            }
         }
     }
 

src/main/java/net/sharksystem/asap/protocol/ASAPPersistentConnection.java

@@ -429,7 +429,7 @@ public void run() {
                 }
             }
             catch(ASAPException asape) {
-                Log.writeLogErr(this, "asap exception while processing PDU - but go ahead: "
+                Log.writeLog(this, "while processing PDU (go ahead): "
                         + asape.getLocalizedMessage());
             }
             catch(IOException ioe) {

src/test/java/net/sharksystem/asap/engine/MultihopTests.java

@@ -358,9 +358,9 @@ public void asapRoutingIsFiniteAndCheckEra() throws IOException, ASAPException,
         System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<");
         alicePeer.startEncounter(TestHelper.getPortNumber(), bobPeer);
         // give your app a moment to process
-        Thread.sleep(1000);
+        Thread.sleep(500);
         alicePeer.stopEncounter(bobPeer);
-        Thread.sleep(1000);
+        Thread.sleep(500);
         Assert.assertEquals(1, bobListener.numberOfMessages);
 
         // check local eras
@@ -375,9 +375,9 @@ public void asapRoutingIsFiniteAndCheckEra() throws IOException, ASAPException,
         System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<");
         claraPeer.startEncounter(TestHelper.getPortNumber(), bobPeer);
         // give your app a moment to process
-        Thread.sleep(1000);
+        Thread.sleep(500);
         claraPeer.stopEncounter(bobPeer);
-        Thread.sleep(1000);
+        Thread.sleep(500);
         Assert.assertEquals(1, claraListener.numberOfMessages);
 
         // no change - receiving message does not change local understanding of an era
@@ -392,9 +392,9 @@ public void asapRoutingIsFiniteAndCheckEra() throws IOException, ASAPException,
         System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<");
         alicePeer.startEncounter(TestHelper.getPortNumber(), claraPeer);
         // give your app a moment to process
-        Thread.sleep(1000);
+        Thread.sleep(500);
         alicePeer.stopEncounter(claraPeer);
-        Thread.sleep(1000);
+        Thread.sleep(500);
         Assert.assertEquals(0, claraListener.numberOfMessages);
 
         // no change - receiving message does not change local understanding of an era