demo-journal.py

import os, sys, json, contextlib, random
from twisted.internet import task, defer, endpoints
from twisted.application import service, internet
from twisted.web import server, static, resource
from wormhole import journal, wormhole

# considerations for state management:
# * be somewhat principled about the data (e.g. have a schema)
# * discourage accidental schema changes
# * avoid surprise mutations by app code (don't hand out mutables)
# * discourage app from keeping state itself: make state object easy enough
#   to use for everything. App should only hold objects that are active
#   (Services, subscribers, etc). App must wire up these objects each time.


def parse(args):
    raise NotImplementedError


def update_my_state():
    raise NotImplementedError


class State(object):
    @classmethod
    def create_empty(klass):
        self = klass()
        # to avoid being tripped up by state-mutation side-effect bugs, we
        # hold the serialized state in RAM, and re-deserialize it each time
        # someone asks for a piece of it.    # iid->invitation_stat
        empty = {"version": 1,
                 "invitations": {},
                 "contacts": [],
                 }
        self._bytes = json.dumps(empty).encode("utf-8")
        return self

    @classmethod
    def from_filename(klass, fn):
        self = klass()
        with open(fn, "rb") as f:
            bytes = f.read()
        self._bytes = bytes
        # version check
        data = self._as_data()
        assert data["version"] == 1
        # schema check?
        return self

    def save_to_filename(self, fn):
        tmpfn = fn + ".tmp"
        with open(tmpfn, "wb") as f:
            f.write(self._bytes)
        os.rename(tmpfn, fn)

    def _as_data(self):
        return json.loads(bytes.decode("utf-8"))

    @contextlib.contextmanager
    def _mutate(self):
        data = self._as_data()
        yield data  # mutable
        self._bytes = json.dumps(data).encode("utf-8")

    def get_all_invitations(self):
        return self._as_data()["invitations"]

    def add_invitation(self, iid, invitation_state):
        with self._mutate() as data:
            data["invitations"][iid] = invitation_state

    def update_invitation(self, iid, invitation_state):
        with self._mutate() as data:
            assert iid in data["invitations"]
            data["invitations"][iid] = invitation_state

    def remove_invitation(self, iid):
        with self._mutate() as data:
            del data["invitations"][iid]

    def add_contact(self, contact):
        with self._mutate() as data:
            data["contacts"].append(contact)


class Root(resource.Resource):
    pass


class Status(resource.Resource):
    def __init__(self, c):
        resource.Resource.__init__(self)
        self._call = c

    def render_GET(self, req):
        data = self._call()
        req.setHeader(b"content-type", "text/plain")
        return data


class Action(resource.Resource):
    def __init__(self, c):
        resource.Resource.__init__(self)
        self._call = c

    def render_POST(self, req):
        req.setHeader(b"content-type", "text/plain")
        try:
            args = json.load(req.content)
        except ValueError:
            req.setResponseCode(500)
            return b"bad JSON"
        data = self._call(args)
        return data


