iotjava/iotrmi/C++/IoTRMICommServer.hpp

   1 /** Class IoTRMICommServer implements the server side
   2  *  of IoTRMIComm class.
   3  *
   4  * @author      Rahmadi Trimananda <rtrimana @ uci.edu>
   5  * @version     1.0
   6  * @since       2017-01-28
   7  */
   8 #ifndef _IOTRMICOMMSERVER_HPP__
   9 #define _IOTRMICOMMSERVER_HPP__
  10
  11 #include <iostream>
  12 #include <string>
  13 #include <atomic>
  14 #include <limits>
  15 #include <thread>
  16 #include <mutex>
  17
  18 #include "IoTRMIComm.hpp"
  19
  20 using namespace std;
  21
  22
  23 class IoTRMICommServer final : public IoTRMIComm {
  24         public:
  25                 IoTRMICommServer(int _portSend, int _portRecv, bool* _bResult);
  26                 ~IoTRMICommServer();
  27                 // Public methods
  28                 void            sendReturnObj(void* retObj, string type, char* methodBytes);
  29                 void            sendReturnObj(void* retObj[], string type[], int numRet, char* methodBytes);
  30                 void            remoteCall(int objectId, int methodId, string paramCls[], void* paramObj[], int numParam);
  31
  32         private:
  33                 IoTSocketServer         *rmiServerSend;
  34                 IoTSocketServer         *rmiServerRecv;
  35
  36                 // Private methods
  37                 void            waitForConnectionOnServerRecv();
  38                 void            waitForConnectionOnServerSend();
  39                 void            waitForPackets(IoTRMICommServer* rmiComm);
  40 };
  41
  42
  43 // Constructor
  44 IoTRMICommServer::IoTRMICommServer(int _portSend, int _portRecv, bool* _bResult) : IoTRMIComm() {
  45
  46         rmiServerSend = new IoTSocketServer(_portSend, _bResult);
  47         rmiServerRecv = new IoTSocketServer(_portRecv, _bResult);
  48         thread th1 (&IoTRMICommServer::waitForConnectionOnServerSend, this);
  49         thread th2 (&IoTRMICommServer::waitForConnectionOnServerRecv, this);
  50         th1.join();
  51         th2.join();
  52         thread th3 (&IoTRMICommServer::waitForPackets, this, this);
  53         th3.detach();
  54 }
  55
  56
  57 // Destructor
  58 IoTRMICommServer::~IoTRMICommServer() {
  59
  60         // Clean up
  61         if (rmiServerSend != NULL) {
  62                 delete rmiServerSend;
  63                 rmiServerSend = NULL;
  64         }
  65         if (rmiServerRecv != NULL) {
  66                 delete rmiServerRecv;
  67                 rmiServerRecv = NULL;
  68         }
  69 }
  70
  71
  72 void IoTRMICommServer::waitForConnectionOnServerRecv() {
  73
  74         rmiServerRecv->connect();
  75 }
  76
  77
  78 void IoTRMICommServer::waitForConnectionOnServerSend() {
  79
  80         rmiServerSend->connect();
  81 }
  82
  83
  84 void IoTRMICommServer::waitForPackets(IoTRMICommServer* rmiComm) {
  85
  86         char* packetBytes = NULL;
  87         int packetLen = 0;
  88         while(true) {
  89                 fflush(NULL);
  90                 packetBytes = rmiComm->rmiServerRecv->receiveBytes(packetBytes, &packetLen);
  91                 fflush(NULL);
  92                 if (packetBytes != NULL) { // If there is method bytes
  93                         int packetType = IoTRMIComm::getPacketType(packetBytes);
  94                         if (packetType == IoTRMIUtil::METHOD_TYPE) {
  95                                 rmiComm->methodQueue.enqueue(packetBytes, packetLen);
  96                         } else if (packetType == IoTRMIUtil::RET_VAL_TYPE) {
  97                                 rmiComm->returnQueue.enqueue(packetBytes, packetLen);
  98                         } else {
  99                                 // TODO: We need to log error message when we come to running this using IoTSlave
 100                                 // TODO: Beware that using "cout" in the process will kill it (as IoTSlave is loaded at runtime)
 101                                 cerr << "IoTRMICommServer: Packet type is unknown: " << packetType << endl;
 102                                 exit(1);
 103                         }
 104                 }
 105                 packetBytes = NULL;
 106                 packetLen = 0;
 107         }
 108 }
 109
 110
 111 // Send return values in bytes to the caller
 112 void IoTRMICommServer::sendReturnObj(void* retObj, string type, char* methodBytes) {
 113
 114         // Critical section that is used by different objects
 115         lock_guard<mutex> guard(sendReturnObjMutex);
 116         // Find the length of return object in bytes
 117         int retLen = rmiUtil->getTypeSize(type);
 118         if (retLen == -1) {
 119                 retLen = rmiUtil->getVarTypeSize(type, retObj);
 120         }
 121         // Copy the header and object bytes
 122         int objAndMethIdLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN;
 123         int headerLen = objAndMethIdLen + IoTRMIUtil::PACKET_TYPE_LEN;
 124         char retAllObjBytes[headerLen+retLen];
 125         // Copy object and method Id first
 126         memcpy(retAllObjBytes, methodBytes, objAndMethIdLen);
 127         // Copy objectId + methodId + packet type in bytes
 128         char packType[IoTRMIUtil::PACKET_TYPE_LEN];
 129         IoTRMIUtil::intToByteArray(IoTRMIUtil::RET_VAL_TYPE, packType);
 130         memcpy(retAllObjBytes + objAndMethIdLen, packType, IoTRMIUtil::PACKET_TYPE_LEN);
 131         // Copy object into byte array
 132         char retObjBytes[retLen];
 133         IoTRMIUtil::getObjectBytes(retObjBytes, retObj, type.c_str());
 134         memcpy(retAllObjBytes + headerLen, retObjBytes, retLen);
 135         fflush(NULL);
 136         rmiServerSend->sendBytes(retAllObjBytes, headerLen+retLen);
 137         fflush(NULL);
 138 }
 139
 140
 141 // Send return values in bytes to the caller (for more than one object - struct)
 142 void IoTRMICommServer::sendReturnObj(void* retObj[], string type[], int numRet, char* methodBytes) {
 143
 144         // Critical section that is used by different objects
 145         lock_guard<mutex> guard(sendReturnObjMutex);
 146         // Find the length of return object in bytes
 147         int retLen = returnLength(retObj, type, numRet);
 148         // Copy the header and object bytes
 149         int objAndMethIdLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN;
 150         int headerLen = objAndMethIdLen + IoTRMIUtil::PACKET_TYPE_LEN;
 151         char retAllObjBytes[headerLen+retLen];
 152         // Copy object and method Id first
 153         memcpy(retAllObjBytes, methodBytes, objAndMethIdLen);
 154         // Copy objectId + methodId + packet type in bytes
 155         char packType[IoTRMIUtil::PACKET_TYPE_LEN];
 156         IoTRMIUtil::intToByteArray(IoTRMIUtil::RET_VAL_TYPE, packType);
 157         memcpy(retAllObjBytes + objAndMethIdLen, packType, IoTRMIUtil::PACKET_TYPE_LEN);
 158         // Copy object into byte array
 159         char retObjBytes[retLen];
 160         returnToBytes(retObj, type, retObjBytes, numRet);
 161         memcpy(retAllObjBytes + headerLen, retObjBytes, retLen);
 162         fflush(NULL);
 163         rmiServerSend->sendBytes(retAllObjBytes, headerLen+retLen);
 164         fflush(NULL);
 165 }
 166
 167
 168 // Calls a method remotely by passing in parameters and getting a return object
 169 void IoTRMICommServer::remoteCall(int objectId, int methodId, string paramCls[],
 170                 void* paramObj[], int numParam) {
 171
 172         // Critical section that is used by different objects
 173         lock_guard<mutex> guard(remoteCallMutex);
 174         // Send input parameters
 175         int len = methodLength(paramCls, paramObj, numParam);
 176         char method[len];
 177         methodToBytes(objectId, methodId, paramCls, paramObj, method, numParam);
 178         // Send bytes
 179         fflush(NULL);
 180         rmiServerSend->sendBytes(method, len);
 181         fflush(NULL);
 182
 183 }
 184 #endif
 185
 186