colorg-server

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright © 2013 Progiciels Bourbeau-Pinard inc.
# François Pinard <pinard@iro.umontreal.ca>, 2013.

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.  */

"""\
colorg server, to orchestrate collobaration between many colorg users.

colorg is a real-time collaborative editing tool originally meant for
Emacs Org mode users.  See https://github.com/pinard/colorg/wiki/.

Usage: colorg-server [OPTION]... [HOST] [PORT]

Options:
  -d FLAGS   Debug according to FLAGS
  -h         Print this help, and do nothing else.
  -q         Be quiet, print only errors.

If HOST is specified, listen for connections on HOST instead of 0.0.0.0.
If PORT is specified, listen for connections on PORT rather than on 7997.
PORT being recognized as a decimal number, HOST may be omitted.

Debug FLAGS is one or more letter among "p" (communication protocol),
"e" (empty polls or execs in protocol) or "c" (modification of
changesets).
"""

__metaclass__ = type
import datetime
import gevent
from gevent import socket
import hashlib
import json
import sys


welcome = 'colorg v1'
encoding = 'UTF-8'
md5_empty = 'd41d8cd98f00b204e9800998ecf8427e'
localhost = '127.0.0.1'

# Debug flags
PROTO = 1
EMPTY = 2
CHSET = 4


class Error(Exception):
    pass


class Warn(Exception):
    pass


class Main:
    debug = 0
    host = '0.0.0.0'
    port = 7997
    verbose = True

    def main(self, *arguments):
        "Main entry point."

        # Decode options.
        import getopt
        options, arguments = getopt.getopt(arguments, 'd:hq')
        for option, value in options:
            if option == '-d':
                for letter in value:
                    self.debug |= {'p': PROTO, 'e': EMPTY, 'c': CHSET}[letter]
            elif option == '-h':
                sys.stdout.write(__doc__)
                return
            elif option == '-q':
                self.verbose = False
        if len(arguments) == 1 and arguments[0].isdigit():
            self.port = int(arguments[0])
        else:
            if len(arguments) >= 1:
                self.host = arguments[0]
            if len(arguments) >= 2:
                self.port = int(arguments[1])
            if len(arguments) > 2:
                sys.exit("More than 2 arguments.  Try %s -h for help."
                         % sys.argv[0])

        # Start it all.
        try:
            gevent.spawn(self.listener).join()
        except KeyboardInterrupt:
            sys.stderr.write('\n')

    def listener(self):
        "For each new connection, create a Client task to handle it."
        handle = socket.socket()
        handle.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
        handle.bind((self.host, self.port))
        handle.listen(1)
        if self.verbose:
            self.report("colorg server accepting connections on %s, port %d.",
                        self.host, self.port)
        while True:
            handle2, address = handle.accept()
            Client(handle2, *address).start()

    def report(self, format, *arguments):
        sys.stderr.write('%s\n' % (format % arguments).rstrip())


class Change:
    """\
A Change is specified by FIRST, DELETED, INSERT and USER.  The number
FIRST is the zero-based position of the start of a change.  The
changes removes DELETED characters starting from that position and
replaces them with the string INSERT.  The change also records that it
orginates by USER, identified by its user number.

For programming convenience, a change also has a few read-only fields:
  - LIMIT, which is the same as FIRST + DELETED;
  - LAST, which is the same as FIRST + len(INSERT);
  - DELTA, which is the same as len(INSERT) - DELETED.

The change deletes Characters from position FIRST to position LIMIT
(excluded), and the replacement text runs from postion FIRST to
position LAST (excluded).  The effect of the change wholly extends the
buffer by DELTA characters when DELTA is positive, or rather shrinks
it by -DELTA characters whenever DELTA is negative.  We always have
that DELTA equals LAST - LIMIT.
"""

    def __init__(self, first, deleted, insert, user):
        self.first = first
        self.deleted = deleted
        self.insert = insert
        self.user = user

    def __repr__(self):
        text = repr(self.insert)
        if text.startswith('u'):
            text = text[1:]
        return '%d-%d+%s' % (self.first, self.deleted, text[1:-1])

    def get_delta(self):
        return len(self.insert) - self.deleted

    def get_last(self):
        return self.first + len(self.insert)

    def get_limit(self):
        return self.first + self.deleted

    delta = property(get_delta)
    last = property(get_last)
    limit = property(get_limit)