class Agent(service.MultiService):
    def __init__(self, basedir, reactor):
        service.MultiService.__init__(self)
        self._basedir = basedir
        self._reactor = reactor

        root = Root()
        site = server.Site(root)
        ep = endpoints.serverFromString(reactor, "tcp:8220")
        internet.StreamServerEndpointService(ep, site).setServiceParent(self)

        self._jm = journal.JournalManager(self._save_state)

        root.putChild(b"", static.Data("root", "text/plain"))
        root.putChild(b"list-invitations", Status(self._list_invitations))
        root.putChild(b"invite", Action(self._invite))  # {petname:}
        root.putChild(b"accept", Action(self._accept))  # {petname:, code:}

        self._state_fn = os.path.join(self._basedir, "state.json")
        self._state = State.from_filename(self._state_fn)

        self._wormholes = {}
        for iid, invitation_state in self._state.get_all_invitations().items():
            def _dispatch(event, *args, **kwargs):
                self._dispatch_wormhole_event(iid, event, *args, **kwargs)
            w = wormhole.journaled_from_data(invitation_state["wormhole"],
                                             reactor=self._reactor,
                                             journal=self._jm,
                                             event_handler=self,
                                             event_handler_args=(iid,))
            self._wormholes[iid] = w
            w.setServiceParent(self)

    def _save_state(self):
        self._state.save_to_filename(self._state_fn)

    def _list_invitations(self):
        inv = self._state.get_all_invitations()
        lines = ["%d: %s" % (iid, inv[iid]) for iid in sorted(inv)]
        return b"\n".join(lines) + b"\n"

    def _invite(self, args):
        print("invite", args)
        petname = args["petname"]
        # it'd be better to use a unique object for the event_handler
        # correlation, but we can't store them into the state database. I'm
        # not 100% sure we need one for the database: maybe it should hold a
        # list instead, and assign lookup keys at runtime. If they really
        # need to be serializable, they should be allocated rather than
        # random.
        iid = random.randint(1, 1000)
        my_pubkey = random.randint(1, 1000)
        with self._jm.process():
            w = wormhole.journaled(reactor=self._reactor, journal=self._jm,
                                   event_handler=self,
                                   event_handler_args=(iid,))
            self._wormholes[iid] = w
            w.setServiceParent(self)
            w.get_code()  # event_handler means code returns via callback
            invitation_state = {"wormhole": w.to_data(),
                                "petname": petname,
                                "my_pubkey": my_pubkey,
                                }
            self._state.add_invitation(iid, invitation_state)
        return b"ok"

    def _accept(self, args):
        print("accept", args)
        petname = args["petname"]
        code = args["code"]
        iid = random.randint(1, 1000)
        my_pubkey = random.randint(2, 2000)
        with self._jm.process():
            w = wormhole.journaled(reactor=self._reactor, journal=self._jm,
                                   event_dispatcher=self,
                                   event_dispatcher_args=(iid,))
            w.set_code(code)
            md = {"my_pubkey": my_pubkey}
            w.send(json.dumps(md).encode("utf-8"))
            invitation_state = {"wormhole": w.to_data(),
                                "petname": petname,
                                "my_pubkey": my_pubkey,
                                }
            self._state.add_invitation(iid, invitation_state)
        return b"ok"

    # dispatch options:
    # * register one function, which takes (eventname, *args)
    #   * to handle multiple wormholes, app must give is a closure
    # * register multiple functions (one per event type)
    # * register an object, with well-known method names
    # * extra: register args and/or kwargs with the callback
    #
    # events to dispatch:
    #  generated_code(code)
    #  got_verifier(verifier_bytes)
    #  verified()
    #  got_data(data_bytes)
    #  closed()

    def wormhole_dispatch_got_code(self, code, iid):
        # we're already in a jm.process() context
        invitation_state = self._state.get_all_invitations()[iid]
        invitation_state["code"] = code
        self._state.update_invitation(iid, invitation_state)
        self._wormholes[iid].set_code(code)
        # notify UI subscribers to update the display

    def wormhole_dispatch_got_verifier(self, verifier, iid):
        pass

    def wormhole_dispatch_verified(self, _, iid):
        pass

    def wormhole_dispatch_got_data(self, data, iid):
        invitation_state = self._state.get_all_invitations()[iid]
        md = json.loads(data.decode("utf-8"))
        contact = {"petname": invitation_state["petname"],
                   "my_pubkey": invitation_state["my_pubkey"],
                   "their_pubkey": md["my_pubkey"],
                   }
        self._state.add_contact(contact)
        self._wormholes[iid].close()  # now waiting for "closed"

    def wormhole_dispatch_closed(self, _, iid):
        self._wormholes[iid].disownServiceParent()
        del self._wormholes[iid]
        self._state.remove_invitation(iid)

    def handle_app_event(self, args, ack_f):  # sample function
        # Imagine here that the app has received a message (not
        # wormhole-related) from some other server, and needs to act on it.
        # Also imagine that ack_f() is how we tell the sender that they can
        # stop sending the message, or how we ask our poller/subscriber
        # client to send a DELETE message. If the process dies before ack_f()
        # delivers whatever it needs to deliver, then in the next launch,
        # handle_app_event() will be called again.
        stuff = parse(args)  # noqa
        with self._jm.process():
            update_my_state()
            self._jm.queue_outbound(ack_f)


def create(reactor, basedir):
    os.mkdir(basedir)
    s = State.create_empty()
    s.save(os.path.join(basedir, "state.json"))
    return defer.succeed(None)


def run(reactor, basedir):
    a = Agent(basedir, reactor)
    a.startService()
    print("agent listening on http://localhost:8220/")
    d = defer.Deferred()
    return d


if __name__ == "__main__":
    command = sys.argv[1]
    basedir = sys.argv[2]
    if command == "create":
        task.react(create, (basedir,))
    elif command == "run":
        task.react(run, (basedir,))
    else:
        print("Unrecognized subcommand '%s'" % command)
        sys.exit(1)