import java.util.Set;
import java.lang.reflect.*;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
/** Class IoTRMIObject is a class that stores info of an object.
private IoTRMIUtil rmiUtil;
private IoTSocketServer rmiServer;
private byte[] methodBytes;
- private Lock lock = new ReentrantLock();
+ private ConcurrentLinkedQueue<byte[]> methodQueue;
+ private Map<Integer,AtomicBoolean> mapSkeletonId;
+ private AtomicBoolean didGetMethodBytes;
/**
ClassNotFoundException, InstantiationException,
IllegalAccessException, IOException {
+ didGetMethodBytes = new AtomicBoolean(false);
rmiUtil = new IoTRMIUtil();
methodBytes = null;
+ methodQueue = new ConcurrentLinkedQueue<byte[]>();
+ mapSkeletonId = new HashMap<Integer,AtomicBoolean>();
rmiServer = new IoTSocketServer(_port);
rmiServer.connect();
+ waitForMethod();
+ wakeUpThread();
}
/**
- * getMethodBytes() waits for method transmission in bytes
+ * waitForMethod() starts a thread that waits for method bytes
+ */
+ public void waitForMethod() {
+
+ Thread thread = new Thread() {
+ public void run() {
+ byte[] methBytes = null;
+ while(true) {
+ try {
+ methBytes = rmiServer.receiveBytes(methBytes);
+ if (methBytes != null) {
+ System.out.println("Command not null: " + Arrays.toString(methBytes));
+ methodQueue.offer(methBytes);
+ } else
+ Thread.sleep(100);
+ methBytes = null;
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ throw new Error("IoTRMICall: Error receiving return value bytes!");
+ }
+ }
+ }
+ };
+ thread.start();
+ }
+
+
+ /**
+ * wakeUpThread() wakes up the correct thread
+ */
+ public void wakeUpThread() {
+
+ Thread thread = new Thread() {
+ public void run() {
+ while(true) {
+ // Take the current method from the queue and wake up the correct thread
+ methodBytes = methodQueue.poll();
+ if (methodBytes != null) { // If there is method bytes
+ int currObjId = getObjectId(methodBytes);
+ AtomicBoolean methRecv = mapSkeletonId.get(currObjId);
+ didGetMethodBytes.set(false);
+ while(!methRecv.compareAndSet(false, true));
+ while(!didGetMethodBytes.get()); // While skeleton is still processing
+ }
+ }
+ }
+ };
+ thread.start();
+ }
+
+
+ /**
+ * registerSkeleton() registers the skeleton to be woken up
+ */
+ public synchronized void registerSkeleton(int objectId, AtomicBoolean methodReceived) {
+
+ mapSkeletonId.put(objectId, methodReceived);
+ }
+
+
+ /**
+ * setGetMethodBytes() set didGetMethodBytes to true after getting the bytes
+ */
+ public boolean setGetMethodBytes() {
+
+ return didGetMethodBytes.compareAndSet(false, true);
+ }
+
+
+ /**
+ * getMethodBytes() get the method in bytes
*/
public byte[] getMethodBytes() throws IOException {
- // Receive method info
- //System.out.println("Method RMIObj before: " + Arrays.toString(methodBytes));
- methodBytes = rmiServer.receiveBytes(methodBytes);
- //System.out.println("Method RMIObj after: " + Arrays.toString(methodBytes));
+ // Just return the methodBytes content
return methodBytes;
}
}
- /**
- * setMethodBytes() sets bytes for method
- */
- /*public void setMethodBytes(byte[] _methodBytes) throws IOException {
-
- // Set method bytes
- methodBytes = _methodBytes;
- }*/
-
-
/**
* getMethodId() gets method Id from bytes
*/
/**
- * sendReturnObj() sends back return Object to client
+ * getMethodParams() overloading
+ */
+ public Object[] getMethodParams(Class<?>[] arrCls, Class<?>[] arrGenValCls, byte[] methodBytes) {
+
+ // Byte scanning position
+ int pos = IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN;
+ Object[] paramObj = new Object[arrCls.length];
+ for (int i=0; i < arrCls.length; i++) {
+
+ String paramType = arrCls[i].getSimpleName();
+ int paramSize = rmiUtil.getTypeSize(paramType);
+ // Get the 32-bit field in the byte array to get the actual
+ // length (this is a param with indefinite length)
+ if (paramSize == -1) {
+ byte[] bytPrmLen = new byte[IoTRMIUtil.PARAM_LEN];
+ System.arraycopy(methodBytes, pos, bytPrmLen, 0, IoTRMIUtil.PARAM_LEN);
+ pos = pos + IoTRMIUtil.PARAM_LEN;
+ paramSize = IoTRMIUtil.byteArrayToInt(bytPrmLen);
+ }
+ byte[] paramBytes = new byte[paramSize];
+ System.arraycopy(methodBytes, pos, paramBytes, 0, paramSize);
+ pos = pos + paramSize;
+ paramObj[i] = IoTRMIUtil.getParamObject(arrCls[i], arrGenValCls[i], paramBytes);
+ }
+
+ return paramObj;
+ }
+
+
+ /**
+ * sendReturnObj() overloading
*/
public void sendReturnObj(Object retObj) throws IOException {
+
+ byte[] retObjBytes = IoTRMIUtil.getObjectBytes(retObj);
+ // Send return value together with OBJECT_ID and METHOD_ID for arbitration
+ byte[] retAllBytes = new byte[IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN + retObjBytes.length];
+ // Copy OBJECT_ID and METHOD_ID
+ System.arraycopy(methodBytes, 0, retAllBytes, 0, IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN);
+ // Copy array of bytes (return object)
+ System.arraycopy(retObjBytes, 0, retAllBytes, IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN, retObjBytes.length);
+ rmiServer.sendBytes(retAllBytes);
+ }
+
+
+ /**
+ * sendReturnObj() static version
+ */
+ public void sendReturnObj(Object retObj, byte[] methodBytes) throws IOException {
+
// Send back return value
byte[] retObjBytes = IoTRMIUtil.getObjectBytes(retObj);
- rmiServer.sendBytes(retObjBytes);
+ // Send return value together with OBJECT_ID and METHOD_ID for arbitration
+ byte[] retAllBytes = new byte[IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN + retObjBytes.length];
+ // Copy OBJECT_ID and METHOD_ID
+ System.arraycopy(methodBytes, 0, retAllBytes, 0, IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN);
+ // Copy array of bytes (return object)
+ System.arraycopy(retObjBytes, 0, retAllBytes, IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN, retObjBytes.length);
+ rmiServer.sendBytes(retAllBytes);
}
projects / iot2.git / blobdiff