echo rpc client

gistfile1.py

# Shamelessly stolen from the Twisted site

from twisted.internet import protocol, reactor, defer, stdio
from twisted.protocols.basic import LineReceiver
from twisted.internet.endpoints import TCP4ClientEndpoint, connectProtocol

import json


class JsonEchoClient(LineReceiver):

    """
    Implements the server-side mechanisms
    """
    def __init__(self, *args):
        # LineReceiver.__init__(self, *args)
        self.acks = {}
        self.msgid = 0
        self.exposed = {
            "acknowledge": self.acknowledge,
            "message"    : self.message
        }
        self.output = None
    
    def connectionMade(self):
        print "Connected!"
        self.message(**{"from": "server",
                       "message": "connected"})
    
    def acknowledge(self, reply, blob):
        pass

    def sendCall(self, method, *args, **kwargs):
        blob = {
                "call": method,
                "id": self.msgid,
                "args": args,
                "kwargs": kwargs
            } 
        self.sendLine(
            json.dumps(blob)
        )
        self.msgid += 1
        self.acks[self.msgid] = {"blob": blob, "deferred": defer.Deferred()}
        
        self.acks[self.msgid]["deferred"].addCallback( self.acknowledge, blob )
        
        return self.acks[self.msgid]["deferred"]


    def sendMessage(self, msg):
        msg = {
            "to": "EchoTest",
            "from": your_name,
            "message": msg,
        }
        return self.sendCall("message", **msg)

    def message(self, **kwargs):

        self.output.sendMessage(kwargs)

    def sendReply(self, reply, msgid):
        blob = {"reply": reply,
                "id": msgid
                }
        self.sendLine(json.dumps(blob))

    def sendError(self, message, id_):
        blob = {"error": True,
                "message": message,
                "id": id_
                }
        self.sendLine(json.dumps(blob))
    
    def reply(self, m):
        # receives a reply from a remote host
        msgid = m["id"]
        if msgid not in self.acks:
            # error
            return self.error("no such id %s" % msgid)
        df = self.acks[msgid]["deferred"]
        df.callback(m)


    def lineReceived(self, line):
        """Receives a line from our remote"""
        print "Line!"
        m = None
        try:
            m = json.loads(line)
        except json.JSONDecodeError as e:
            self.sendCall(json.dumps(
                {
                    "error": True,
                    "message": "json parse error {error}".format(e)}
                ))
            print "received non-json-decodable string"
            return
        
        # Implementing the RPC system.
        if "call" in m:
            if m["call"] in self.exposed:
                method = self.exposed[ m["call"] ]
                reply = method(*m["args"], **m["kwargs"])
                self.sendReply(reply, m["id"])
            else:
                self.call("error", "no method {method}".format(m["call"]), m["id"])

        elif "reply" in m:
            self.reply(m)

        elif "error" in m:
            print "Couldn't call remote method: %s" % m["message"]
        
class StdioProtocol(LineReceiver):

    from os import linesep as delimiter

    def connectionMade(self):
        self.connected = True

    def __init__(self, wire):
        self.chat = wire
        wire.output = self

    def lineReceived(self, line):
        # print "Got line: %s" % line
        
        self.chat.sendMessage(line)
    def sendMessage(self, line):
        if not self.connected:
            raise NotConnectedError("Not connected")
        self.sendLine(str("<{from}> {message}".format(**line) ))



def on_connect(protocol):
    # Remember that a protocol proxies for one wire connection
    df = protocol.sendMessage("hello remote server")
    def ack(m):
        print "Got an %s!" % m
    df.addCallback(ack)


wire = JsonEchoClient()
stdio_ = StdioProtocol(wire)
stdio.StandardIO(stdio_)

endpoint = TCP4ClientEndpoint(reactor, "localhost", 1235 )
deferred = connectProtocol(endpoint, wire)
deferred.addCallback(on_connect)

reactor.run()