sample of blocking msgpack-rpc client impl in python

test-blocking-msgpackrpc.py

#!/usr/bin/env python

import socket
import msgpack
import time
import random

REQUEST = 0
RESPONSE = 1
NOTIFY = 2

class Address(object):
    def __init__(self, host, port):
        self._host = host
        self._port = port

    @property
    def host(self):
        return self._host

    @property
    def port(self):
        return self._port

    def unpack(self):
        return (self._host, self._port)

    def socket(self, family=socket.AF_UNSPEC):
        res = socket.getaddrinfo(self._host, self._port, family,
                                 socket.SOCK_STREAM, 0, socket.AI_PASSIVE)[0]
        af, socktype, proto, canonname, sockaddr = res
        sock = socket.socket(af, socktype, proto)

#       set_close_exec(sock.fileno())
#       sock.setblocking(0)
        if af == socket.AF_INET6:
            if hasattr(socket, "IPPROTO_IPV6"):
                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)

        return sock

class ClientTransport(object):
    def __init__(self, session, address, timeout=10):
        self._session = session
        self._address = address
        self._sockets = []
        self._closed = False
        self._packer = msgpack.Packer()
        self._unpacker = msgpack.Unpacker()
        self._timeout = timeout

    def send_message(self, message):
        sock = self._address.socket()
        sock.settimeout(self._timeout)
        sock.connect(self._address.unpack())
        self._sockets.append(sock)
        sock.send(self._packer.pack(message))

    def recv_response(self, msgid):
        sock = self._sockets.pop()
        chunk = sock.recv(4096)
        self._unpacker.feed(chunk)
        for message in self._unpacker:
            msgsize = len(message)
            if msgsize != 4 and msgsize != 3:

                raise RuntimeError("Invalid MessagePack-RPC protocol: message = {0}".format(message))
            msgtype = message[0]
            if msgtype == RESPONSE and msgid == message[1]:
                error, result = message[2], message[3]
                if error:
                    raise RuntimeError("RPCError: %s" % repr(error))
                else:
                    return result

class Client(object):
    def __init__(self, address, timeout=10):
        self._address = address
        self._timeout = timeout
        self._transport = ClientTransport(self, address, timeout=timeout)

    @property
    def address(self):
        return self._address

    def call(self, method, *args):
        msgid = self.send_request(method, args)
        return self.recv_response(msgid)

    def send_request(self, method, args):
#       msgid = int(random.random() * 0xFFFFFFFF)
        msgid = int(time.time()) & 0xFFFFFFFF # msgid should be increased call by call
        self._transport.send_message([REQUEST, msgid, method, args])
        return msgid

    def recv_response(self, msgid):
        return self._transport.recv_response(msgid)

if __name__ == '__main__':
    address = Address('localhost', 9090)
    client = Client(address, timeout=3)
    got = client.call('ping')
    print('pong:' % repr(got))

# vim:set ft=python :