iotjava/iotrmi/C++/IoTRMIComm.hpp

   1 /** Class IoTRMIComm combines the functionalities
   2  *  of IoTRMICall and IoTRMIObject to create a single
   3  *  communication class with two sockets serving one
   4  *  directional traffic for each.
   5  *
   6  * @author      Rahmadi Trimananda <rtrimana @ uci.edu>
   7  * @version     1.0
   8  * @since       2017-01-28
   9  */
  10 #ifndef _IOTRMICOMM_HPP__
  11 #define _IOTRMICOMM_HPP__
  12
  13 #include <iostream>
  14 #include <string>
  15 #include <atomic>
  16 #include <limits>
  17 #include <thread>
  18 #include <mutex>
  19
  20 #include "IoTSocketServer.hpp"
  21 #include "IoTSocketClient.hpp"
  22 #include "ConcurrentLinkedListQueue.cpp"
  23
  24 using namespace std;
  25
  26 std::atomic<bool> didGetMethodBytes(false);
  27 std::atomic<bool> didGetReturnBytes(false);
  28
  29 mutex regSkelMutex;
  30 mutex regStubMutex;
  31 mutex retValMutex;
  32 mutex remoteCallMutex;
  33 mutex sendReturnObjMutex;
  34
  35 class IoTRMIComm {
  36         public:
  37                 IoTRMIComm();
  38                 ~IoTRMIComm();
  39                 // Public methods
  40                 virtual void            sendReturnObj(void* retObj, string type, char* methodBytes) = 0;
  41                 virtual void            sendReturnObj(void* retObj[], string type[], int numRet, char* methodBytes) = 0;
  42                 int                     returnLength(void* retObj[], string retCls[], int numRet);
  43                 char*                   returnToBytes(void* retObj[], string retCls[], char* retBytes, int numRet);
  44                 char*                   getMethodBytes();
  45                 int                     getMethodLength();
  46                 int                     getObjectIdFromMethod();
  47                 static int              getObjectId(char* packetBytes);
  48                 static int              getMethodId(char* packetBytes);
  49                 static int              getPacketType(char* packetBytes);
  50                 void**                  getMethodParams(string paramCls[], int numParam, void* paramObj[], char* methodBytes);
  51                 void                    registerSkeleton(int objectId, bool* methodReceived);
  52                 void                    registerStub(int objectId, int methodId, bool* retValueReceived);
  53                 int                     getObjectIdCounter();
  54                 void                    setObjectIdCounter(int objIdCounter);
  55                 void                    decrementObjectIdCounter();
  56
  57                 int                     methodLength(string paramCls[], void* paramObj[], int numParam);
  58                 char*                   methodToBytes(int objectId, int methId, string paramCls[], void* paramObj[],
  59                                                 char* method, int numParam);
  60                 virtual void            remoteCall(int objectId, int methodId, string paramCls[],
  61                                                 void* paramObj[], int numParam) = 0;
  62                 void*                   getReturnValue(string retType, void* retObj);
  63
  64                 void**                  getStructObjects(string retType[], int numRet, void* retObj[]);
  65                 void**                  getReturnObjects(char* retBytes, string retCls[], int numRet, void* retObj[]);
  66
  67         protected:
  68                 IoTRMIUtil                      *rmiUtil;
  69                 char*                           methodBytes;
  70                 int                             methodLen;
  71                 char*                           retValueBytes;
  72                 int                             retValueLen;
  73                 ConcurrentLinkedListQueue       methodQueue;
  74                 ConcurrentLinkedListQueue       returnQueue;
  75                 map<int,bool*>                  mapSkeletonId;
  76                 map<string,bool*>               mapStubId;
  77                 int                             objectIdCounter = std::numeric_limits<int>::max();
  78
  79         private:
  80                 // Private methods
  81                 void                            wakeUpThreadOnMethodCall(IoTRMIComm* rmiComm);
  82                 void                            wakeUpThreadOnReturnValue(IoTRMIComm* rmiComm);
  83 };
  84
  85
  86 // Constructor
  87 IoTRMIComm::IoTRMIComm() {
  88
  89         rmiUtil = new IoTRMIUtil();
  90         methodBytes = NULL;
  91         retValueBytes = NULL;
  92         methodLen = 0;
  93         retValueLen = 0;
  94         thread th1 (&IoTRMIComm::wakeUpThreadOnMethodCall, this, this);
  95         th1.detach();
  96         thread th2 (&IoTRMIComm::wakeUpThreadOnReturnValue, this, this);
  97         th2.detach();
  98
  99 }
 100
 101
 102 // Destructor
 103 IoTRMIComm::~IoTRMIComm() {
 104
 105         // Clean up
 106         if (rmiUtil != NULL) {
 107                 delete rmiUtil;
 108                 rmiUtil = NULL;
 109         }
 110 }
 111
 112
 113 void IoTRMIComm::wakeUpThreadOnMethodCall(IoTRMIComm* rmiComm) {
 114
 115         int methLen = 0;
 116         while(true) {
 117                 // Convert back to char*
 118                 char* queueHead = rmiComm->methodQueue.deQAndGetLength(&methLen);
 119                 if (queueHead != NULL) {
 120                         rmiComm->methodBytes = queueHead;
 121                         rmiComm->methodLen = methLen;
 122                         //IoTRMIUtil::printBytes(rmiComm->methodBytes, rmiComm->methodLen, false);
 123                         int currObjId = rmiComm->getObjectId(rmiComm->methodBytes);
 124                         auto search = rmiComm->mapSkeletonId.find(currObjId);
 125                         bool* methRecv = search->second;
 126                         didGetMethodBytes.exchange(false);
 127                         *methRecv = true;
 128                         while(!didGetMethodBytes);
 129                 }
 130         }
 131 }
 132
 133
 134 void IoTRMIComm::wakeUpThreadOnReturnValue(IoTRMIComm* rmiComm) {
 135
 136         int retLen = 0;
 137         while(true) {
 138                 // Convert back to char*
 139                 char* queueHead = rmiComm->returnQueue.deQAndGetLength(&retLen);
 140                 if (queueHead != NULL) {
 141                         rmiComm->retValueBytes = queueHead;
 142                         rmiComm->retValueLen = retLen;
 143                         //IoTRMIUtil::printBytes(rmiComm->retValueBytes, rmiComm->retValueLen, false);
 144                         int objectId = rmiComm->getObjectId(rmiComm->retValueBytes);
 145                         int methodId = rmiComm->getMethodId(rmiComm->retValueBytes);
 146                         string strKey = to_string(objectId) + "-" + to_string(methodId);
 147                         auto search = rmiComm->mapStubId.find(strKey);
 148                         bool* retRecv = search->second;
 149                         didGetReturnBytes.exchange(false);
 150                         *retRecv = true;
 151                         while(!didGetReturnBytes);
 152                 }
 153         }
 154 }
 155
 156
 157 // registerSkeleton() registers the skeleton to be woken up
 158 void IoTRMIComm::registerSkeleton(int objectId, bool* methodReceived) {
 159
 160         lock_guard<mutex> guard(regSkelMutex);
 161         mapSkeletonId.insert(make_pair(objectId, methodReceived));
 162 }
 163
 164
 165 // registerStub() registers the skeleton to be woken up
 166 void IoTRMIComm::registerStub(int objectId, int methodId, bool* retValueReceived) {
 167
 168         lock_guard<mutex> guard(regStubMutex);
 169         string strKey = to_string(objectId) + "-" + to_string(methodId);
 170         mapStubId.insert(make_pair(strKey, retValueReceived));
 171 }
 172
 173
 174 // getObjectIdCounter() gets object Id counter
 175 int     IoTRMIComm::getObjectIdCounter() {
 176
 177         return objectIdCounter;
 178 }
 179
 180
 181 // setObjectIdCounter() sets object Id counter
 182 void IoTRMIComm::setObjectIdCounter(int objIdCounter) {
 183
 184         objectIdCounter = objIdCounter;
 185 }
 186
 187
 188 // decrementObjectIdCounter() gets object Id counter
 189 void IoTRMIComm::decrementObjectIdCounter() {
 190
 191         objectIdCounter--;
 192 }
 193
 194
 195 // Get method bytes from the socket
 196 char* IoTRMIComm::getMethodBytes() {
 197
 198         // Get method bytes
 199         return methodBytes;
 200 }
 201
 202
 203 // Get method length from the socket
 204 int IoTRMIComm::getMethodLength() {
 205
 206         // Get method bytes
 207         return methodLen;
 208 }
 209
 210
 211 // Get object Id from bytes
 212 int IoTRMIComm::getObjectIdFromMethod() {
 213
 214         char objectIdBytes[IoTRMIUtil::OBJECT_ID_LEN];
 215         memcpy(objectIdBytes, methodBytes, IoTRMIUtil::OBJECT_ID_LEN);
 216         // Get method signature
 217         int objectId = 0;
 218         IoTRMIUtil::byteArrayToInt(&objectId, objectIdBytes);
 219
 220         return objectId;
 221 }
 222
 223
 224 // Get object Id from bytes (static version)
 225 int IoTRMIComm::getObjectId(char* packetBytes) {
 226
 227         char objectIdBytes[IoTRMIUtil::OBJECT_ID_LEN];
 228         memcpy(objectIdBytes, packetBytes, IoTRMIUtil::OBJECT_ID_LEN);
 229         // Get method signature
 230         int objectId = 0;
 231         IoTRMIUtil::byteArrayToInt(&objectId, objectIdBytes);
 232
 233         return objectId;
 234 }
 235
 236
 237 // Get methodId from bytes (static version)
 238 int IoTRMIComm::getMethodId(char* packetBytes) {
 239
 240         // Get method Id
 241         char methodIdBytes[IoTRMIUtil::METHOD_ID_LEN];
 242         int offset = IoTRMIUtil::OBJECT_ID_LEN;
 243         memcpy(methodIdBytes, packetBytes + offset, IoTRMIUtil::METHOD_ID_LEN);
 244         // Get method signature
 245         int methodId = 0;
 246         IoTRMIUtil::byteArrayToInt(&methodId, methodIdBytes);
 247
 248         return methodId;
 249 }
 250
 251
 252 // Get methodId from bytes (static version)
 253 int IoTRMIComm::getPacketType(char* packetBytes) {
 254
 255         // Get method Id
 256         char packetTypeBytes[IoTRMIUtil::METHOD_ID_LEN];
 257         int offset = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN;
 258         memcpy(packetTypeBytes, packetBytes + offset, IoTRMIUtil::PACKET_TYPE_LEN);
 259         // Get method signature
 260         int packetType = 0;
 261         IoTRMIUtil::byteArrayToInt(&packetType, packetTypeBytes);
 262
 263         return packetType;
 264 }
 265
 266
 267 // Get method parameters and return an array of parameter objects
 268 //
 269 // For primitive objects:
 270 // | 32-bit method ID | m-bit actual data (fixed length)  |
 271 //
 272 // For string, arrays, and non-primitive objects:
 273 // | 32-bit method ID | 32-bit length | n-bit actual data | ...
 274 void** IoTRMIComm::getMethodParams(string paramCls[], int numParam, void* paramObj[], char* methodBytes) {
 275
 276         // Byte scanning position
 277         int pos = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN + IoTRMIUtil::PACKET_TYPE_LEN;
 278         for (int i = 0; i < numParam; i++) {
 279                 int paramLen = rmiUtil->getTypeSize(paramCls[i]);
 280                 // Get the 32-bit field in the byte array to get the actual
 281                 //              length (this is a param with indefinite length)
 282                 if (paramLen == -1) {
 283                         char bytPrmLen[IoTRMIUtil::PARAM_LEN];
 284                         memcpy(bytPrmLen, methodBytes + pos, IoTRMIUtil::PARAM_LEN);
 285                         pos = pos + IoTRMIUtil::PARAM_LEN;
 286                         int* prmLenPtr = IoTRMIUtil::byteArrayToInt(&paramLen, bytPrmLen);
 287                         paramLen = *prmLenPtr;
 288                 }
 289                 char paramBytes[paramLen];
 290                 memcpy(paramBytes, methodBytes + pos, paramLen);
 291                 pos = pos + paramLen;
 292                 paramObj[i] = IoTRMIUtil::getParamObject(paramObj[i], paramCls[i].c_str(), paramBytes, paramLen);
 293         }
 294
 295         return paramObj;
 296 }
 297
 298
 299 // Find the bytes length of a return object (struct that has more than 1 member)
 300 int     IoTRMIComm::returnLength(void* retObj[], string retCls[], int numRet) {
 301
 302         // Get byte arrays and calculate return bytes length
 303         int returnLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN + IoTRMIUtil::PACKET_TYPE_LEN;
 304         for (int i = 0; i < numRet; i++) {
 305                 // Find the return length
 306                 int retObjLen = rmiUtil->getTypeSize(retCls[i]);
 307                 if (retObjLen == -1) { // Store the length of the field - indefinite length
 308                         retObjLen = rmiUtil->getVarTypeSize(retCls[i], retObj[i]);
 309                         // Some space for return length, i.e. 32 bits for integer
 310                         returnLen = returnLen + IoTRMIUtil::RETURN_LEN;
 311                 }
 312                 // Calculate returnLen
 313                 returnLen = returnLen + retObjLen;
 314         }
 315
 316         return returnLen;
 317 }
 318
 319
 320 // Convert return object (struct members) into bytes
 321 char* IoTRMIComm::returnToBytes(void* retObj[], string retCls[], char* retBytes, int numRet) {
 322
 323         int pos = 0;
 324         // Get byte arrays and calculate return bytes length
 325         for (int i = 0; i < numRet; i++) {
 326                 // Find the return length
 327                 int retObjLen = rmiUtil->getTypeSize(retCls[i]);
 328                 if (retObjLen == -1) { // Store the length of the field - indefinite length
 329                         retObjLen = rmiUtil->getVarTypeSize(retCls[i], retObj[i]);
 330                         // Write the return length
 331                         char retLenBytes[IoTRMIUtil::RETURN_LEN];
 332                         IoTRMIUtil::intToByteArray(retObjLen, retLenBytes);
 333                         memcpy(retBytes + pos, retLenBytes, IoTRMIUtil::RETURN_LEN);
 334                         pos = pos + IoTRMIUtil::RETURN_LEN;
 335                 }
 336                 // Get array of bytes and put it in the array of array of bytes
 337                 char objBytes[retObjLen];
 338                 IoTRMIUtil::getObjectBytes(objBytes, retObj[i], retCls[i].c_str());
 339                 memcpy(retBytes + pos, objBytes, retObjLen);
 340                 pos = pos + retObjLen;
 341         }
 342
 343         return retBytes;
 344 }
 345
 346
 347 // Get return value for single values (non-structs)
 348 void* IoTRMIComm::getReturnValue(string retType, void* retObj) {
 349
 350         // Receive return value and return it to caller
 351         lock_guard<mutex> guard(retValMutex);
 352         // Copy just the actual return value bytes
 353         int headerLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN + IoTRMIUtil::PACKET_TYPE_LEN;
 354         int retActualLen = retValueLen - headerLen;
 355         //char *retActualBytes = new char[retActualLen];
 356         char retActualBytes[retActualLen];
 357         memcpy(retActualBytes, retValueBytes + headerLen, retActualLen);
 358         //IoTRMIUtil::printBytes(retActualBytes, retActualLen, false);
 359         retObj = IoTRMIUtil::getParamObject(retObj, retType.c_str(), retActualBytes, retActualLen);
 360         // Delete received bytes object
 361         delete[] retValueBytes;
 362         //delete[] retActualBytes;
 363
 364         return retObj;
 365 }
 366
 367
 368 // Get a set of return objects (struct)
 369 void** IoTRMIComm::getStructObjects(string retType[], int numRet, void* retObj[]) {
 370
 371         // Critical section that is used by different objects
 372         lock_guard<mutex> guard(retValMutex);
 373         // Copy just the actual return value bytes
 374         int headerLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN + IoTRMIUtil::PACKET_TYPE_LEN;
 375         int retActualLen = retValueLen - headerLen;
 376         char retActualBytes[retActualLen];
 377         memcpy(retActualBytes, retValueBytes + headerLen, retActualLen);
 378         // Return size of array of struct
 379         retObj = getReturnObjects(retActualBytes, retType, numRet, retObj);
 380         // Delete received bytes object
 381         delete[] retValueBytes;
 382
 383         return retObj;
 384 }
 385
 386
 387 // Find the bytes length of a method
 388 int IoTRMIComm::methodLength(string paramCls[], void* paramObj[], int numParam) {
 389
 390         // Get byte arrays and calculate method bytes length
 391         // Start from the object Id + method Id...
 392         int methodLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN + IoTRMIUtil::PACKET_TYPE_LEN;
 393         for (int i = 0; i < numParam; i++) {
 394                 // Find the parameter length
 395                 int paramLen = rmiUtil->getTypeSize(paramCls[i]);
 396                 if (paramLen == -1) { // Store the length of the field - indefinite length
 397                         paramLen = rmiUtil->getVarTypeSize(paramCls[i], paramObj[i]);
 398                         // Some space for param length, i.e. 32 bits for integer
 399                         methodLen = methodLen + IoTRMIUtil::PARAM_LEN;
 400                 }
 401                 // Calculate methodLen
 402                 methodLen = methodLen + paramLen;
 403         }
 404         return methodLen;
 405 }
 406
 407
 408 // Convert method and its parameters into bytes
 409 char* IoTRMIComm::methodToBytes(int objectId, int methId, string paramCls[],
 410                 void* paramObj[], char* method, int numParam) {
 411
 412         // Get object Id in bytes
 413         char objId[IoTRMIUtil::OBJECT_ID_LEN];
 414         IoTRMIUtil::intToByteArray(objectId, objId);
 415         memcpy(method, objId, IoTRMIUtil::OBJECT_ID_LEN);
 416         int pos = IoTRMIUtil::OBJECT_ID_LEN;
 417         // Get method Id in bytes
 418         char methodId[IoTRMIUtil::METHOD_ID_LEN];
 419         IoTRMIUtil::intToByteArray(methId, methodId);
 420         memcpy(method + pos, methodId, IoTRMIUtil::METHOD_ID_LEN);
 421         pos = pos + IoTRMIUtil::METHOD_ID_LEN;
 422         char packetType[IoTRMIUtil::PACKET_TYPE_LEN];
 423         IoTRMIUtil::intToByteArray(IoTRMIUtil::METHOD_TYPE, methodId);
 424         memcpy(method + pos, methodId, IoTRMIUtil::PACKET_TYPE_LEN);
 425         pos = pos + IoTRMIUtil::PACKET_TYPE_LEN;
 426         // Get byte arrays and calculate method bytes length
 427         for (int i = 0; i < numParam; i++) {
 428                 // Find the parameter length
 429                 int paramLen = rmiUtil->getTypeSize(paramCls[i]);
 430                 if (paramLen == -1) { // Store the length of the field - indefinite length
 431                         paramLen = rmiUtil->getVarTypeSize(paramCls[i], paramObj[i]);
 432                         // Write the parameter length
 433                         char prmLenBytes[IoTRMIUtil::PARAM_LEN];
 434                         IoTRMIUtil::intToByteArray(paramLen, prmLenBytes);
 435                         memcpy(method + pos, prmLenBytes, IoTRMIUtil::PARAM_LEN);
 436                         pos = pos + IoTRMIUtil::PARAM_LEN;
 437                 }
 438                 // Get array of bytes and put it in the array of array of bytes
 439                 char objBytes[paramLen];
 440                 IoTRMIUtil::getObjectBytes(objBytes, paramObj[i], paramCls[i].c_str());
 441                 memcpy(method + pos, objBytes, paramLen);
 442                 pos = pos + paramLen;
 443         }
 444
 445         return method;
 446 }
 447
 448
 449 // Get return objects for structs
 450 void** IoTRMIComm::getReturnObjects(char* retBytes, string retCls[], int numRet, void* retObj[]) {
 451
 452         // Byte scanning position
 453         int pos = 0;
 454         for (int i = 0; i < numRet; i++) {
 455                 int retLen = rmiUtil->getTypeSize(retCls[i]);
 456                 // Get the 32-bit field in the byte array to get the actual
 457                 //              length (this is a param with indefinite length)
 458                 if (retLen == -1) {
 459                         char bytRetLen[IoTRMIUtil::RETURN_LEN];
 460                         memcpy(bytRetLen, retBytes + pos, IoTRMIUtil::RETURN_LEN);
 461                         pos = pos + IoTRMIUtil::RETURN_LEN;
 462                         int* retLenPtr = IoTRMIUtil::byteArrayToInt(&retLen, bytRetLen);
 463                         retLen = *retLenPtr;
 464                 }
 465                 char retObjBytes[retLen];
 466                 memcpy(retObjBytes, retBytes + pos, retLen);
 467                 pos = pos + retLen;
 468                 retObj[i] = IoTRMIUtil::getParamObject(retObj[i], retCls[i].c_str(), retObjBytes, retLen);
 469         }
 470
 471         return retObj;
 472 }
 473 #endif
 474
 475