view release on metacpan or  search on metacpan

src/subversion/tools/server-side/svnpubsub/svnpubsub/client.py  view on Meta::CPAN

# Generic client for SvnPubSub
#
# ### usage...
#
#
# EVENTS
#
#   connected: a connection to the server has been opened (though not
#                 necessarily established)
#   closed:    the connection was closed. reconnect will be attempted.
#   error:     an error closed the connection. reconnect will be attempted.
#   ping:      the server has sent a keepalive
#   stale:     no activity has been seen, so the connection will be closed
#                 and reopened
#

import asyncore
import asynchat
import socket
import functools
import time
import json
try:
  import urlparse
except ImportError:
  import urllib.parse as urlparse

# How long the polling loop should wait for activity before returning.
TIMEOUT = 30.0

# Always delay a bit when trying to reconnect. This is not precise, but sets
# a minimum amount of delay. At the moment, there is no further backoff.
RECONNECT_DELAY = 25.0

# If we don't see anything from the server for this amount time, then we
# will drop and reconnect. The TCP connection may have gone down without
# us noticing it somehow.
STALE_DELAY = 60.0


class SvnpubsubClientException(Exception):
  pass

class Client(asynchat.async_chat):

  def __init__(self, url, commit_callback, event_callback):
    asynchat.async_chat.__init__(self)

    self.last_activity = time.time()
    self.ibuffer = []

    self.url = url
    parsed_url = urlparse.urlsplit(url)
    if parsed_url.scheme != 'http':
      raise ValueError("URL scheme must be http: '%s'" % url)
    host = parsed_url.hostname
    port = parsed_url.port
    resource = parsed_url.path
    if parsed_url.query:
      resource += "?%s" % parsed_url.query
    if parsed_url.fragment:
      resource += "#%s" % parsed_url.fragment

    self.event_callback = event_callback

    self.parser = JSONRecordHandler(commit_callback, event_callback)

    # Wait for the end of headers. Then we start parsing JSON.
    self.set_terminator(b'\r\n\r\n')
    self.skipping_headers = True

    self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
      self.connect((host, port))
    except:
      self.handle_error()
      return

    self.push(('GET %s HTTP/1.0\r\n\r\n' % resource).encode('ascii'))

  def handle_connect(self):
    self.event_callback('connected', None)

  def handle_close(self):
    self.event_callback('closed', None)
    self.close()

  def handle_error(self):
    self.event_callback('error', None)
    self.close()

  def found_terminator(self):
    if self.skipping_headers:
      self.skipping_headers = False
      # Each JSON record is terminated by a null character
      self.set_terminator(b'\0')
    else:
      record = b"".join(self.ibuffer)
      self.ibuffer = []
      self.parser.feed(record.decode())

  def collect_incoming_data(self, data):
    # Remember the last time we saw activity
    self.last_activity = time.time()

    if not self.skipping_headers:
      self.ibuffer.append(data)


class JSONRecordHandler:
  def __init__(self, commit_callback, event_callback):
    self.commit_callback = commit_callback
    self.event_callback = event_callback

  def feed(self, record):
    obj = json.loads(record)
    if 'svnpubsub' in obj:
      actual_version = obj['svnpubsub'].get('version')
      EXPECTED_VERSION = 1
      if actual_version != EXPECTED_VERSION:
        raise SvnpubsubClientException("Unknown svnpubsub format: %r != %d"
                                       % (actual_format, expected_format))