Encounter Manager ready to be used in large scale a-hoc network.

Author: thsc42

src/main/java/net/sharksystem/asap/ASAPEncounterManagerAdmin.java

@@ -17,32 +17,37 @@ public interface ASAPEncounterManagerAdmin {
     /**
      * @return set of ID to which an open connection exists right now.
      */
-    public abstract Set<CharSequence> getConnectedPeerIDs();
+    Set<CharSequence> getConnectedPeerIDs();
 
     /**
      * Add a peerID to what no connection should be established.
      * Note: Adding a peer to the deny list does not necessarily terminate an
      * existing connection to that peer.
      * @param peerID
      */
-    public abstract void addToDenyList(CharSequence peerID);
+    void addToDenyList(CharSequence peerID);
 
     /**
      * Remove a peerID from deny list
      * @param peerID
      */
-    public abstract void removeFromDenyList(CharSequence peerID);
+    void removeFromDenyList(CharSequence peerID);
 
     /**
      * Get PeerID set to which no connection should be established
      * @return
      */
-    public abstract Set<CharSequence> getDenyList();
+    Set<CharSequence> getDenyList();
+
+    /**
+     * remove all entries from deny list
+     */
+    void clearDenyList();
 
     /**
      * Cancel a connection to a peer. This method call does not change the deny list.
      * @param peerID
      */
-    public abstract void cancelConnection(CharSequence peerID);
+    void cancelConnection(CharSequence peerID);
 
 }

src/main/java/net/sharksystem/asap/ASAPEncounterManagerImpl.java

@@ -1,19 +1,25 @@
 package net.sharksystem.asap;
 
+import net.sharksystem.SharkException;
+import net.sharksystem.asap.fs.ExtraDataFS;
 import net.sharksystem.asap.protocol.ASAPConnection;
 import net.sharksystem.asap.protocol.ASAPConnectionListener;
+import net.sharksystem.asap.utils.ASAPSerialization;
 import net.sharksystem.asap.utils.PeerIDHelper;
 import net.sharksystem.utils.streams.StreamPair;
 import net.sharksystem.utils.Log;
 
 import java.io.*;
 import java.util.*;
 
