Tests: Add test JNI-153 - implement/emulate pEp4FM threading model

test/java/foundation/pEp/jniadapter/test/jni153/CTXMultiNode.java

@@ -1,7 +1,6 @@
 package foundation.pEp.jniadapter.test.jni153;
 
 import foundation.pEp.jniadapter.Engine;
-import foundation.pEp.jniadapter.Identity;
 import foundation.pEp.jniadapter.Message;
 import foundation.pEp.jniadapter.decrypt_message_Return;
 import foundation.pEp.jniadapter.test.utils.TestCallbacks;
@@ -13,7 +12,7 @@ import foundation.pEp.pitytest.AbstractTestContext;
 import foundation.pEp.pitytest.TestContextInterface;
 
 import java.util.List;
-import java.util.Vector;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.stream.Collectors;
 
 import static foundation.pEp.pitytest.TestLogger.log;
@@ -21,16 +20,18 @@ import static foundation.pEp.pitytest.TestLogger.log;
 public class CTXMultiNode extends AbstractTestContext {
     public Engine engine;
     public TestCallbacks callbacks;
-    public FsMQManager transport;
+    public Transport transport;
 
     // Model
     public TestModel model;
+
     // Mappings
     private NodeName ownNodeName;
     private TestNode ownNode;
     public TestIdentity myself;
     public TestIdentity partner;
 
+
     CTXMultiNode(NodeName ownNodeName) {
         this.ownNodeName = ownNodeName;
     }
@@ -54,19 +55,68 @@ public class CTXMultiNode extends AbstractTestContext {
         // Transport
         // Create own transport identity and Transport
         FsMQIdentity transportIdent = myself.getTransportIdent(ownNodeName);
-        transport = new FsMQManager(transportIdent);
+        transport = new Transport(transportIdent, model.getAllIdents());
+
+        return this;
+    }
+
+    private TestModel setupModel() {
+        TestModel ret = new TestModel();
+
+        ret.getNode(NodeName.NODE_A1).setRole(Role.ALICE);
+        ret.getNode(NodeName.NODE_B1).setRole(Role.BOB);
+
+        ret.getIdent(Role.ALICE).setDefaultPartner(Role.BOB);
+        ret.getIdent(Role.BOB).setDefaultPartner(Role.ALICE);
+//        ret.getIdent(Role.CAROL).setDefaultPartner(Role.ALICE);
+
+        return ret;
+    }
+
+}
 
