pEp.foundation
/
pEpJNIAdapter
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

FsMQManager / FsMQIdentity

JNI-96
heck 5 years ago
parent
05bed8968b
commit
57824f2563
3 changed files with 194 additions and 65 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 21
      test/java/foundation/pEp/jniadapter/test/utils/transport/fsmqmanager/FsMQIdentity.java
  2. 206
      test/java/foundation/pEp/jniadapter/test/utils/transport/fsmqmanager/FsMQManager.java
  3. 32
      test/java/foundation/pEp/jniadapter/test/utils/transport/fsmqmanager/test/regression/TestMain.java

21
test/java/foundation/pEp/jniadapter/test/utils/transport/fsmqmanager/FsMQIdentity.java
View File

@ -0,0 +1,21 @@
package foundation.pEp.jniadapter.test.utils.transport.fsmqmanager;
class FsMQIdentity implements java.io.Serializable {
private String address = null;
private String qDir = null;
FsMQIdentity(String address, String qDir) {
this.address = address;
this.qDir = qDir;
}
public String getAddress() {
return address;
}
public String getqDir() {
return qDir;
}
}

206
test/java/foundation/pEp/jniadapter/test/utils/transport/fsmqmanager/FsMQManager.java
View File

@ -1,76 +1,204 @@
package foundation.pEp.jniadapter.test.utils.transport.fsmqmanager; package foundation.pEp.jniadapter.test.utils.transport.fsmqmanager;
import foundation.pEp.jniadapter.Identity;
import foundation.pEp.jniadapter.test.framework.TestUtils; import foundation.pEp.jniadapter.test.framework.TestUtils;
import foundation.pEp.jniadapter.test.utils.transport.fsmsgqueue.FsMsgQueue; import foundation.pEp.jniadapter.test.utils.transport.fsmsgqueue.FsMsgQueue;
import java.util.HashMap; import java.io.*;
import java.util.Map; import java.util.*;
import java.util.stream.Collectors;
import static foundation.pEp.jniadapter.test.framework.TestLogger.log; import static foundation.pEp.jniadapter.test.framework.TestLogger.log;
public class FsMQManager { public class FsMQManager {
private String ownAddress; private FsMQIdentity self = null;
private FsMsgQueue ownQueue; private List<FsMQIdentity> identities = new ArrayList<>();
private Map<String, FsMsgQueue> peerQueues = new HashMap<>(); private Map<String, FsMsgQueue> identityAddressQueues = new HashMap<String, FsMsgQueue>();
private static String SIGNALONLINEMSG = "SIGONLINE"; private static String SYNMSG = "SYN";
private static String SYNACKMSG = "SYNACK";
private static String ACKMSG = "ACK";
public FsMQManager(String ownAddr, String ownQueueDir) { public FsMQManager(String ownAddr, String ownQueueDir) {
ownAddress = ownAddr; self = new FsMQIdentity(ownAddr, ownQueueDir);
ownQueue = new FsMsgQueue(ownQueueDir); addOrUpdateIdentity(self);
} }
public void addPeer(String address, String queueDir) { // Identity address must be unique
FsMsgQueue q = new FsMsgQueue(queueDir); public void addOrUpdateIdentity(FsMQIdentity ident) {
peerQueues.put(address, q); try {
getIdentityForAddress(ident.getAddress());
} catch (UnknownIdentityException e) {
// Good, add new ident
addIdent(ident);
return;
}
// Ok, update ident
removeIdent(ident);
addIdent(ident);
return;
} }
public void sendMsgToPeer(String address, String msg) throws UnknownPeerException { public void sendMsgToIdentity(FsMQIdentity ident, String msg) throws UnknownIdentityException, IOException {
getQueueForPeer(address).add(msg); FsMQMessage mqMsg = new FsMQMessage(self, msg);
String serializedStr = mqMsg.serialize();
getQueueForIdentity(ident).add(serializedStr);
} }
public void waitForPeerOnline(String address) { public void clearOwnQueue() {
String msg = ""; getQueueForIdentity(self).clear();
while (msg != "startup from " + address) { }
log("Waiting for " + address);
msg = waitForMsg(); public String waitForMsg() throws UnknownIdentityException, IOException, ClassNotFoundException {
} String ret = null;
FsMsgQueue onwQueue = getQueueForIdentity(self);
FsMQMessage mqMsg = null;
do {
while (onwQueue.isEmpty()) {
TestUtils.sleep(100);
}
String serializedMsg = onwQueue.remove();
mqMsg = FsMQMessage.deserialize(serializedMsg);
} while (doHandshakeProtocol(mqMsg));
ret = mqMsg.msg;
return ret;
}
// undefined behaviour if already existing
private void addIdent(FsMQIdentity ident) {
identities.add(ident);
createQueueForIdent(ident);
}
// undefined behaviour if already existing
// Removes the identity from identities and identityQueues by address
private void removeIdent(FsMQIdentity ident) {
identities.removeIf(i -> i.getAddress().equals(ident.getAddress()));
identityAddressQueues.entrySet().removeIf(iq -> iq.getKey().equals(ident.getAddress()));
} }
public void clearOwnQueue() { private void createQueueForIdent(FsMQIdentity ident) {
ownQueue.clear(); FsMsgQueue q = new FsMsgQueue(ident.getqDir());
identityAddressQueues.put(ident.getAddress(), q);
} }
public String waitForMsg() { private FsMsgQueue getQueueForIdentity(FsMQIdentity ident) throws UnknownIdentityException {
while (ownQueue.isEmpty()) { FsMsgQueue ret = null;
TestUtils.sleep(100); ret = identityAddressQueues.get(ident.getAddress());
if (ret != null) {
throw new UnknownIdentityException("Unknown identity address: " + ident.getAddress());
} }
return ownQueue.remove(); return ret;
} }
public void sendSigOnlineToPeer(String address) { private FsMQIdentity getIdentityForAddress(String address) throws UnknownIdentityException, IllegalStateException {
String msg = SIGNALONLINEMSG + " " + ownAddress; FsMQIdentity ret = null;
log("Sending SIGONLINE to: " + address); List<FsMQIdentity> matches = identities.stream().filter(i -> i.getAddress().equals(address)).collect(Collectors.toList());
sendMsgToPeer(address, msg); if (matches.size() <= 0) {
throw new UnknownIdentityException("No identity with address:" + address);
}
if (matches.size() > 1) {
throw new IllegalStateException("Identity address not unique: " + address);
}
ret = matches.get(0);
return ret;
} }
public void broadcastSigOnline() {
for (String k : peerQueues.keySet()) { // public void handshake(FsMQIdentity ident) {
sendSigOnlineToPeer(k); // String msg = "";
// sendSYN(ident);
// while (msg != SYNACKMSG + " " + ident.getAddress()) {
// log("Waiting for SYNACK from " + ident.getAddress());
// msg = waitForMsg();
// }
// sendACK(ident);
// }
private boolean doHandshakeProtocol(FsMQMessage msg) {
boolean ret = false;
//
// if(msg.matches(SYNMSG)) {
//
// }
// if(msg.matches(SYNACK)) {
//
// }
return ret;
}
// public void sendSYN(FsMQIdentity ident) {
// String msg = SYNMSG + " " + self.getAddress();
// log("Sending SYN to: " + ident.getAddress());
// sendMsgToIdentity(ident, msg);
// }
//
// public void sendACK(FsMQIdentity ident) {
// String msg = ACKMSG + " " + self.getAddress();
// log("Sending ACK to: " + ident.getAddress());
// sendMsgToIdentity(ident, msg);
// }
}
class FsMQMessage implements java.io.Serializable {
FsMQIdentity from = null;
FsMQHandshakeHeader header = null;
String msg = null;
FsMQMessage(FsMQIdentity from, String msg) throws IllegalStateException{
if(from == null || msg == null) {
throw new IllegalStateException("from and msg cant be null");
} }
this.from = from;
this.msg = msg;
}
public String getMsg() {
return msg;
}
public FsMQHandshakeHeader getHeader() {
return header;
} }
private FsMsgQueue getQueueForPeer(String address) throws UnknownPeerException { public void setHeader(FsMQHandshakeHeader header) {
FsMsgQueue ret = peerQueues.get(address); this.header = header;
if (ret == null) { }
throw new UnknownPeerException("No peer with address:" + address);
public static FsMQMessage deserialize(String serializedMsg) throws IOException, ClassNotFoundException {
FsMQMessage ret = null;
byte[] data = Base64.getDecoder().decode(serializedMsg);
ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data));
Object obj = ois.readObject();
ois.close();
if(!(obj instanceof FsMQMessage)) {
throw new ClassNotFoundException("Unvalid serialized string");
} else {
ret = (FsMQMessage) obj;
} }
return ret; return ret;
} }
public String serialize() throws IOException {
String ret = null;
ByteArrayOutputStream baos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(baos);
oos.writeObject(this);
oos.close();
ret = Base64.getEncoder().encodeToString(baos.toByteArray());
return ret;
}
class FsMQHandshakeHeader implements java.io.Serializable {
String operation = null;
}
} }
class UnknownPeerException extends RuntimeException { class UnknownIdentityException extends RuntimeException {
UnknownPeerException(String message) { UnknownIdentityException(String message) {
super(message); super(message);
} }
} }

