diff --git a/.hgignore b/.hgignore index 0bc7b92..48fc886 100644 --- a/.hgignore +++ b/.hgignore @@ -27,6 +27,7 @@ Makefile* # ostinato generated files version.cpp +pkg_info.json # vim swap files *.swp diff --git a/binding/rpc.py b/binding/rpc.py index 1d68ad2..9e3be2d 100644 --- a/binding/rpc.py +++ b/binding/rpc.py @@ -15,10 +15,12 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see +from google.protobuf.message import EncodeError, DecodeError from google.protobuf.service import RpcChannel from google.protobuf.service import RpcController import socket import struct +import sys class OstinatoRpcController(RpcController): def __init__(self): @@ -29,7 +31,13 @@ class OstinatoRpcChannel(RpcChannel): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) def connect(self, host, port): - self.sock.connect((host, port)) + self.peer = '%s:%d' % (host, port) + try: + self.sock.connect((host, port)) + except socket.error, e: + print 'ERROR: Unable to connect to Drone %s (%s)' % ( + self.peer, str(e)) + sys.exit(1) def disconnect(self): self.sock.close() @@ -40,36 +48,81 @@ class OstinatoRpcChannel(RpcChannel): OST_PB_MSG_TYPE_RESPONSE = 2 OST_PB_MSG_TYPE_BLOB = 3 - req = request.SerializeToString() - self.sock.sendall(struct.pack('>HHI', - OST_PB_MSG_TYPE_REQUEST, method.index, len(req)) + req) + try: + req = request.SerializeToString() + self.sock.sendall(struct.pack('>HHI', + OST_PB_MSG_TYPE_REQUEST, method.index, len(req)) + req) + except EncodeError, e: + print 'ERROR: Failed to serialize %s arg for RPC %s() ' \ + 'to Drone %s (%s)' % ( + type(request).__name__, method.name, self.peer, e) + sys.exit(1) + except socket.error, e: + print 'ERROR: Failed to invoke RPC %s() to Drone %s (%s)' % ( + method.name, self.peer, e) + sys.exit(1) # receive and parse header - hdr = '' - while len(hdr) < OST_PB_MSG_HDR_SIZE: - chunk = self.sock.recv(OST_PB_MSG_HDR_SIZE - len(hdr)) - if chunk == '': - raise RuntimeError("socket connection broken") - hdr = hdr + chunk + try: + hdr = '' + while len(hdr) < OST_PB_MSG_HDR_SIZE: + chunk = self.sock.recv(OST_PB_MSG_HDR_SIZE - len(hdr)) + if chunk == '': + raise RuntimeError("socket connection closed by peer") + hdr = hdr + chunk + except socket.error, e: + print 'ERROR: Failed to receive msg reply for RPC %s() ' \ + 'from Drone %s (%s)' % ( + method.name, self.peer, e) + sys.exit(1) + except RuntimeError, e: + print 'ERROR: Drone %s closed connection receiving msg reply ' \ + 'for RPC %s() (%s)' % ( + self.peer, method.name, e) + sys.exit(1) - (type, method, resp_len) = struct.unpack('>HHI', hdr) + (msg_type, method_index, resp_len) = struct.unpack('>HHI', hdr) + + # verify response method is same as the one requested + if method_index != method.index: + print 'ERROR: Received Reply for Method %d; expecting reply for ' \ + 'method %d (%s)' % ( + method_index, method.index, method.name) + sys.exit(1) # receive and parse the actual response message - resp = '' - while len(resp) < resp_len: - chunk = self.sock.recv(resp_len - len(resp)) - if chunk == '': - raise RuntimeError("socket connection broken") - resp = resp + chunk + try: + resp = '' + while len(resp) < resp_len: + chunk = self.sock.recv(resp_len - len(resp)) + if chunk == '': + raise RuntimeError("socket connection closed by peer") + resp = resp + chunk + except socket.error, e: + print 'ERROR: Failed to receive reply for RPC %s() ' \ + 'from Drone %s (%s)' % ( + method.name, self.peer, e) + sys.exit(1) + except RuntimeError, e: + print 'ERROR: Drone %s closed connection receiving reply ' \ + 'for RPC %s() (%s)' % ( + self.peer, method.name, e) + sys.exit(1) - if type == OST_PB_MSG_TYPE_RESPONSE: - response = response_class() - response.ParseFromString(resp) - elif type == OST_PB_MSG_TYPE_BLOB: + if msg_type == OST_PB_MSG_TYPE_RESPONSE: + try: + response = response_class() + response.ParseFromString(resp) + except DecodeError, e: + print 'ERROR: Failed to parse %s response for RPC %s() ' \ + 'from Drone %s (%s)' % ( + type(response).__name__, method.name, self.peer, e) + sys.exit(1) + elif msg_type == OST_PB_MSG_TYPE_BLOB: response = resp else: - print 'unsupported msg type %d received in respone to %s' % \ - type, method.name + print 'ERROR: unsupported msg type %d received in respone to %s' % \ + msg_type, method.name controller.response = response