-        // Add all transport identities of the model
+class Transport {
-        for (TestIdentity ti : model.getAllIdents()) {
+    private FsMQManager fsMQTransport = null;
-            transport.identities.addAll(ti.getAllTransportIdents());
+    private EncryptingSenderThread sender = null;
+    private DecryptingReceiverThread receiver = null;
+    private FsMQIdentity myself = null;
+    private List<TestIdentity> peers = null;
+
+    // Message queues
+    private LinkedBlockingQueue<Message> txQueue = new LinkedBlockingQueue<>();
+    private LinkedBlockingQueue<Message> rxQueue = new LinkedBlockingQueue<>();
+
+    public Transport(FsMQIdentity ownIdent, List<TestIdentity> peers) {
+        this.myself = ownIdent;
+        this.peers = peers;
+        this.sender = new EncryptingSenderThread(this, txQueue);
+        this.receiver = new DecryptingReceiverThread(this, rxQueue);
+        this.fsMQTransport = new FsMQManager(ownIdent);
+
+        for (TestIdentity ti : peers) {
+            fsMQTransport.identities.addAll(ti.getAllTransportIdents());
         }
+    }
 
-        return this;
+    public void clearOwnQueue() {
+        fsMQTransport.clearOwnQueue();
+    }
+
+    public boolean canReceiveAsync() {
+        return !rxQueue.isEmpty();
     }
 
-    public void send(String pEpAddress, String msg) {
+    public void sendAsync(Message msg) {
+        txQueue.add(msg);
+    }
+
+    public Message receiveAsyncNonBlocking() {
+        return rxQueue.remove();
+    }
+
+    public void sendRaw(String pEpAddress, String msg) {
         //Find identity for address
-        List<TestIdentity> res = model.getAllIdents().stream().filter(i -> {
+        List<TestIdentity> res = peers.stream().filter(i -> {
             return i.pEpIdent.address.equals(pEpAddress);
         }).collect(Collectors.toList());
 
@@ -78,30 +128,100 @@ public class CTXMultiNode extends AbstractTestContext {
         TestIdentity id = res.get(0);
 
         for (FsMQIdentity tID : id.getAllTransportIdents()) {
-            transport.sendMessage(tID.getAddress(), msg);
+            fsMQTransport.sendMessage(tID.getAddress(), msg);
 //            log("send() to: " + tID.getAddress());
         }
     }
 
-    private TestModel setupModel() {
+    public String receiveRaw() {
-        TestModel ret = new TestModel();
+        FsMQMessage rx = fsMQTransport.receiveMessage(2000);
+        return rx.getMsg();
+    }
 
-        ret.getNode(NodeName.NODE_A1).setRole(Role.ALICE);
+    public void start() {
-        ret.getNode(NodeName.NODE_B1).setRole(Role.BOB);
+        sender.start();
+        receiver.start();
+    }
+}
 
-        ret.getIdent(Role.ALICE).setDefaultPartner(Role.BOB);
-        ret.getIdent(Role.BOB).setDefaultPartner(Role.ALICE);
-//        ret.getIdent(Role.CAROL).setDefaultPartner(Role.ALICE);
 
-        return ret;
+class EncryptingSenderThread extends Thread {
+    private Engine engine = null;
+    private Transport transport = null;
+    private LinkedBlockingQueue<Message> queue;
+
+    public EncryptingSenderThread(Transport transport, LinkedBlockingQueue queue) {
+        this.transport = transport;
+        this.queue = queue;
+    }
+
+    @Override
+    public void run() {
+        engine = new Engine();
+        Message msg;
+        String msgEnc;
+        while (true) {
+            try {
+                msg = queue.take();
+                if (msg.getTo().size() != 1) {
+                    throw new RuntimeException("Sorry, msg.To has to have exactly 1 receiver for now");
+                }
+                String to = msg.getTo().get(0).address;
+                msgEnc = encryptMessage(msg);
+                transport.sendRaw(to, msgEnc);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    public String encryptMessage(Message msg) {
+        String longMessage = msg.getLongmsg();
+
+        Message msgEnc = engine.encrypt_message(msg, null, Message.EncFormat.PEP);
+        String encFormat;
+        String transportMsg;
+        if (msgEnc != null) {
+            encFormat = "CRYPT";
+            transportMsg = msgEnc.encodeMIME();
+        } else {
+            encFormat = "PLAIN";
+            transportMsg = msg.encodeMIME();
+        }
+        log("<- : [" + encFormat + "] - " + longMessage);
+        return transportMsg;
+    }
+}
+
+
+class DecryptingReceiverThread extends Thread {
+    private Engine engine = null;
+    private Transport transport = null;
+    private LinkedBlockingQueue<Message> queue;
+
+    public DecryptingReceiverThread(Transport transport, LinkedBlockingQueue queue) {
+        this.transport = transport;
+        this.queue = queue;
     }
 
-    public Message reveiveMessage() {
+    @Override
-        FsMQMessage rx = null;
+    public void run() {
-        rx = transport.receiveMessage(2000);
+        engine = new Engine();
+        String msg;
+        Message msgDec;
+        while (true) {
+            msg = transport.receiveRaw();
+            msgDec = decryptMessage(msg);
+            try {
+                queue.put(msgDec);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
 
-        // Receive
+    private Message decryptMessage(String msg) {
-        Message msgIn = new Message(rx.getMsg());
+        Message msgIn = new Message(msg);
         Message msgInDec = null;
 
         decrypt_message_Return decRet = engine.decrypt_message(msgIn, null, 0);
@@ -116,31 +236,4 @@ public class CTXMultiNode extends AbstractTestContext {
         log("-> : [" + encFormat + "] - " + msgInDec.getLongmsg());
         return msgInDec;
     }
-
+}
-    public void sendMessage(Identity toIdent, String longMessage) {
-        // Reply
-        Message reply = new Message();
-        Vector<Identity> to = new Vector<>();
-        to.add(toIdent);
-        reply.setTo(to);
-        reply.setFrom(myself.pEpIdent);
-        reply.setShortmsg("Reply");
-        reply.setLongmsg(longMessage);
-        reply.setDir(Message.Direction.Outgoing);
-
-        Message replyEnc = engine.encrypt_message(reply, null, Message.EncFormat.PEP);
-
-        String encFormat;
-        longMessage = reply.getLongmsg();
-        String transportMsg;
-        if (replyEnc != null) {
-            encFormat = "CRYPT";
-            transportMsg = replyEnc.encodeMIME();
-        } else {
-            encFormat = "PLAIN";
-            transportMsg = reply.encodeMIME();
-        }
-        log("<- : [" + encFormat + "] - " + longMessage);
-        send(toIdent.address, transportMsg);
-    }
-}

test/java/foundation/pEp/jniadapter/test/jni153/TestAlice.java

@@ -18,19 +18,21 @@ class TestAlice {
         CTXMultiNode JNI153Ctx = new CTXMultiNode(NodeName.NODE_A1);
 
         new TestUnit<CTXMultiNode>("test", JNI153Ctx, ctx -> {
+            ctx.transport.clearOwnQueue();
             ctx.myself.pEpIdent = ctx.engine.myself(ctx.myself.pEpIdent);
             log(AdapterTestUtils.identityToString(ctx.myself.pEpIdent, true));
-            ctx.transport.clearOwnQueue();
+
+            ctx.transport.start();
+
+            ctx.transport.sendAsync(AdapterTestUtils.newOutMessage(ctx.myself.pEpIdent, ctx.partner.pEpIdent, "UNIQUE_" + String.valueOf(0)));
             int counter = 0;
+
             while (true) {
-                Message src = AdapterTestUtils.makeNewTestMessage(ctx.myself.pEpIdent, ctx.partner.pEpIdent, Message.Direction.Outgoing);
+                while (ctx.transport.canReceiveAsync()) {
-                src.setLongmsg("UNIQUE_" + String.valueOf(counter));
+                    ctx.transport.sendAsync(AdapterTestUtils.newOutMessage(ctx.myself.pEpIdent, ctx.partner.pEpIdent, "UNIQUE_" + String.valueOf(counter)));
-                ctx.sendMessage(ctx.partner.pEpIdent, src.getLongmsg());
+                    Message msg = ctx.transport.receiveAsyncNonBlocking();
-                ctx.reveiveMessage();
+                    counter++;
-
+                }
-                counter++;
-                TestUtils.sleep(3000);
-//                TestUtils.readKey();
             }
         });
 

test/java/foundation/pEp/jniadapter/test/jni153/TestBob.java

@@ -1,5 +1,6 @@
 package foundation.pEp.jniadapter.test.jni153;
 
+import foundation.pEp.jniadapter.Engine;
 import foundation.pEp.jniadapter.Message;
 import foundation.pEp.jniadapter.test.utils.AdapterTestUtils;
 import foundation.pEp.jniadapter.test.utils.model.NodeName;
@@ -7,6 +8,9 @@ import foundation.pEp.pitytest.TestSuite;
 import foundation.pEp.pitytest.TestUnit;
 import foundation.pEp.pitytest.utils.TestUtils;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import static foundation.pEp.pitytest.TestLogger.log;
 
 class TestBob {
@@ -17,24 +21,31 @@ class TestBob {
         CTXMultiNode JNI153Ctx = new CTXMultiNode(NodeName.NODE_B1);
 
         new TestUnit<CTXMultiNode>("test", JNI153Ctx, ctx -> {
+            ExecutorService executorService = Executors.newFixedThreadPool(10);
+            ctx.transport.clearOwnQueue();
             ctx.myself.pEpIdent = ctx.engine.myself(ctx.myself.pEpIdent);
             log(AdapterTestUtils.identityToString(ctx.myself.pEpIdent, true));
-            ctx.transport.clearOwnQueue();
-            int counter = 0;
 
+            ctx.transport.start();
+
+            int counter = 0;
             while (true) {
-                TestUtils.sleep(3000);
-                Message msgRx = ctx.reveiveMessage();
 
-                //Mistrust
+                while (ctx.transport.canReceiveAsync()) {
-                if (counter == 1) {
+                    Message msgRx = ctx.transport.receiveAsyncNonBlocking();
-                    log("Mistrusting");
+                    log(AdapterTestUtils.identityToString(msgRx.getFrom(),true));
-                    ctx.engine.keyMistrusted(msgRx.getFrom());
+                    ctx.transport.sendAsync(AdapterTestUtils.newOutMessage(ctx.myself.pEpIdent, msgRx.getFrom(), msgRx.getLongmsg() + " - ACK"));
-                }
+                    counter++;
 
-                ctx.sendMessage(msgRx.getFrom(), msgRx.getLongmsg() + " - ACK");
+                    if (counter % 4 == 0) {
+                        executorService.submit(() -> {
+                            Engine eng = new Engine();
+                            log("Mistrusting");
+                            eng.keyMistrusted(msgRx.getFrom());
 
-                counter++;
+                        });
+                    }
+                }
             }
         });
 

test/java/foundation/pEp/jniadapter/test/utils/AdapterTestUtils.java

@@ -338,6 +338,30 @@ public class AdapterTestUtils {
         return msg;
     }
 
+    public static Message newOutMessage(Identity from, Identity to, String longMessage) {
+        Message ret = new Message();
+        Vector<Identity> toList = new Vector<>();
+        toList.add(to);
+        ret.setTo(toList);
+        ret.setFrom(from);
+//        ret.setShortmsg();
+        ret.setLongmsg(longMessage);
+        ret.setDir(Message.Direction.Outgoing);
+        return ret;
+    }
+
+//    public static Message newInMessage(Identity from, Identity to, String longMessage) {
+//        Message ret = new Message();
+//        Vector<Identity> toList = new Vector<>();
+//        toList.add(to);
+//        ret.setTo(toList);
+//        ret.setFrom(from);
+////        ret.setShortmsg();
+//        ret.setLongmsg(longMessage);
+//        ret.setDir(Message.Direction.Outgoing);
+//        return ret;
+//    }
+
     public static Message generateSourceMessage(Identity from, Identity to, long id, long longMsgLen) {
         Message msg = new Message();
         Vector<Identity> vID = new Vector<Identity>();