class ChangeSet(list):
    """\
A changeset is a sequence of changes, kept sorted so they would apply
from the beginning to the end.  The value of FIRST in any change in a
changeset considers that all previous changes would have been already
applied.  Changes do not overlap and do not even touch.  Formally
stated, the value of FIRST in any change in a changeset is always
strictly greater than the value of LAST for the preceding change.
"""

    def __init__(self, resource, qualifier):
        list.__init__(self)
        self.resource = resource
        self.qualifier = qualifier

    def add(self, change):
        """\
Add CHANGE to a changeset.  If CHANGE.user is None, it is assumed that
the change has to be applied verbatim, with strict obedience in case
of clashes; this happens when modifications all originate from the
same user.  If CHANGE.user is not None, the change must be applied
cautiously so all work is preserved: deletes are done once at most and
previous inserts are never deleted; which is the case when changes
come from different users.
"""
        if run.debug & CHSET:
            print "Entering add(%r)" % change
            print self.dump('>')
        self.check('> add')
        verbatim = change.user is None
        # The new change is added at the end, then gets percolated down.  While
        # this is going, CHANGE is self[INDEX + 1] and PRIOR just below is
        # self[INDEX].  Changes self[:INDEX + 1] and self[INDEX + 2:]) are kept
        # clean.  CHANGE gets modified destructively in the process, so the
        # caller should pass a copy if the original has to be preserved.
        index = len(self)
        self.append(change)
        while index > 0:
            index -= 1
            prior = self[index]
            if change.first > prior.last:
                # CHANGE is already OK as it stands.
                break
            if change.first == prior.last:
                # CHANGE merely concatenates after PRIOR.  Merge it in.
                prior.insert += change.insert
                prior.deleted += change.deleted
                # User information is parly lost; this is innocuous.
                del self[index + 1]
                break
            if not verbatim:
                # FIXME: Should rather try to guess the user's intent!
                self.shift_after(index + 1, change.deleted)
                change.deleted = 0
            if change.first >= prior.first:
                # CHANGE is modifying PRIOR.insert.
                left = prior.insert[:change.first - prior.first]
                if change.limit < prior.last:
                    # Insert CHANGE.insert within PRIOR.insert.
                    right = prior.insert[change.limit - prior.first:]
                    prior.insert = left + change.insert + right
                else:
                    # Concatenate CHANGE.insert after truncated PRIOR.insert.
                    prior.insert = left + change.insert
                    prior.deleted += change.limit - prior.last
                del self[index + 1]
                break
            if change.limit >= prior.last:
                # All of PRIOR.insert is getting deleted, so remove PRIOR.
                change.deleted += prior.deleted - len(prior.insert)
                del self[index]
                continue
            # CHANGE is likely to bubble down.
            if change.limit >= prior.first:
                # The initial part of PRIOR.insert is getting deleted.
                removed = change.last - prior.first
                prior.insert = prior.insert[removed:]
                change.deleted -= removed
            # There is no delete overlap, so relocate and exchange.
            change.first -= prior.delta
            prior.first += change.delta
            if change.first <= prior.first:
                # With negative deltas, order may happen to be OK: recheck!
                index += 1
                continue
            self[index] = change
            self[index + 1] = prior
        if run.debug & CHSET:
            print self.dump('<')
            self.check('< add')

    def check(self, title=None):
        "Check for a healthy changeset, else dump it within an Error."
        last = -1
        for change in self:
            if change.first < 0 or change.deleted < 0 or change.first <= last:
                raise Error(self.dump(title or "Invalid changeset"))
            last = change.last

    def dump(self, title):
        "Produce a string representing the changeset, for debugging."
        fragments = []
        write = fragments.append
        write(u'%s.%s %s ' % (self.resource, self.qualifier, title))
        last = -1
        for change in self:
            write(' %r' % change)
            if change.first < 0 or change.deleted < 0 or change.first <= last:
                write(' ??')
            last = change.last
        return ''.join(fragments)

    def reset_first(self, change):
        "Set FIRST as it should be if this changeset was applied first."
        for cursor in self:
            # Nothing to do if the change appears later in the buffer.
            if change.last > cursor.first:
                if change.first > cursor.last:
                    # Merely shift if the change appears sooner.
                    delta = change.last - cursor.limit
                    change.first += delta
                else:
                    # FIXME!
                    pass

    def shift_after(self, index, delta):
        "For all of self[INDEX + 1:] changes, add DELTA to their FIRST value."
        index += 1
        while index < len(self):
            self[index].first += delta
            index += 1