32
test/java/foundation/pEp/jniadapter/test/utils/transport/fsmqmanager/test/regression/TestMain.java
View File

@ -4,24 +4,18 @@ import static foundation.pEp.jniadapter.test.framework.TestLogger.*;
import foundation.pEp.jniadapter.test.utils.transport.fsmqmanager.*; import foundation.pEp.jniadapter.test.utils.transport.fsmqmanager.*;
import foundation.pEp.jniadapter.test.framework.*; import foundation.pEp.jniadapter.test.framework.*;
import java.util.HashMap;
import java.util.Map;
class FsMsgQueueTestContext extends AbstractTestContext { class FsMsgQueueTestContext extends AbstractTestContext {
Map<String, String> peers;
String ownAddress = "Alice"; String ownAddress = "Alice";
String ownQDir = "../resources/fsmsgqueue-test/alice"; String ownQDir = "../resources/fsmsgqueue-test/alice";
String addressBob = "Bob";
String addressCarol = "Carol"; FsMQIdentity bob = null;
FsMQManager qm; FsMQManager qm;
@Override @Override
public void init() throws Throwable { public void init() throws Throwable {
peers = new HashMap<>(); bob = new FsMQIdentity("Bob","../resources/fsmsgqueue-test/bob");
peers.put(addressBob, "../resources/fsmsgqueue-test/bob");
peers.put(addressCarol, "../resources/fsmsgqueue-test/carol");
} }
} }
@ -39,25 +33,11 @@ class TestMain {
ctx.qm.clearOwnQueue(); ctx.qm.clearOwnQueue();
}).add(); }).add();
new TestUnit<FsMsgQueueTestContext>("Add peer bob", testCtx, ctx -> { new TestUnit<FsMsgQueueTestContext>("Add ident bob", testCtx, ctx -> {
for(String k : ctx.peers.keySet()) { log("Adding ident: " + ctx.bob.getAddress());
log("Adding peer: " + k); ctx.qm.addOrUpdateIdentity(ctx.bob);
ctx.qm.addPeer(k, ctx.peers.get(k));
}
}).add();
new TestUnit<FsMsgQueueTestContext>("Broadcast online", testCtx, ctx -> {
ctx.qm.broadcastSigOnline();
}).add(); }).add();
new TestUnit<FsMsgQueueTestContext>("Wait for bob", testCtx, ctx -> {
log("Waiting for Bob to signal online");
ctx.qm.waitForPeerOnline(ctx.addressBob);
log("Bob is online");
}).add();
TestSuite.run(); TestSuite.run();
} }

Write Preview
Loading…
Cancel
Save