titan_rpc_client.py

"""
Simple client for the JSONRPC Server
"""
import asyncio
from pulsar.apps import rpc


proxy = rpc.JsonProxy("http://localhost:8060")


@asyncio.coroutine
def main():
    name, vid = yield from proxy.titan.create_person(
        "jon", "[email protected]", "https://jon.com/")
    print("Created vertex {} named {}".format(vid, name))


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()

titan_rpc_server.py

"""
Basic JSONRPC WSGI server with Pulsar. Could easily implement custom
RPC and serve on simple sockets or websockets
"""
import asyncio

from pulsar import ensure_future
from pulsar.apps import rpc, wsgi
from pulsar.apps.wsgi.utils import LOGGER
from pulsar.utils.httpurl import JSON_CONTENT_TYPES

from goblin import properties
from goblin.connection import setup, close_global_pool
from goblin.models import Vertex
from gremlinclient.aiohttp import Pool


class Person(Vertex):
    name = properties.String()
    email = properties.Email()
    url = properties.URL()


class TitanRPC(rpc.JSONRPC):

    def rpc_create_person(self, request, name, email, url):
        person = yield from Person.create(name=name, url=url, email=email)
        LOGGER.info("Created: {}".format(person))
        return [person.name, person.id]


class TitanRPCSite(wsgi.LazyWsgi):
    """Handler for the RPCServer"""

    def __init__(self):
        setup("localhost", pool_class=Pool, future=asyncio.Future)

    def setup(self, environ):
        commands = rpc.PulsarServerCommands()
        json_handler = commands.putSubHandler('titan', TitanRPC())
        middleware = wsgi.Router("/", post=json_handler,
                                 accept_content_types=JSON_CONTENT_TYPES)
        response = [wsgi.GZipMiddleware(200)]
        return wsgi.WsgiHandler(middleware=[wsgi.wait_for_body_middleware,
                                            middleware],
                                response_middleware=response,
                                async=True)


class TitanRPCServer(wsgi.WSGIServer):

    def monitor_stopping(self, monitor):
        loop = monitor._loop
        loop.call_soon(ensure_future, close_global_pool())


def server(callable=None, **params):
    return TitanRPCServer(TitanRPCSite(), **params)


if __name__ == "__main__":
    server().start()