iotjava/iotrmi/Java/IoTRMIComm.java

   1 package iotrmi.Java;
   2
   3 import java.io.IOException;
   4 import java.nio.ByteBuffer;
   5 import java.util.Arrays;
   6 import java.util.ArrayList;
   7 import java.util.HashMap;
   8 import java.util.List;
   9 import java.util.Map;
  10 import java.util.Set;
  11 import java.lang.reflect.*;
  12
  13 import java.util.concurrent.*;
  14 import java.util.concurrent.atomic.AtomicBoolean;
  15
  16
  17 /** Abstract class IoTRMIComm is a class that combines IoTRMIObject and IoTRMICall
  18  *  <p>
  19  *  We will arbitrate packets into 2 queues and wake up the right threads/callers.
  20  *  We separate traffics one-directionally.
  21  *
  22  * @author      Rahmadi Trimananda <rtrimana @ uci.edu>
  23  * @version     1.0
  24  * @since       2017-01-27
  25  */
  26 public abstract class IoTRMIComm {
  27
  28         /**
  29          * Class Properties
  30          */
  31         protected IoTRMIUtil rmiUtil;
  32         protected byte[] methodBytes;
  33         protected byte[] retValueBytes;
  34         protected ConcurrentLinkedQueue<byte[]> methodQueue;
  35         protected ConcurrentLinkedQueue<byte[]> returnQueue;
  36         protected Map<Integer,AtomicBoolean> mapSkeletonId;
  37         protected Map<String,AtomicBoolean> mapStubId;
  38         protected AtomicBoolean didGetMethodBytes;
  39         protected AtomicBoolean didGetReturnBytes;
  40         protected int objectIdCounter = Integer.MAX_VALUE;
  41
  42         /**
  43          * Constructor (for skeleton)
  44          */
  45         public IoTRMIComm() throws
  46                 ClassNotFoundException, InstantiationException,
  47                         IllegalAccessException, IOException {
  48
  49                 rmiUtil = new IoTRMIUtil();
  50                 methodBytes = null;
  51                 retValueBytes = null;
  52                 methodQueue = new ConcurrentLinkedQueue<byte[]>();
  53                 returnQueue = new ConcurrentLinkedQueue<byte[]>();
  54                 mapSkeletonId = new HashMap<Integer,AtomicBoolean>();
  55                 mapStubId = new HashMap<String,AtomicBoolean>();
  56                 didGetMethodBytes = new AtomicBoolean(false);
  57                 didGetReturnBytes = new AtomicBoolean(false);
  58                 wakeUpThreadOnMethodCall();
  59                 wakeUpThreadOnReturnValue();
  60         }
  61
  62
  63         /**
  64          * wakeUpThreadOnMethodCall() wakes up the correct thread when receiving method call
  65          */
  66         private void wakeUpThreadOnMethodCall() {
  67
  68                 Thread thread = new Thread() {
  69                         public void run() {
  70                                 while(true) {
  71                                         // Take the current method from the queue and wake up the correct thread
  72                                         methodBytes = methodQueue.poll();
  73                                         if (methodBytes != null) {      // If there is method bytes
  74                                                 int currObjId = getObjectId(methodBytes);
  75                                                 AtomicBoolean methRecv = mapSkeletonId.get(currObjId);
  76                                                 didGetMethodBytes.set(false);
  77                                                 while(!methRecv.compareAndSet(false, true));
  78                                                 while(!didGetMethodBytes.get());        // While skeleton is still processing
  79                                         }
  80                                 }
  81                         }
  82                 };
  83                 thread.start();
  84         }
  85
  86
  87         /**
  88          * wakeUpThreadOnReturnValue() wakes up the correct thread when receiving return value
  89          */
  90         private void wakeUpThreadOnReturnValue() {
  91
  92                 Thread thread = new Thread() {
  93                         public void run() {
  94                                 while(true) {
  95                                         // Take the current method from the queue and wake up the correct thread
  96                                         retValueBytes = returnQueue.poll();
  97                                         if (retValueBytes != null) {    // If there is method bytes
  98                                                 System.out.println("retValBytes in wake up thread: " + Arrays.toString(retValueBytes));
  99                                                 int objectId = getObjectId(retValueBytes);
 100                                                 int methodId = getMethodId(retValueBytes);
 101                                                 String strKey = objectId + "-" + methodId;
 102                                                 AtomicBoolean retRecv = mapStubId.get(strKey);
 103                                                 //System.out.println("boolean status: " + retRecv + " with key: " + strKey);
 104                                                 didGetReturnBytes.set(false);
 105                                                 while(!retRecv.compareAndSet(false, true));
 106                                                 //System.out.println("boolean status: " + retRecv + " - map has: " + mapStubId.size());
 107                                                 while(!didGetReturnBytes.get());        // While skeleton is still processing
 108                                         }
 109                                 }
 110                         }
 111                 };
 112                 thread.start();
 113         }
 114
 115
 116         /**
 117          * registerSkeleton() registers the skeleton to be woken up
 118          */
 119         public synchronized void registerSkeleton(int objectId, AtomicBoolean methodReceived) {
 120
 121                 mapSkeletonId.put(objectId, methodReceived);
 122         }
 123
 124
 125         /**
 126          * registerStub() registers the skeleton to be woken up
 127          */
 128         public synchronized void registerStub(int objectId, int methodId, AtomicBoolean retValueReceived) {
 129
 130                 String strKey = objectId + "-" + methodId;
 131                 //System.out.println("Key exist? " + mapStubId.containsKey(strKey));
 132                 mapStubId.put(strKey, retValueReceived);
 133                 //System.out.println("\n\nAdding keyBytes: " + strKey + " now size: " + mapStubId.size() + "\n\n");
 134         }
 135
 136
 137         /**
 138          * getObjectIdCounter() gets object Id counter
 139          */
 140         public int getObjectIdCounter() {
 141
 142                 return objectIdCounter;
 143         }
 144
 145
 146         /**
 147          * setObjectIdCounter() sets object Id counter
 148          */
 149         public void setObjectIdCounter(int objIdCounter) {
 150
 151                 objectIdCounter = objIdCounter;
 152         }
 153
 154
 155         /**
 156          * decrementObjectIdCounter() gets object Id counter
 157          */
 158         public void decrementObjectIdCounter() {
 159
 160                 objectIdCounter--;
 161         }
 162
 163
 164         /**
 165          * setGetMethodBytes() set didGetMethodBytes to true after getting the bytes
 166          */
 167         public boolean setGetMethodBytes() {
 168
 169                 return didGetMethodBytes.compareAndSet(false, true);
 170         }
 171
 172
 173         /**
 174          * setGetReturnBytes() set didGetReturnBytes if there is a new return value already
 175          */
 176         public synchronized boolean setGetReturnBytes() {
 177
 178                 return didGetReturnBytes.compareAndSet(false, true);
 179         }
 180
 181
 182         /**
 183          * getMethodBytes() get the method in bytes
 184          */
 185         public byte[] getMethodBytes() throws IOException {
 186
 187                 // Just return the methodBytes content
 188                 return methodBytes;
 189         }
 190
 191
 192         /**
 193          * static version of getObjectId()
 194          */
 195         public static int getObjectId(byte[] packetBytes) {
 196
 197                 // Get object Id bytes
 198                 byte[] objectIdBytes = new byte[IoTRMIUtil.OBJECT_ID_LEN];
 199                 System.arraycopy(packetBytes, 0, objectIdBytes, 0, IoTRMIUtil.OBJECT_ID_LEN);
 200                 // Get object Id
 201                 int objectId = IoTRMIUtil.byteArrayToInt(objectIdBytes);
 202                 return objectId;
 203         }
 204
 205
 206         /**
 207          * static version of getMethodId()
 208          */
 209         public static int getMethodId(byte[] packetBytes) {
 210
 211                 // Get method Id bytes
 212                 byte[] methodIdBytes = new byte[IoTRMIUtil.METHOD_ID_LEN];
 213                 // Method Id is positioned after object Id in the byte array
 214                 System.arraycopy(packetBytes, IoTRMIUtil.OBJECT_ID_LEN, methodIdBytes, 0, IoTRMIUtil.METHOD_ID_LEN);
 215                 // Get method Id
 216                 int methodId = IoTRMIUtil.byteArrayToInt(methodIdBytes);
 217                 // Get method Id
 218                 return methodId;
 219         }
 220
 221
 222         /**
 223          * static version of getPacketType() - either method or return value (position is after object Id and method Id)
 224          */
 225         public static int getPacketType(byte[] packetBytes) {
 226
 227                 // Get packet type bytes
 228                 byte[] packetTypeBytes = new byte[IoTRMIUtil.PACKET_TYPE_LEN];
 229                 int offset = IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN;
 230                 System.arraycopy(packetBytes, offset, packetTypeBytes, 0, IoTRMIUtil.PACKET_TYPE_LEN);
 231                 // Get packet type (for now we assume 1 as method and -1 as return value
 232                 int packetType = IoTRMIUtil.byteArrayToInt(packetTypeBytes);
 233                 return packetType;
 234         }
 235
 236
 237         /**
 238          * getMethodParams() gets method params based on byte array received
 239          * <p>
 240          * Basically this is the format of a method in bytes:
 241          * 1) 32-bit value of object ID
 242          * 2) 32-bit value of method ID
 243          * 3) m parameters with n-bit value each (m x n-bit)
 244          * For the parameters that don't have definite length,
 245          * we need to extract the length from a preceding 32-bit
 246          * field in front of it.
 247          *
 248          * For primitive objects:
 249          * | 32-bit object ID | 32-bit method ID | m-bit actual data (fixed length)  | ...
 250          *
 251          * For string, arrays, and non-primitive objects:
 252          * | 32-bit object ID | 32-bit method ID | 32-bit length | n-bit actual data | ...
 253          *
 254          */
 255         public Object[] getMethodParams(Class<?>[] arrCls, Class<?>[] arrGenValCls, byte[] methodBytes) {
 256
 257                 // Byte scanning position
 258                 int pos = IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN + IoTRMIUtil.PACKET_TYPE_LEN;
 259                 Object[] paramObj = new Object[arrCls.length];
 260                 for (int i=0; i < arrCls.length; i++) {
 261
 262                         String paramType = arrCls[i].getSimpleName();
 263                         int paramSize = rmiUtil.getTypeSize(paramType);
 264                         // Get the 32-bit field in the byte array to get the actual
 265                         //              length (this is a param with indefinite length)
 266                         if (paramSize == -1) {
 267                                 byte[] bytPrmLen = new byte[IoTRMIUtil.PARAM_LEN];
 268                                 System.arraycopy(methodBytes, pos, bytPrmLen, 0, IoTRMIUtil.PARAM_LEN);
 269                                 pos = pos + IoTRMIUtil.PARAM_LEN;
 270                                 paramSize = IoTRMIUtil.byteArrayToInt(bytPrmLen);
 271                         }
 272                         byte[] paramBytes = new byte[paramSize];
 273                         System.arraycopy(methodBytes, pos, paramBytes, 0, paramSize);
 274                         pos = pos + paramSize;
 275                         paramObj[i] = IoTRMIUtil.getParamObject(arrCls[i], arrGenValCls[i], paramBytes);
 276                 }
 277
 278                 return paramObj;
 279         }
 280
 281
 282         /**
 283          * sendReturnObj() abstract version
 284          */
 285         public abstract void sendReturnObj(Object retObj, byte[] methodBytes);
 286
 287
 288         /**
 289          * sendReturnObj() abstract version
 290          */
 291         public abstract void sendReturnObj(Class<?>[] retCls, Object[] retObj, byte[] methodBytes);
 292
 293
 294         /**
 295          * returnToBytes() takes array of objects and generates bytes
 296          */
 297         public byte[] returnToBytes(Class<?>[] retCls, Object[] retObj) {
 298
 299                 // Get byte arrays and calculate method bytes length
 300                 int numbRet = retObj.length;
 301                 int retLen = 0;
 302                 byte[][] objBytesArr = new byte[numbRet][];
 303                 for (int i = 0; i < numbRet; i++) {
 304                         // Get byte arrays for the objects
 305                         objBytesArr[i] = IoTRMIUtil.getObjectBytes(retObj[i]);
 306                         String clsName = retCls[i].getSimpleName();
 307                         int retObjLen = rmiUtil.getTypeSize(clsName);
 308                         if (retObjLen == -1) {          // indefinite length - store the length first
 309                                 retLen = retLen + IoTRMIUtil.RETURN_LEN;
 310                         }
 311                         retLen = retLen + objBytesArr[i].length;
 312                 }
 313                 // Construct return in byte array
 314                 byte[] retBytes = new byte[retLen];
 315                 int pos = 0;
 316                 // Iteration for copying bytes
 317                 for (int i = 0; i < numbRet; i++) {
 318
 319                         String clsName = retCls[i].getSimpleName();
 320                         int retObjLen = rmiUtil.getTypeSize(clsName);
 321                         if (retObjLen == -1) {          // indefinite length
 322                                 retObjLen = objBytesArr[i].length;
 323                                 byte[] retLenBytes = IoTRMIUtil.intToByteArray(retObjLen);
 324                                 System.arraycopy(retLenBytes, 0, retBytes, pos, IoTRMIUtil.RETURN_LEN);
 325                                 pos = pos + IoTRMIUtil.RETURN_LEN;
 326                         }
 327                         System.arraycopy(objBytesArr[i], 0, retBytes, pos, retObjLen);
 328                         pos = pos + retObjLen;
 329                 }
 330
 331                 return retBytes;
 332         }
 333
 334
 335         /**
 336          * remoteCall() abstract version
 337          */
 338         public abstract void remoteCall(int objectId, int methodId, Class<?>[] paramCls, Object[] paramObj);
 339
 340
 341         /**
 342          * methodToBytes() returns byte representation of a method
 343          */
 344         public byte[] methodToBytes(int objectId, int methId, Class<?>[] paramCls, Object[] paramObj) {
 345
 346                 // Initialized to the length of method ID
 347                 int methodLen = IoTRMIUtil.OBJECT_ID_LEN;
 348                 byte[] objId = IoTRMIUtil.intToByteArray(objectId);
 349                 // Get method ID in bytes
 350                 byte[] methodId = IoTRMIUtil.intToByteArray(methId);
 351                 // Get byte arrays and calculate method bytes length
 352                 int numbParam = paramObj.length;
 353                 methodLen = methodLen + IoTRMIUtil.METHOD_ID_LEN;
 354                 methodLen = methodLen + IoTRMIUtil.PACKET_TYPE_LEN;
 355                 byte[][] objBytesArr = new byte[numbParam][];
 356                 for (int i = 0; i < numbParam; i++) {
 357                         // Get byte arrays for the objects
 358                         objBytesArr[i] = IoTRMIUtil.getObjectBytes(paramObj[i]);
 359                         String clsName = paramCls[i].getSimpleName();
 360                         int paramLen = rmiUtil.getTypeSize(clsName);
 361                         if (paramLen == -1) {           // indefinite length - store the length first
 362                                 methodLen = methodLen + IoTRMIUtil.PARAM_LEN;
 363                         }
 364                         methodLen = methodLen + objBytesArr[i].length;
 365                 }
 366                 // Construct method in byte array
 367                 byte[] method = new byte[methodLen];
 368                 int pos = 0;
 369                 System.arraycopy(objId, 0, method, 0, IoTRMIUtil.METHOD_ID_LEN);
 370                 pos = pos + IoTRMIUtil.OBJECT_ID_LEN;
 371                 System.arraycopy(methodId, 0, method, pos, IoTRMIUtil.METHOD_ID_LEN);
 372                 pos = pos + IoTRMIUtil.METHOD_ID_LEN;
 373                 int packetType = IoTRMIUtil.METHOD_TYPE;        // This is a method
 374                 byte[] packetTypeBytes = IoTRMIUtil.intToByteArray(packetType);
 375                 System.arraycopy(packetTypeBytes, 0, method, pos, IoTRMIUtil.PACKET_TYPE_LEN);
 376                 pos = pos + IoTRMIUtil.PACKET_TYPE_LEN;
 377                 // Second iteration for copying bytes
 378                 for (int i = 0; i < numbParam; i++) {
 379
 380                         String clsName = paramCls[i].getSimpleName();
 381                         int paramLen = rmiUtil.getTypeSize(clsName);
 382                         if (paramLen == -1) {           // indefinite length
 383                                 paramLen = objBytesArr[i].length;
 384                                 byte[] paramLenBytes = IoTRMIUtil.intToByteArray(paramLen);
 385                                 System.arraycopy(paramLenBytes, 0, method, pos, IoTRMIUtil.PARAM_LEN);
 386                                 pos = pos + IoTRMIUtil.PARAM_LEN;
 387                         }
 388                         System.arraycopy(objBytesArr[i], 0, method, pos, paramLen);
 389                         pos = pos + paramLen;
 390                 }
 391
 392                 return method;
 393         }
 394
 395
 396         /**
 397          * getReturnValue() returns return value object
 398          */
 399         public Object getReturnValue(Class<?> retType, Class<?> retGenTypeVal) {
 400
 401                 // Receive return value and return it to caller
 402                 // Now just strip off the object ID and method ID
 403                 int headerLen = IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN + IoTRMIUtil.PACKET_TYPE_LEN;
 404                 int valByteLen = retValueBytes.length - headerLen;
 405                 byte[] retValBytes = new byte[valByteLen];
 406                 // Method Id is positioned after object Id in the byte array
 407                 System.arraycopy(retValueBytes, headerLen, retValBytes, 0, valByteLen);
 408                 Object retObj = IoTRMIUtil.getParamObject(retType, retGenTypeVal, retValBytes);
 409                 // This means the right object and method have gotten the return value, so we set this back to false
 410                 return retObj;
 411         }
 412
 413
 414         /**
 415          * getStructObjects() calls a method remotely by passing in parameters and getting a return Object
 416          */
 417         public Object[] getStructObjects(Class<?>[] retType, Class<?>[] retGenTypeVal) {
 418
 419                 // Receive return value and return it to caller
 420                 // Now just strip off the object ID and method ID
 421                 int headerLen = IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN + IoTRMIUtil.PACKET_TYPE_LEN;
 422                 int valByteLen = retValueBytes.length - headerLen;
 423                 byte[] retValBytes = new byte[valByteLen];
 424                 // Method Id is positioned after object Id in the byte array
 425                 System.arraycopy(retValueBytes, headerLen, retValBytes, 0, valByteLen);
 426                 Object[] retObj = getReturnObjects(retValBytes, retType, retGenTypeVal);
 427
 428                 return retObj;
 429         }
 430
 431
 432         /**
 433          * remoteCall() calls a method remotely by passing in parameters and getting a return Object
 434          */
 435         public Object[] getReturnObjects(byte[] retBytes, Class<?>[] arrCls, Class<?>[] arrGenValCls) {
 436
 437                 // Byte scanning position
 438                 int pos = 0;
 439                 Object[] retObj = new Object[arrCls.length];
 440                 for (int i=0; i < arrCls.length; i++) {
 441
 442                         String retType = arrCls[i].getSimpleName();
 443                         int retSize = rmiUtil.getTypeSize(retType);
 444                         // Get the 32-bit field in the byte array to get the actual
 445                         //              length (this is a param with indefinite length)
 446                         if (retSize == -1) {
 447                                 byte[] bytRetLen = new byte[IoTRMIUtil.RETURN_LEN];
 448                                 System.arraycopy(retBytes, pos, bytRetLen, 0, IoTRMIUtil.RETURN_LEN);
 449                                 pos = pos + IoTRMIUtil.RETURN_LEN;
 450                                 retSize = IoTRMIUtil.byteArrayToInt(bytRetLen);
 451                         }
 452                         byte[] retObjBytes = new byte[retSize];
 453                         System.arraycopy(retBytes, pos, retObjBytes, 0, retSize);
 454                         pos = pos + retSize;
 455                         retObj[i] = IoTRMIUtil.getParamObject(arrCls[i], arrGenValCls[i], retObjBytes);
 456                 }
 457
 458                 return retObj;
 459         }
 460
 461 }