refactured pdu executor - made it inner class of persistent connection

Author: thsc42

src/net/sharksystem/asap/ASAPEngine.java

@@ -31,11 +31,11 @@ public abstract class ASAPEngine implements ASAPStorage, ASAPProtocolEngine {
     protected ASAPMemento memento = null;
 
     /* private */ final private ASAPChunkStorage chunkStorage;
-    private boolean dropDeliveredChunks = false;
+    protected boolean dropDeliveredChunks = false;
 
     private ASAPOnlineMessageSender asapOnlineMessageSender;
     protected boolean contentChanged = false;
-    private boolean sendReceivedChunks = false;
+    protected boolean sendReceivedChunks = false;
 
     protected ASAPEngine(ASAPChunkStorage chunkStorage, CharSequence chunkContentFormat)
             throws ASAPException, IOException {
@@ -436,16 +436,24 @@ public void handleASAPInterest(ASAP_Interest_PDU_1_0 asapInterest, ASAP_1_0 prot
         //>>>>>>>>>>>>>>>>>>>debug
 
         if(this.isSendReceivedChunks()) {
+            System.out.println(this.getLogStart() + "send also received chunks - if any");
 
+            for(CharSequence sender : this.getSender()) {
+                System.out.println(this.getLogStart() + "send chunks received from: " + sender);
+                ASAPChunkStorage incomingChunkStorage = this.getIncomingChunkStorage(sender);
+
+                this.sendChunks(peer, incomingChunkStorage, protocol, workingEra, lastEra, os);
+            }
         }
     }
 
     private boolean isSendReceivedChunks() {
         return this.sendReceivedChunks;
     }
 
-    public void setSendReceivedChunks(boolean on) {
+    public void setSendReceivedChunks(boolean on) throws IOException {
         this.sendReceivedChunks = on;
+        this.saveStatus();
     }
 
     private void sendChunks(String peer, ASAPChunkStorage chunkStorage,
@@ -564,8 +572,9 @@ private boolean isDropDeliveredChunks() {
         return this.dropDeliveredChunks;
     }
 
-    public void setDropDeliveredChunks(boolean drop) {
+    public void setDropDeliveredChunks(boolean drop) throws IOException {
         this.dropDeliveredChunks = drop;
+        this.saveStatus();
     }
 
     static int nextEra(int workingEra) {

src/net/sharksystem/asap/ASAPMementoFS.java

@@ -41,6 +41,8 @@ public void save(ASAPEngine engine) throws IOException {
         dos.writeInt(engine.era);
         dos.writeInt(engine.oldestEra);
         dos.writeBoolean(engine.contentChanged);
+        dos.writeBoolean(engine.dropDeliveredChunks);
+        dos.writeBoolean(engine.sendReceivedChunks);
 
         // write lastSeen hash map
         if(engine.lastSeen != null && !engine.lastSeen.isEmpty()) {
@@ -61,6 +63,8 @@ private void setDefaults(ASAPEngine engine) {
         engine.era = ASAPEngine.DEFAULT_INIT_ERA;
         engine.oldestEra = ASAPEngine.DEFAULT_INIT_ERA;
         engine.lastSeen = new HashMap<>();
+        engine.dropDeliveredChunks = false;
+        engine.sendReceivedChunks = false;
     }
 
     public void restore(ASAPEngine engine) throws IOException {
@@ -80,6 +84,8 @@ public void restore(ASAPEngine engine) throws IOException {
         engine.era = dis.readInt();
         engine.oldestEra = dis.readInt();
         engine.contentChanged = dis.readBoolean();
+        engine.dropDeliveredChunks = dis.readBoolean();
+        engine.sendReceivedChunks = dis.readBoolean();
 
         // try to read lastSeen list
         boolean first = true;

src/net/sharksystem/asap/ASAPPDUExecutor.java

@@ -32,11 +32,11 @@ void handleASAPAssimilate(ASAP_AssimilationPDU_1_0 asap_assimilation, ASAP_1_0 p
      * to all their recipients. If no set, they remain intact.
      * @param drop
      */
-    void setDropDeliveredChunks(boolean drop);
+    void setDropDeliveredChunks(boolean drop) throws IOException;
 
     /**
      * engine can deliver local message but also received messages - default false - send no received messages
      * @param on
      */
-    void setSendReceivedChunks(boolean on);
+    void setSendReceivedChunks(boolean on) throws IOException;
 }

src/net/sharksystem/asap/ASAPProtocolEngine.java

@@ -70,6 +70,23 @@ public interface ASAPStorage {
      */
     public void setRecipients(CharSequence urlTarget, List<CharSequence> recipients) throws IOException;
 
+    /**
+     * Chunks are delivered when seeing other peers. This flag allows to decide whether delivered chunks
+     * are to be deleted.
+     * @param drop
+     */
+    void setDropDeliveredChunks(boolean drop) throws IOException;
+
+    /**
+     * Chunks are delivered when seeing other peers. Default behaviour is to send only message which
+     * are in local peers own storage. A peer can also have received messages in an incoming storage.
+     * This flag allows to force even delivery of received messages from incoming storages. Basis of
+     * multihop communication.
+     *
+     * @param drop
+     */
+    void setSendReceivedChunks(boolean drop) throws IOException;
+
     /**
      /**
      * returns recipient list

src/net/sharksystem/asap/ASAPStorage.java

@@ -1,9 +1,9 @@
 package net.sharksystem.asap;
 
-class EngineSetting {
-    final CharSequence folder;
-    ASAPChunkReceivedListener listener;
-    ASAPEngine engine;
+public class EngineSetting {
+    public final CharSequence folder;
+    public ASAPChunkReceivedListener listener;
+    public ASAPEngine engine;
 
     EngineSetting(CharSequence folder, ASAPChunkReceivedListener listener) {
         this.folder = folder;

src/net/sharksystem/asap/EngineSetting.java

@@ -65,4 +65,6 @@ Thread getExecutorThread(ASAP_PDU_1_0 asappdu, InputStream is, OutputStream os,
     void addOnlinePeersChangedListener(ASAPOnlinePeersChangedListener listener);
 
     void removeOnlinePeersChangedListener(ASAPOnlinePeersChangedListener listener);
+
+    EngineSetting getEngineSettings(CharSequence format) throws ASAPException;
 }

src/net/sharksystem/asap/MultiASAPEngineFS.java

@@ -147,7 +147,7 @@ public ASAPEngine getASAPEngine(CharSequence appName, CharSequence format)
         return asapEngine;
     }
 
-    private EngineSetting getEngineSettings(CharSequence format) throws ASAPException {
+    public EngineSetting getEngineSettings(CharSequence format) throws ASAPException {
         EngineSetting folderAndListener = folderMap.get(format);
         if(folderAndListener == null)
             throw new ASAPException("no folder for owner / format: " + owner + "/" + format);
@@ -351,7 +351,7 @@ public void pushInterests(OutputStream os) throws IOException, ASAPException {
     public Thread getExecutorThread(ASAP_PDU_1_0 asappdu, InputStream is, OutputStream os,
                                     ThreadFinishedListener threadFinishedListener) throws ASAPException {
         // process pdu
-        return new ASAPPDUExecutor(asappdu, is, os,
+        return new ASAPPersistentConnection.ASAPPDUExecutor(asappdu, is, os,
                 this.getEngineSettings(asappdu.getFormat()),
                 new ASAP_Modem_Impl(), threadFinishedListener);
     }

src/net/sharksystem/asap/MultiASAPEngineFS_Impl.java

@@ -36,7 +36,7 @@ public ASAPPersistentConnection(InputStream is, OutputStream os, MultiASAPEngine
     }
 
     private String getLogStart() {
-        return this.getClass().getSimpleName() + ": ";
+        return this.getClass().getSimpleName() + "(connected to: " + this.remotePeer + "): ";
     }
 
     private void setRemotePeer(String remotePeerName) {
@@ -225,26 +225,37 @@ public void run() {
                 System.out.println(this.getLogStart() + "read valid pdu");
                 this.setRemotePeer(asappdu.getPeer());
                 // process received pdu
-                try {
-                    this.executor =
-                            this.multiASAPEngineFS.getExecutorThread(asappdu, this.is, this.os, this);
-                    // get exclusive access to streams
-                    System.out.println(this.startLog() + "asap pdu executor going to wait for stream access");
-                    this.wait4ExclusiveStreamsAccess();
+
+                if(asappdu.getFormat().equalsIgnoreCase(ASAP_1_0.ASAP_MANAGEMENT_FORMAT.toString())) {
+                    System.out.println(this.getLogStart()
+                            + "got asap management message - not processed, took remote peer name only");
+                } else {
                     try {
-                        System.out.println(this.startLog() + "asap pdu executor got stream access - process pdu");
-                        this.runObservedThread(executor, maxExecutionTime);
-                    } catch (ASAPExecTimeExceededException e) {
-                        System.out.println(this.startLog() + "asap pdu processing took longer than allowed");
-                        this.terminate("asap pdu processing took longer than allowed", e);
-                        break;
-                    } finally {
-                        // wake waiting thread if any
-                        this.releaseStreamsLock();
-                        System.out.println(this.startLog() + "asap pdu executor release locks");
+                        this.executor =
+                                new ASAPPDUExecutor(asappdu,
+                                        this.is, this.os,
+                                        this.multiASAPEngineFS.getEngineSettings(asappdu.getFormat()),
+                                        new ASAP_Modem_Impl(),
+                                        this);
+//                        this.multiASAPEngineFS.getExecutorThread(asappdu, this.is, this.os, this);
+                        // get exclusive access to streams
+                        System.out.println(this.startLog() + "asap pdu executor going to wait for stream access");
+                        this.wait4ExclusiveStreamsAccess();
+                        try {
+                            System.out.println(this.startLog() + "asap pdu executor got stream access - process pdu");
+                            this.runObservedThread(executor, maxExecutionTime);
+                        } catch (ASAPExecTimeExceededException e) {
+                            System.out.println(this.startLog() + "asap pdu processing took longer than allowed");
+                            this.terminate("asap pdu processing took longer than allowed", e);
+                            break;
+                        } finally {
+                            // wake waiting thread if any
+                            this.releaseStreamsLock();
+                            System.out.println(this.startLog() + "asap pdu executor release locks");
+                        }
+                    } catch (ASAPException e) {
+                        System.out.println(this.getLogStart() + " problem when executing asap received pdu: " + e);
                     }
-                }  catch (ASAPException e) {
-                    System.out.println(this.getLogStart() + " problem when executing asap received pdu: " + e);
                 }
             }
         }
@@ -327,5 +338,84 @@ private StringBuilder startLog() {
 
         return sb;
     }
-}
 
+    public static class ASAPPDUExecutor extends Thread {
+        private final ASAP_PDU_1_0 asapPDU;
+        private final InputStream is;
+        private final OutputStream os;
+        private final EngineSetting engineSetting;
+        private final ASAP_1_0 protocol;
+        private final ThreadFinishedListener threadFinishedListener;
+
+        public ASAPPDUExecutor(ASAP_PDU_1_0 asapPDU, InputStream is, OutputStream os,
+                               EngineSetting engineSetting, ASAP_1_0 protocol,
+                               ThreadFinishedListener threadFinishedListener) {
+            this.asapPDU = asapPDU;
+            this.is = is;
+            this.os = os;
+            this.engineSetting = engineSetting;
+            this.protocol = protocol;
+            this.threadFinishedListener = threadFinishedListener;
+
+            StringBuilder sb = new StringBuilder();
+            sb.append("ASAPPDUExecutor created - ");
+            sb.append("folder: " + engineSetting.folder + " | ");
+            sb.append("engine: " + engineSetting.engine.getClass().getSimpleName() + " | ");
+            if(engineSetting.listener != null) {
+                sb.append("listener: " + engineSetting.listener.getClass().getSimpleName());
+            }
+
+            System.out.println(sb.toString());
+        }
+
+        private void finish() {
+            if(this.threadFinishedListener != null) {
+                this.threadFinishedListener.finished(this);
+            }
+        }
+
+        public void run() {
+            if(engineSetting.engine == null) {
+                System.err.println("ASAPPDUExecutor called without engine set - fatal");
+                this.finish();
+                return;
+            }
+
+            System.out.println("ASAPPDUExecutor calls engine: " + engineSetting.engine.getClass().getSimpleName());
+
+            try {
+                switch (asapPDU.getCommand()) {
+                    case ASAP_1_0.INTEREST_CMD:
+                        System.out.println("ASAPPDUExecutor call handleASAPInterest");
+                        engineSetting.engine.handleASAPInterest((ASAP_Interest_PDU_1_0) asapPDU, protocol, os);
+                        break;
+                    case ASAP_1_0.OFFER_CMD:
+                        System.out.println("ASAPPDUExecutor call handleASAPOffer");
+                        engineSetting.engine.handleASAPOffer((ASAP_OfferPDU_1_0) asapPDU, protocol, os);
+                        break;
+                    case ASAP_1_0.ASSIMILATE_CMD:
+                        System.out.println("ASAPPDUExecutor call handleASAPAssimilate");
+                        engineSetting.engine.handleASAPAssimilate((ASAP_AssimilationPDU_1_0) asapPDU, protocol, is, os,
+                                engineSetting.listener);
+                        break;
+
+                    default:
+                        System.err.println(
+                                this.getClass().getSimpleName() + ": " + "unknown ASAP command: " + asapPDU.getCommand());
+                }
+            }
+            catch(IOException | ASAPException e) {
+                System.err.println("Exception while processing ASAP PDU - close streams" + e.getLocalizedMessage());
+                try {
+                    os.close(); // more important to close than input stream - do it first
+                    is.close();
+                } catch (IOException ex) {
+                    ex.printStackTrace();
+                }
+            }
+            finally {
+                this.finish();
+            }
+        }
+    }
+}