-public class ASAPEncounterManagerImpl implements ASAPEncounterManager,
-        ASAPEncounterManagerAdmin,
+public class ASAPEncounterManagerImpl implements ASAPEncounterManager, ASAPEncounterManagerAdmin,
         ASAPConnectionListener {
     public static final long DEFAULT_WAIT_BEFORE_RECONNECT_TIME = 1000; // a second - debugging
     public static final long DEFAULT_WAIT_TO_AVOID_RACE_CONDITION = 500; // milliseconds - worked fine with BT.
+    public static final String DATASTORAGE_FILE_EXTENSION = "em";
+    private static final CharSequence ENCOUNTER_MANAGER_DENY_LIST_KEY = "denylist";
+    private final CharSequence peerID;
+    private ExtraDataFS extraDataStorage = null;
 
     private int randomValue;
     private long waitBeforeReconnect;
@@ -41,23 +47,35 @@ public class ASAPEncounterManagerImpl implements ASAPEncounterManager,
     /** remember remote address of peers (they can have more than one): peerID -> remote address */
     private Map<CharSequence, Set<CharSequence>> peerRemoteAddresses = new HashMap<>();
 
-    public ASAPEncounterManagerImpl(ASAPConnectionHandler asapConnectionHandler) {
-        this(asapConnectionHandler, DEFAULT_WAIT_BEFORE_RECONNECT_TIME);
+    public ASAPEncounterManagerImpl(ASAPConnectionHandler asapConnectionHandler, CharSequence peerID) throws SharkException, IOException {
+        this(asapConnectionHandler, peerID, DEFAULT_WAIT_BEFORE_RECONNECT_TIME);
     }
 
-    public ASAPEncounterManagerImpl(ASAPConnectionHandler asapConnectionHandler, long waitingPeriod) {
+    public ASAPEncounterManagerImpl(ASAPConnectionHandler asapConnectionHandler, CharSequence peerID,
+                                    long waitingPeriod, CharSequence rootFolder) throws SharkException, IOException {
         this.asapConnectionHandler = asapConnectionHandler;
+        this.peerID = peerID;
         this.randomValue = new Random(System.currentTimeMillis()).nextInt();
         this.waitBeforeReconnect = waitingPeriod;
-        this.restoreDenyList();
+
+        if(rootFolder != null && rootFolder.length() > 0) {
+            // create folder
+            this.extraDataStorage = new ExtraDataFS(rootFolder + "/" + ENCOUNTER_MANAGER_DENY_LIST_KEY, DATASTORAGE_FILE_EXTENSION);
+            this.restoreDenyList();
+        }
+    }
+
+    public ASAPEncounterManagerImpl(ASAPConnectionHandler asapConnectionHandler, CharSequence peerID,
+                                    long waitingPeriod) throws SharkException, IOException {
+        this(asapConnectionHandler, peerID, waitingPeriod, null);
     }
 
     private boolean coolDownOver(CharSequence id, ASAPEncounterConnectionType connectionType) {
         Date now = new Date();
         Date lastEncounter = this.encounterDate.get(id);
 
         if(lastEncounter == null) {
-            Log.writeLog(this, this.toString(), "device/peer not in encounteredDevices - should connect");
+            Log.writeLog(this, this.toString(), "device/peer not in encounteredDevices");
             this.encounterDate.put(id, now);
             return true;
         }
@@ -87,7 +105,13 @@ private boolean coolDownOver(CharSequence id, ASAPEncounterConnectionType connec
     }
 
     @Override
-    public boolean shouldCreateConnectionToPeer(CharSequence remoteAdressOrPeerID, ASAPEncounterConnectionType connectionType) {
+    public boolean shouldCreateConnectionToPeer(CharSequence remoteAdressOrPeerID,
+                                                ASAPEncounterConnectionType connectionType) {
+        Log.writeLog(this, this.toString(), "should connect to " + remoteAdressOrPeerID + " ?");
+        // on deny list?
+        if(this.denyList.contains(remoteAdressOrPeerID)) return false;
+        Log.writeLog(this, this.toString(), remoteAdressOrPeerID + " not on deny list");
+
         // do we have a connection under a peerID?
         StreamPair streamPair = this.openStreamPairs.get(remoteAdressOrPeerID);
         if(streamPair != null) {
@@ -131,29 +155,39 @@ public void handleEncounter(StreamPair streamPair, ASAPEncounterConnectionType c
 
     private void handleEncounter(StreamPair streamPair, ASAPEncounterConnectionType connectionType, boolean initiator,
                                  boolean raceCondition) throws IOException {
+        // always exchange peerIDs
+        DataOutputStream dos = new DataOutputStream(streamPair.getOutputStream());
+        dos.writeUTF(this.peerID.toString());
+        DataInputStream dis = new DataInputStream(streamPair.getInputStream());
+        String remotePeerID = dis.readUTF();
+
+        if(remotePeerID != null && remotePeerID.length() > 0) {
+            streamPair.setEndpointID(remotePeerID);
+        }
 
-        CharSequence streamPairID = streamPair.getSessionID();
+        CharSequence connectionID = streamPair.getEndpointID();
+        if(connectionID == null || connectionID.length() == 0) connectionID = streamPair.getSessionID();
 
-        Log.writeLog(this, this.toString(), "socket called: handle new encounter" + streamPair);
+        Log.writeLog(this, this.toString(), "decide whether to pursue this new encounter: " + streamPair);
 
         // should we connect in the first place
-        if (!this.shouldCreateConnectionToPeer(streamPairID, connectionType)) {
+        if (!this.shouldCreateConnectionToPeer(connectionID, connectionType)) {
             // no - than shut it down.
+            Log.writeLog(this, this.toString(),
+                    "close connection (on deny list or in cool down)");
             streamPair.close();
             return;
         }
 
         // new stream pair is ok. Is there a race condition expected ?
         if(raceCondition) {
             // avoid the nasty race condition
-            boolean waited = this.waitBeforeASAPSessionLaunch(
-                    streamPair.getInputStream(),
-                    streamPair.getOutputStream(),
-                    initiator, DEFAULT_WAIT_TO_AVOID_RACE_CONDITION);
+            Log.writeLog(this, this.toString(), "solve race condition");
+            boolean waited = this.solveRaceCondition(streamPair, initiator, DEFAULT_WAIT_TO_AVOID_RACE_CONDITION);
 
             // ask again?
             if (waited) {
-                if (!this.shouldCreateConnectionToPeer(streamPairID, connectionType)) {
+                if (!this.shouldCreateConnectionToPeer(connectionID, connectionType)) {
                     streamPair.close();
                     return;
                 }
@@ -162,7 +196,7 @@ private void handleEncounter(StreamPair streamPair, ASAPEncounterConnectionType
 
         // we a through with it - remember that new stream pair
         Log.writeLog(this, this.toString(), "remember streamPair: " + streamPair);
-        this.openStreamPairs.put(streamPairID, streamPair);
+        this.openStreamPairs.put(connectionID, streamPair);
 
         Log.writeLog(this, this.toString(), "going to launch a new asap connection");
 
@@ -174,35 +208,46 @@ private void handleEncounter(StreamPair streamPair, ASAPEncounterConnectionType
 
             asapConnection.addASAPConnectionListener(this);
 
-            this.openASAPConnections.put(asapConnection, streamPairID);
+            this.openASAPConnections.put(asapConnection, connectionID);
 
         } catch (IOException | ASAPException e) {
             Log.writeLog(this, this.toString(), "while launching asap connection: "
                     + e.getLocalizedMessage());
         }
     }
 
-    private boolean waitBeforeASAPSessionLaunch(InputStream is, OutputStream os, boolean connectionInitiator,
-                           long waitInMillis) throws IOException {
+    private boolean solveRaceCondition(StreamPair streamPair, boolean connectionInitiator,
+                                       long waitInMillis) throws IOException {
         // run a little negotiation before we start
-        DataOutputStream dos = new DataOutputStream(os);
+        DataOutputStream dos = new DataOutputStream(streamPair.getOutputStream());
         int remoteValue = 0;
+        String remotePeerID = null;
 
         try {
+            // write protocol unit
             dos.writeInt(this.randomValue);
-            DataInputStream dis = new DataInputStream(is);
+            dos.writeUTF(this.peerID.toString());
+
+            // read it
+            DataInputStream dis = new DataInputStream(streamPair.getInputStream());
             remoteValue = dis.readInt();
+            remotePeerID = dis.readUTF();
+
+            if(remotePeerID != null && remotePeerID.length() > 0) streamPair.setEndpointID(remotePeerID);
         } catch (IOException e) {
             // decision is made - this connection is dead
-            os.close();
-            is.close();
+            streamPair.close();
         }
 
         StringBuilder sb = new StringBuilder();
-        sb.append("try to solve race condition: localValue == ");
+        sb.append("try to solve race condition: random (local/remote) == ");
         sb.append(this.randomValue);
-        sb.append(" | remoteValue == ");
+        sb.append("/");
         sb.append(remoteValue);
+        sb.append(" | peerID == ");
+        sb.append(this.peerID);
+        sb.append("/");
+        sb.append(remotePeerID);
         sb.append(" | initiator == ");
         sb.append(connectionInitiator);
 
@@ -301,30 +346,59 @@ public synchronized void asapConnectionTerminated(Exception terminatingException
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
     private Set<CharSequence> denyList = new HashSet<>();
-
-    //// housekeeping
-    private void restoreDenyList() {
-        // TODO
-        //this.denyList = new HashSet<>();
-        Log.writeLog(this, "need to implement restoreDenyList");
+    
+    //// housekeeping deny list
+    public void clearDenyList() {
+        this.denyList = new HashSet<>();
+        try {
+            this.saveDenyList();
+        } catch (IOException | SharkException e) {
+            Log.writeLogErr(this, this.toString(), "cannot persist deny list: " + e.getLocalizedMessage());
+        }
+    }
+    private void restoreDenyList() throws SharkException, IOException {
+        if(this.extraDataStorage == null) {
+            Log.writeLog(this, this.toString(), "no persistent storage for deny list");
+        } else {
+            byte[] denyListBytes = this.extraDataStorage.getExtra(ENCOUNTER_MANAGER_DENY_LIST_KEY);
+            if(denyListBytes != null && denyListBytes.length > 0) {
+                ByteArrayInputStream bais = new ByteArrayInputStream(denyListBytes);
+                this.denyList = ASAPSerialization.readCharSequenceSetParameter(bais);
+            } else {
+                this.clearDenyList();
+            }
+        }
     }
 
-    private void saveDenyList() {
-        // TODO
-        Log.writeLog(this, "need to implement saveDenyList");
+    private void saveDenyList() throws IOException, SharkException {
+        if(this.extraDataStorage == null) {
+            Log.writeLog(this, this.toString(), "no persistent storage for deny list");
+        } else {
+            ByteArrayOutputStream baos = new ByteArrayOutputStream();
+            ASAPSerialization.writeCharSequenceSetParameter(this.denyList, baos);
+            this.extraDataStorage.putExtra(ENCOUNTER_MANAGER_DENY_LIST_KEY, baos.toByteArray());
+        }
     }
 
     ///// manage deny list
     @Override
     public void addToDenyList(CharSequence peerID) {
         this.denyList.add(peerID);
-        this.saveDenyList();
+        try {
+            this.saveDenyList();
+        } catch (IOException | SharkException e) {
+            Log.writeLogErr(this, this.toString(), "cannot persist deny list: " + e.getLocalizedMessage());
+        }
     }
 
     @Override
     public void removeFromDenyList(CharSequence peerID) {
         this.denyList.remove(peerID);
-        this.saveDenyList();
+        try {
+            this.saveDenyList();
+        } catch (IOException | SharkException e) {
+            Log.writeLogErr(this, this.toString(), "cannot persist deny list: " + e.getLocalizedMessage());
+        }
     }
 
     @Override
@@ -349,7 +423,8 @@ public void cancelConnection(CharSequence peerID) {
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
     public String toString() {
-        return this.asapConnectionHandler.toString();
+        if(this.asapConnectionHandler != null) return this.asapConnectionHandler.toString();
+        else return "null";
     }
 
 }

src/main/java/net/sharksystem/asap/apps/TCPServerSocketAcceptor.java

@@ -16,7 +16,7 @@ public TCPServerSocketAcceptor(int portNumber, ASAPEncounterManager encounterMan
         this.encounterManager = encounterManager;
         SocketFactory socketFactory = new SocketFactory(portNumber, this);
 
-        Log.writeLog(this, "start socket factory");
+        Log.writeLog(this, "start socket factory - no race condition assumed");
         new Thread(socketFactory).start();
     }
 

src/main/java/net/sharksystem/asap/fs/ExtraDataFS.java

@@ -20,9 +20,14 @@ public ExtraDataFS(CharSequence rootFolderName) throws SharkException, IOExcepti
 
     }
 
-    public ExtraDataFS(CharSequence rootFolderName, CharSequence fileExtension) throws IOException, SharkException {
-        this.rootFolderName = rootFolderName;
-        this.fileExtension = fileExtension;
+    public ExtraDataFS(CharSequence rootFolder, CharSequence fileName) throws IOException, SharkException {
+        this.rootFolderName = rootFolder;
+        this.fileExtension = fileName;
+        File rootFolderFile = new File(rootFolder.toString());
+        if(!rootFolderFile.exists()) {
+            // create
+            rootFolderFile.mkdirs();
+        }
         this.restoreExtraData();
     }
 
@@ -76,7 +81,8 @@ public void saveExtraData() throws IOException {
             // write key
             ASAPSerialization.writeCharSequenceParameter(key, os);
             // value
-            ASAPSerialization.writeByteArray(this.extraData.get(key), os);
+            byte[] v = this.extraData.get(key);
+            ASAPSerialization.writeByteArray(v, os);
         }
 
         os.close();

src/main/java/net/sharksystem/utils/streams/StreamPair.java

@@ -16,5 +16,7 @@ public interface StreamPair {
 
     CharSequence getSessionID();
 
-    CharSequence getEndpointAddress();
+    CharSequence getEndpointID();
+
+    void setEndpointID(CharSequence peerID);
 }