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;
11 import java.lang.reflect.*;
13 import java.util.concurrent.*;
14 import java.util.concurrent.atomic.AtomicBoolean;
17 /** Class IoTRMICommServer is a class that extends IoTRMI
19 * We will arbitrate packets into 2 queues and wake up the right threads/callers.
20 * We separate traffics one-directionally.
22 * @author Rahmadi Trimananda <rtrimana @ uci.edu>
26 public class IoTRMICommServer extends IoTRMIComm {
31 private IoTSocketServer rmiServerSend;
32 private IoTSocketServer rmiServerRecv;
33 private AtomicBoolean didServerSendConnect;
34 private AtomicBoolean didServerRecvConnect;
38 * Constructor (for skeleton)
40 public IoTRMICommServer(int _portSend, int _portRecv) throws
41 ClassNotFoundException, InstantiationException,
42 IllegalAccessException, IOException {
45 didServerSendConnect = new AtomicBoolean(false);
46 didServerRecvConnect = new AtomicBoolean(false);
47 rmiServerSend = new IoTSocketServer(_portSend);
48 rmiServerRecv = new IoTSocketServer(_portRecv);
49 waitForConnectionOnServerSend();
50 waitForConnectionOnServerRecv();
51 while(!didServerSendConnect.get()); // Wait until server is connected
52 while(!didServerRecvConnect.get()); // Wait until server is connected
53 waitForPacketsOnServer();
58 * waitForConnectionOnServerRecv() starts a thread that waits server connection
60 public void waitForConnectionOnServerRecv() {
62 Thread thread = new Thread() {
65 rmiServerRecv.connect();
66 didServerRecvConnect.set(true);
67 } catch (Exception ex) {
69 throw new Error("IoTRMICommServer: Error starting receiver server!");
78 * waitForConnectionOnServerSend() starts a thread that waits server connection
80 public void waitForConnectionOnServerSend() {
82 Thread thread = new Thread() {
85 rmiServerSend.connect();
86 didServerSendConnect.set(true);
87 } catch (Exception ex) {
89 throw new Error("IoTRMICommServer: Error starting sender server!");
98 * waitForPacketsOnServer() starts a thread that waits for packet bytes on server side
100 public void waitForPacketsOnServer() {
102 Thread thread = new Thread() {
104 byte[] packetBytes = null;
107 packetBytes = rmiServerRecv.receiveBytes(packetBytes);
108 if (packetBytes != null) {
109 System.out.println("Packet received: " + Arrays.toString(packetBytes));
110 int packetType = IoTRMIComm.getPacketType(packetBytes);
111 if (packetType == IoTRMIUtil.METHOD_TYPE) {
112 System.out.println("Method packet: " + Arrays.toString(packetBytes));
113 methodQueue.offer(packetBytes);
114 } else if (packetType == IoTRMIUtil.RET_VAL_TYPE) {
115 System.out.println("Return value packet: " + Arrays.toString(packetBytes));
116 returnQueue.offer(packetBytes);
118 throw new Error("IoTRMICommServer: Packet type is unknown: " + packetType);
120 // Thread.sleep(100);
122 } catch (Exception ex) {
123 ex.printStackTrace();
124 throw new Error("IoTRMICommServer: Error receiving return value bytes on server!");
134 * sendReturnObj() for non-struct objects (server side)
136 public synchronized void sendReturnObj(Object retObj, byte[] methodBytes) {
138 // Send back return value
139 byte[] retObjBytes = IoTRMIUtil.getObjectBytes(retObj);
140 // Send return value together with OBJECT_ID and METHOD_ID for arbitration
141 int objMethIdLen = IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN;
142 int headerLen = objMethIdLen + IoTRMIUtil.PACKET_TYPE_LEN;
143 byte[] retAllBytes = new byte[headerLen + retObjBytes.length];
144 // Copy OBJECT_ID and METHOD_ID
145 System.arraycopy(methodBytes, 0, retAllBytes, 0, objMethIdLen);
146 int packetType = IoTRMIUtil.RET_VAL_TYPE; // This is a return value
147 byte[] packetTypeBytes = IoTRMIUtil.intToByteArray(packetType);
148 System.arraycopy(packetTypeBytes, 0, retAllBytes, objMethIdLen, IoTRMIUtil.PACKET_TYPE_LEN);
149 // Copy array of bytes (return object)
150 System.arraycopy(retObjBytes, 0, retAllBytes, headerLen, retObjBytes.length);
152 rmiServerSend.sendBytes(retAllBytes);
153 } catch (IOException ex) {
154 ex.printStackTrace();
155 throw new Error("IoTRMICommServer: Error sending bytes in sendReturnObj()!");
161 * sendReturnObj() overloaded to send multiple return objects for structs (server side)
163 public synchronized void sendReturnObj(Class<?>[] retCls, Object[] retObj, byte[] methodBytes) {
165 // Send back return value
166 byte[] retObjBytes = returnToBytes(retCls, retObj);
167 // Send return value together with OBJECT_ID and METHOD_ID for arbitration
168 int objMethIdLen = IoTRMIUtil.OBJECT_ID_LEN + IoTRMIUtil.METHOD_ID_LEN;
169 int headerLen = objMethIdLen + IoTRMIUtil.PACKET_TYPE_LEN;
170 byte[] retAllBytes = new byte[headerLen + retObjBytes.length];
171 // Copy OBJECT_ID and METHOD_ID
172 System.arraycopy(methodBytes, 0, retAllBytes, 0, objMethIdLen);
173 int packetType = IoTRMIUtil.RET_VAL_TYPE; // This is a return value
174 byte[] packetTypeBytes = IoTRMIUtil.intToByteArray(packetType);
175 System.arraycopy(packetTypeBytes, 0, retAllBytes, objMethIdLen, IoTRMIUtil.PACKET_TYPE_LEN);
176 // Copy array of bytes (return object)
178 rmiServerSend.sendBytes(retAllBytes);
179 } catch (IOException ex) {
180 ex.printStackTrace();
181 throw new Error("IoTRMICommServer: Error sending bytes in sendReturnObj()!");
187 * remoteCall() calls a method remotely by passing in parameters (server side)
189 public synchronized void remoteCall(int objectId, int methodId, Class<?>[] paramCls, Object[] paramObj) {
192 byte[] methodBytes = methodToBytes(objectId, methodId, paramCls, paramObj);
194 rmiServerSend.sendBytes(methodBytes);
195 } catch (IOException ex) {
196 ex.printStackTrace();
197 throw new Error("IoTRMICommServer: Error when sending bytes in remoteCall()!");