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 : 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         cout << "Wait on connection ServerRecv!" << endl;
  75         rmiServerRecv->connect();
  76         cout << "Connected on connection ServerRecv!" << endl;
  77 }
  78
  79
  80 void IoTRMICommServer::waitForConnectionOnServerSend() {
  81
  82         cout << "Wait on connection ServerSend!" << endl;
  83         rmiServerSend->connect();
  84         cout << "Connected on connection ServerSend!" << endl;
  85 }
  86
  87
  88 void IoTRMICommServer::waitForPackets(IoTRMICommServer* rmiComm) {
  89
  90         char* packetBytes = NULL;
  91         int packetLen = 0;
  92         //cout << "Starting waitForPacketsOnServer()" << endl;
  93         while(true) {
  94                 fflush(NULL);
  95                 packetBytes = rmiComm->rmiServerRecv->receiveBytes(packetBytes, &packetLen);
  96                 fflush(NULL);
  97                 if (packetBytes != NULL) { // If there is method bytes
  98                         //IoTRMIUtil::printBytes(packetBytes, packetLen, false);
  99                         int packetType = IoTRMIComm::getPacketType(packetBytes);
 100                         if (packetType == IoTRMIUtil::METHOD_TYPE) {
 101                                 rmiComm->methodQueue.enqueue(packetBytes, packetLen);
 102                         } else if (packetType == IoTRMIUtil::RET_VAL_TYPE) {
 103                                 rmiComm->returnQueue.enqueue(packetBytes, packetLen);
 104                         } else {
 105                                 // TODO: We need to log error message when we come to running this using IoTSlave
 106                                 // TODO: Beware that using "cout" in the process will kill it (as IoTSlave is loaded in Sentinel)
 107                                 cerr << "IoTRMICommServer: Packet type is unknown: " << packetType << endl;
 108                                 exit(1);
 109                         }
 110                 }
 111                 packetBytes = NULL;
 112                 packetLen = 0;
 113         }
 114 }
 115
 116
 117 // Send return values in bytes to the caller
 118 void IoTRMICommServer::sendReturnObj(void* retObj, string type, char* methodBytes) {
 119
 120         // Critical section that is used by different objects
 121         lock_guard<mutex> guard(sendReturnObjMutex);
 122         // Find the length of return object in bytes
 123         int retLen = rmiUtil->getTypeSize(type);
 124         if (retLen == -1) {
 125                 retLen = rmiUtil->getVarTypeSize(type, retObj);
 126         }
 127         // Copy the header and object bytes
 128         int objAndMethIdLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN;
 129         int headerLen = objAndMethIdLen + IoTRMIUtil::PACKET_TYPE_LEN;
 130         char retAllObjBytes[headerLen+retLen];
 131         // Copy object and method Id first
 132         memcpy(retAllObjBytes, methodBytes, objAndMethIdLen);
 133         // Copy objectId + methodId + packet type in bytes
 134         char packType[IoTRMIUtil::PACKET_TYPE_LEN];
 135         IoTRMIUtil::intToByteArray(IoTRMIUtil::RET_VAL_TYPE, packType);
 136         memcpy(retAllObjBytes + objAndMethIdLen, packType, IoTRMIUtil::PACKET_TYPE_LEN);
 137         // Copy object into byte array
 138         char retObjBytes[retLen];
 139         IoTRMIUtil::getObjectBytes(retObjBytes, retObj, type.c_str());
 140         memcpy(retAllObjBytes + headerLen, retObjBytes, retLen);
 141         fflush(NULL);
 142         IoTRMIUtil::printBytes(retAllObjBytes, headerLen+retLen, false);
 143         rmiServerSend->sendBytes(retAllObjBytes, headerLen+retLen);
 144         fflush(NULL);
 145 }
 146
 147
 148 // Send return values in bytes to the caller (for more than one object - struct)
 149 void IoTRMICommServer::sendReturnObj(void* retObj[], string type[], int numRet, char* methodBytes) {
 150
 151         // Critical section that is used by different objects
 152         lock_guard<mutex> guard(sendReturnObjMutex);
 153         // Find the length of return object in bytes
 154         int retLen = returnLength(retObj, type, numRet);
 155         // Copy the header and object bytes
 156         int objAndMethIdLen = IoTRMIUtil::OBJECT_ID_LEN + IoTRMIUtil::METHOD_ID_LEN;
 157         int headerLen = objAndMethIdLen + IoTRMIUtil::PACKET_TYPE_LEN;
 158         char retAllObjBytes[headerLen+retLen];
 159         // Copy object and method Id first
 160         memcpy(retAllObjBytes, methodBytes, objAndMethIdLen);
 161         // Copy objectId + methodId + packet type in bytes
 162         char packType[IoTRMIUtil::PACKET_TYPE_LEN];
 163         IoTRMIUtil::intToByteArray(IoTRMIUtil::RET_VAL_TYPE, packType);
 164         memcpy(retAllObjBytes + objAndMethIdLen, packType, IoTRMIUtil::PACKET_TYPE_LEN);
 165         // Copy object into byte array
 166         char retObjBytes[retLen];
 167         returnToBytes(retObj, type, retObjBytes, numRet);
 168         memcpy(retAllObjBytes + headerLen, retObjBytes, retLen);
 169         fflush(NULL);
 170         rmiServerSend->sendBytes(retAllObjBytes, headerLen+retLen);
 171         fflush(NULL);
 172 }
 173
 174
 175 // Calls a method remotely by passing in parameters and getting a return object
 176 void IoTRMICommServer::remoteCall(int objectId, int methodId, string paramCls[],
 177                 void* paramObj[], int numParam) {
 178
 179         // Critical section that is used by different objects
 180         lock_guard<mutex> guard(remoteCallMutex);
 181         // Send input parameters
 182         int len = methodLength(paramCls, paramObj, numParam);
 183         char method[len];
 184         methodToBytes(objectId, methodId, paramCls, paramObj, method, numParam);
 185         // Send bytes
 186         fflush(NULL);
 187         rmiServerSend->sendBytes(method, len);
 188         fflush(NULL);
 189
 190 }
 191 #endif
 192
 193