class Client(gevent.Greenlet):
    "Receive all requests from a single client.  Produce a reply for each."
    ordinal = 0
    registry = {}
    login = None
    last_reported = None

    def __init__(self, handle, host, port):
        super(Client, self).__init__()
        self.ordinal = Client.ordinal
        Client.ordinal += 1
        Client.registry[self.ordinal] = self
        self.input = handle.makefile()
        self.output = handle.makefile('w', 0)
        self.host = host
        self.port = port
        if run.verbose:
            self.report("> Incoming connection.")
        text = '"%s"' % welcome
        if run.debug & PROTO:
            self.report('< %s', text)
        self.output.write(text + '\n')
        self.outgoing_queue = []

    def __repr__(self):
        return '%s@%s:%d' % (
            '???' if self.login is None else self.login,
            'localhost' if self.host == localhost else self.host,
            self.port)

    def _run(self):

        # Process incoming commands.
        while True:
            line = self.input.readline()
            if not line:
                if run.verbose:
                    self.report("> Disconnected.")
                break
            if run.debug & PROTO and (run.debug & EMPTY or line != '"poll"\n'):
                self.report('> %s', line.rstrip())
            try:
                try:
                    command = json.loads(line)
                except ValueError, exception:
                    raise Warn(str(exception))
                result = self.dispatch(command)
            except Error, exception:
                result = ['error', str(exception)]
            except Warn, exception:
                result = ['warn', str(exception)]
            text = json.dumps(result)
            assert '\n' not in text, repr(text)
            if run.debug & PROTO and (run.debug & EMPTY or text != '"exec"'):
                self.report('< %s', text)
            self.output.write(text + '\n')

        # Leave all resources.
        for resource in Resource.registry.itervalues():
            if self in resource.clients:
                resource.clients.remove(self)

    def dispatch(self, command):
        if command and isinstance(command, list):
            action = command[0]
            arguments = command[1:]
        else:
            action = command
            arguments = ()
        processor = getattr(self, 'do_%s' % action, None)
        if processor is None:
            raise Warn("Unrecognized command: %r" % command)
        try:
            return processor(*arguments)
        except TypeError, exception:
            raise Warn(str(exception))

    def do_alter(self, resource, first, deleted, insert, include_self=False):
        resource = self.resolve_resource(resource)
        if self not in resource.clients:
            raise Error("Resource is not monitored: %r" % resource)
        resource.alter(first, deleted, insert)
        for client in resource.clients:
            if include_self or client is not self:
                client.outgoing_queue.append(
                    ('alter', resource.ordinal, first, deleted, insert,
                     self.ordinal))
        return 'done'

    def do_chat(self, user, insert):
        client = self.resolve_user(user)
        client.outgoing_queue.append(['chat', self.ordinal, insert])
        return 'done'

    def do_create(self, name):
        if not isinstance(name, (str, unicode)):
            raise Warn("Name should be a string: %r" % name)
        if Resource.get(name) is not None:
            raise Warn("Resource name is already taken: %r" % name)
        resource = Resource(name)
        resource.clients.append(self)
        return ['done', resource.ordinal]

    def do_join(self, resource, md5sum):
        resource = self.resolve_resource(resource)
        if self in resource.clients:
            raise Warn("Already monitoring: %r" % resource)
        if md5sum == md5_empty:
            self.outgoing_queue.append(
                ('alter', resource.ordinal, 0, 0, resource.get_contents(), 0))
        elif md5sum != resource.md5sum():
            raise Warn("Checksums do not match: %r %s here, %s remotely."
                       % (resource, resource.md5sum(), md5sum))
        resource.clients.append(self)
        return ['done', resource.ordinal]

    def do_leave(self, resource):
        resource = self.resolve_resource(resource)
        if self not in resource.clients:
            raise Warn("Was not monitoring: %r" % resource)
        resource.clients.remove(self)
        return 'done'

    def do_login(self, name):
        if not isinstance(name, (str, unicode)):
            raise Warn("Login name should be a string: %r" % name)
        self.login = name
        Client.last_reported = None
        return ['done', self.ordinal]

    def do_logout(self):
        if self.login is None:
            raise Warn("Not logged in: %r" % self)
        del self.login
        Client.last_reported = None
        return 'done'

    def do_resources(self):
        return ['done'] + [
            [resource.name, key]
            for key, resource in sorted(Resource.registry.items())]

    def do_poll(self, *subcommands):
        # Group incoming commands into changesets, one per resource.
        incoming_by_resource = {}
        for command in subcommands:
            if command[0] == 'alter':
                _, resource, first, deleted, insert = command
                incoming = incoming_by_resource.get(resource)
                if incoming is None:
                    incoming = incoming_by_resource[resource] = ChangeSet(
                        resource, 'incoming')
                incoming.add(Change(first, deleted, insert, None))
            else:
                raise Error("Expecting an alter command: %r" % command)
        # Group outgoing commands into changesets, one per resource.
        outgoing_by_resource = {}
        outgoing_queue = []
        for command in self.outgoing_queue:
            if command[0] == 'alter':
                _, resource, first, deleted, insert, user = command
                outgoing = outgoing_by_resource.get(resource)
                if outgoing is None:
                    outgoing = outgoing_by_resource[resource] = ChangeSet(
                        resource, 'outgoing')
                outgoing.add(Change(first, deleted, insert, user))
            else:
                outgoing_queue.append(command)
        self.outgoing_queue = []
        # Handle each resource separately.
        for resource in set(incoming_by_resource) | set(outgoing_by_resource):
            incoming = incoming_by_resource.get(resource)
            outgoing = outgoing_by_resource.get(resource)
            if outgoing is None:
                # We only have INCOMING.
                for change in incoming:
                    self.do_alter(
                        resource, change.first, change.deleted, change.insert)
            elif incoming is None:
                # We only have OUTGOING.
                for change in outgoing:
                    outgoing_queue.append(
                        ['alter', resource, change.first, change.deleted,
                         change.insert, change.user])
            else:
                # We have both INCOMING and OUTGOING.
                undos = []
                redos = []
                for change in incoming:
                    undos.append([
                        'alter', resource, change.first, len(change.insert),
                        change.deleted, self.ordinal])
                    outgoing.reset_first(change)
                    redos.append(change)
                for change in redos:
                    self.do_alter(resource, change.first, change.deleted,
                                  change.insert, include_self=True)
                undos.reverse()
                outgoing_queue = undos + outgoing_queue
        if outgoing_queue:
            return ['exec'] + outgoing_queue
        return 'exec'

    def do_users(self):
        return ['done'] + [
            [repr(client), key]
            for key, client in sorted(Client.registry.items())]

    @staticmethod
    def get(name):
        if isinstance(name, int):
            return Client.registry.get(name)
        for key, client in Client.registry.iteritems():
            if name == client.name:
                return client

    def report(self, format, *arguments):
        if Client.last_reported is not self:
            run.report('%r', self)
            Client.last_reported = self
        now = datetime.datetime.now().time()
        run.report('  %.2d:%.2d:%.2d ' + format,
                   now.hour, now.minute, now.second, *arguments)

    def resolve_resource(self, resource):
        "Return the Resource structure for a given RESOURCE name or number."
        found = Resource.get(resource)
        if found is None:
            raise Warn("Resource is not found: %r" % resource)
        return found

    def resolve_user(self, user):
        "Return the Client structure for a given USER name or number."
        client = Client.get(user)
        if client is None:
            raise Warn("User is not found: %r" % user)
        return client


class Resource:
    ordinal = 0
    registry = {}

    def __init__(self, name):
        self.name = name
        self.ordinal = Resource.ordinal
        Resource.ordinal += 1
        Resource.registry[self.ordinal] = self
        self.contents = ''
        self.clients = []

    def __repr__(self):
        return repr(self.name)

    def alter(self, first, deleted, insert):
        self.contents = (self.contents[:first]
                         + insert
                         + self.contents[first + deleted:])

    @staticmethod
    def get(name):
        if isinstance(name, int):
            return Resource.registry.get(name)
        for key, resource in Resource.registry.iteritems():
            if name == resource.name:
                return resource

    def get_contents(self):
        return self.contents

    def md5sum(self):
        return hashlib.md5(self.contents.encode(encoding)).hexdigest()


run = Main()
main = run.main

if __name__ == '__main__':
    main(*sys.argv[1:])