titan_pulsar.py

"""
This example demonstrates how to use Pulsar actors to create
nodes with Goblin. This is the basic concept on which other Pulsar/Goblin
apps would be based, but it may be better to use Pulsar's higher
level Application class interface...
Based on:
https://github.com/quantmind/pulsar/blob/master/examples/snippets/greeter.py
"""
import asyncio
import pulsar

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


nodes = [
    {"name": "dave", "email": "[email protected]", "url": "https://dave.com/"},
    {"name": "jon", "email": "[email protected]", "url": "https://jon.com/"},
    {"name": "leif", "email": "[email protected]", "url": "https://leif.com/"}]


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


@pulsar.command()
@asyncio.coroutine
def create_person(request, message):
    name = message["name"]
    url = message["url"]
    email = message["email"]
    person = yield from Person.create(name=name, url=url, email=email)
    request.actor.logger.info("Created: {}".format(person))
    return person


class Creator:

    def __init__(self):
        # Allow passing of config args
        setup('localhost', pool_class=Pool, future=asyncio.Future)
        cfg = pulsar.Config()
        cfg.parse_command_line()
        # Arbiter controls the main event loop in master process
        arbiter = pulsar.arbiter(cfg=cfg)
        self.cfg = arbiter.cfg
        # Conforms to the Pulsar definiton of an async object
        self._loop = arbiter._loop
        self._loop.call_later(1, pulsar.ensure_future, self())
        arbiter.start()

    @asyncio.coroutine
    def __call__(self, actor=None):
        if actor is None:
            # This creates an actor in its own process with its own loop
            actor = yield from pulsar.spawn(name="creator")
        if nodes:
            node = nodes.pop()
            self._loop.logger.info("Creating: {}".format(node["name"]))
            # Send the task of creating a person to the actor
            yield from pulsar.send(actor, 'create_person', node)
            self._loop.call_soon(pulsar.ensure_future, self(actor))
        else:
            # Stop the event loop
            yield from close_global_pool()
            pulsar.arbiter().stop()

if __name__ == "__main__":
    Creator()