| ################################################# |
| # Pubsubber.py: a minimalistic svnwcsub program # |
| ################################################# |
| |
| from threading import Thread |
| from datetime import datetime |
| import os |
| import sys |
| import logging |
| import atexit |
| import signal |
| import json |
| import re |
| import time |
| import subprocess |
| import base64 |
| |
| version = 2 |
| if sys.hexversion < 0x03000000: |
| print("Using Python 2...") |
| import httplib, urllib, urllib2, ConfigParser as configparser, socket |
| socket._fileobject.default_bufsize = 0 |
| else: |
| print("Using Python 3") |
| version = 3 |
| import http.client, urllib.request, urllib.parse, configparser |
| |
| |
| debug = False |
| logger = None |
| watching={} # dict: key = path to watch, value = set() of paths to update on match |
| |
| ############################################################ |
| # Daemon class, courtesy of an anonymous good-hearted soul # |
| ############################################################ |
| class daemon: |
| """A generic daemon class. |
| |
| Usage: subclass the daemon class and override the run() method.""" |
| |
| def __init__(self, pidfile, logfile = None): |
| self.pidfile = pidfile |
| if logfile == None: |
| self.logfile = os.devnull |
| else: |
| self.logfile = logfile |
| |
| def daemonize(self): |
| """Daemonize class. UNIX double fork mechanism.""" |
| |
| try: |
| pid = os.fork() |
| if pid > 0: |
| # exit first parent |
| sys.exit(0) |
| except OSError as err: |
| sys.stderr.write('fork #1 failed: {0}\n'.format(err)) |
| sys.exit(1) |
| |
| # decouple from parent environment |
| os.chdir('/') |
| os.setsid() |
| os.umask(0) |
| |
| # do second fork |
| try: |
| pid = os.fork() |
| if pid > 0: |
| |
| # exit from second parent |
| sys.exit(0) |
| except OSError as err: |
| sys.stderr.write('fork #2 failed: {0}\n'.format(err)) |
| sys.exit(1) |
| |
| # redirect standard file descriptors |
| sys.stdout.flush() |
| sys.stderr.flush() |
| si = open(os.devnull, 'r') |
| so = open(self.logfile, 'a+') |
| se = open(self.logfile, 'a+') |
| |
| os.dup2(si.fileno(), sys.stdin.fileno()) |
| os.dup2(so.fileno(), sys.stdout.fileno()) |
| os.dup2(se.fileno(), sys.stderr.fileno()) |
| |
| # write pidfile |
| atexit.register(self.delpid) |
| |
| pid = str(os.getpid()) |
| with open(self.pidfile,'w+') as f: |
| f.write(pid + '\n') |
| logger.info("Created %s", self.pidfile) |
| |
| def delpid(self): |
| logger.info("Removing %s", self.pidfile) |
| os.remove(self.pidfile) |
| |
| def start(self): |
| """Start the daemon.""" |
| |
| # Check for a pidfile to see if the daemon already runs |
| try: |
| with open(self.pidfile,'r') as pf: |
| |
| pid = int(pf.read().strip()) |
| except IOError: |
| pid = None |
| |
| if pid: |
| message = "pidfile {0} already exist. " + \ |
| "Daemon already running?\n" |
| sys.stderr.write(message.format(self.pidfile)) |
| sys.exit(1) |
| |
| # Start the daemon |
| self.daemonize() |
| self.run() |
| |
| def stop(self): |
| """Stop the daemon.""" |
| |
| # Get the pid from the pidfile |
| try: |
| with open(self.pidfile,'r') as pf: |
| pid = int(pf.read().strip()) |
| except IOError: |
| pid = None |
| |
| if not pid: |
| message = "pidfile {0} does not exist. " + \ |
| "Daemon not running?\n" |
| sys.stderr.write(message.format(self.pidfile)) |
| return # not an error in a restart |
| |
| # Try killing the daemon process |
| try: |
| # Try gentle stop first |
| os.kill(pid, signal.SIGINT) |
| time.sleep(0.2) |
| while 1: |
| os.kill(pid, signal.SIGTERM) |
| time.sleep(0.1) |
| except OSError as err: |
| e = str(err.args) |
| if e.find("No such process") > 0: |
| if os.path.exists(self.pidfile): |
| os.remove(self.pidfile) |
| else: |
| print (str(err.args)) |
| sys.exit(1) |
| |
| def restart(self): |
| """Restart the daemon.""" |
| self.stop() |
| self.start() |
| |
| def run(self): |
| """You should override this method when you subclass Daemon. |
| |
| It will be called after the process has been daemonized by |
| start() or restart().""" |
| |
| |
| |
| #################### |
| # Helper functions # |
| #################### |
| |
| |
| # read_chunk: iterator for reading chunks from the stream |
| # since this is all handled via urllib now, this is quite rudimentary |
| def read_chunk(req): |
| while True: |
| try: |
| line = req.readline().strip() |
| if line: |
| yield line |
| else: |
| print("No more lines?") |
| break |
| except Exception as info: |
| |
| break |
| return |
| |
| |
| ######################### |
| # Main listener program # |
| ######################### |
| |
| # PubSub class: handles connecting to a pubsub service and checking commits |
| class PubSubClient(Thread): |
| def __init__(self): |
| Thread.__init__(self) |
| self.setDaemon(True) |
| |
| def run(self): |
| logger.info("Watching %s", watching) |
| while True: |
| |
| self.req = None |
| while not self.req: |
| try: |
| if version == 3: |
| self.req = urllib.request.urlopen(self.url, None, 30) |
| else: |
| self.req = urllib2.urlopen(self.url, None, 30) |
| |
| logger.info("Connected to %s", self.url) |
| except: |
| logger.warn("Failed to connect to %s", self.url) |
| time.sleep(30) |
| continue |
| |
| for line in read_chunk(self.req): |
| if version == 3: |
| line = str( line, encoding='ascii' ).rstrip('\r\n,').replace('\x00','') # strip away any old pre-0.9 commas from gitpubsub chunks and \0 in svnpubsub chunks |
| else: |
| line = str( line ).rstrip('\r\n,').replace('\x00','') # strip away any old pre-0.9 commas from gitpubsub chunks and \0 in svnpubsub chunks |
| try: |
| obj = json.loads(line) |
| if "commit" in obj and "repository" in obj['commit']: |
| # If it's our public svn repo, then... |
| if obj['commit']['repository'] == "13f79535-47bb-0310-9956-ffa450edef68": |
| #Grab some vars |
| commit = obj['commit'] |
| # e.g. {"committer": "sebb", "log": "Ensure we exit on control+C", "repository": "13f79535-47bb-0310-9956-ffa450edef68", "format": 1, |
| # "changed": {"comdev/reporter.apache.org/trunk/scandist.py": {"flags": "U "}}, |
| # "date": "2015-07-13 13:38:33 +0000 (Mon, 13 Jul 2015)", "type": "svn", "id": 1690668} |
| svnuser = commit['committer'] |
| revision = commit['id'] |
| filePaths = set() |
| # e.g. {"comdev/reporter.apache.org/trunk/scandist.py": {"flags": "U "}} |
| for path in commit['changed']: |
| for watchPath in watching: |
| # Check if the commit is for our part of the repo |
| match = re.match("^%s" % watchPath, path) |
| if match: |
| filePath = str(watching[watchPath]) |
| if debug: |
| print("Matched '" + path + "' against '" + watchPath + "'; would run 'svn up " + filePath + "'") |
| filePaths.update(watching[watchPath]) |
| if filePaths: |
| for filePath in filePaths: |
| if debug: |
| print("Matched 'r" + str(revision) + "'; would run 'svn up " + filePath + "'") |
| else: |
| time.sleep(3) |
| logger.info("svn up %s", filePath) |
| subprocess.call(['svn','up', filePath]) |
| else: |
| logger.debug("Did not match 'r" + str(revision) + "' against ' " + str(watching.keys()) + "'") |
| if debug: |
| print("Did not match 'r" + str(revision) + "' against ' " + str(watching.keys()) + "'") |
| |
| |
| except ValueError as detail: |
| continue |
| |
| |
| |
| |
| |
| ################ |
| # Main program # |
| ################ |
| """ |
| According to https://svn.apache.org/repos/asf/subversion/trunk/tools/server-side/svnpubsub/svnpubsub/server.py |
| |
| # URLs are constructed from 3 parts: |
| # /${notification}/${optional_type}/${optional_repository} |
| # |
| # Notifications can be sent for commits or metadata (e.g., revprop) changes. |
| # If the type is included in the URL, you will only get notifications of that type. |
| # The type can be * and then you will receive notifications of any type. |
| # |
| # If the repository is included in the URL, you will only receive |
| # messages about that repository. The repository can be * and then you |
| # will receive messages about all repositories. |
| |
| """ |
| |
| def main(): |
| if debug: |
| print("Foreground test mode enabled, no updates will be made") |
| for watchPath in watching: |
| print("Watching: '" + watchPath + "' for updates to '" + str(watching[watchPath]) + "'") |
| |
| |
| |
| # Start the svn thread |
| svn_thread = PubSubClient() |
| svn_thread.url = "http://svn-master.apache.org:2069/commits/*" |
| svn_thread.start() |
| |
| while True: |
| try: |
| time.sleep(60) |
| except KeyboardInterrupt: |
| logger.info("Detected shutdown interrupt") |
| pass |
| |
| ############## |
| # Daemonizer # |
| ############## |
| class MyDaemon(daemon): |
| def run(self): |
| main() |
| |
| def usage(): |
| print("usage: %s start|stop|restart|foreground ([repo-path] [file-path])*" % sys.argv[0]) |
| print("for example: %s start comdev/projects.apache.org /var/www/projects.apache.org" % sys.argv[0]) |
| |
| def handle_args(args): |
| if len(args) % 2 != 0: |
| usage() |
| sys.exit("Need an even number of repo/file paths, found "+str(len(args))) |
| else: |
| for i in range(0, len(args), 2): |
| try: |
| watching[args[i]].update(args[i+1]) |
| except KeyError: |
| watching[args[i]] = set([args[i+1]]) |
| |
| if __name__ == "__main__": |
| logger = logging.getLogger('pubsubber') |
| logger.setLevel(logging.INFO) |
| formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s') |
| ch = logging.StreamHandler() |
| ch.setFormatter(formatter) |
| logger.addHandler(ch) |
| logfile = os.environ.get('LOGFILE') |
| logger.info("LOGFILE=%s", logfile) |
| daemon = MyDaemon('/tmp/pubsubber.pid', logfile) |
| if len(sys.argv) >= 2: |
| if 'start' == sys.argv[1]: |
| handle_args(sys.argv[2:]) |
| daemon.start() |
| elif 'stop' == sys.argv[1]: |
| daemon.stop() |
| elif 'restart' == sys.argv[1]: |
| daemon.restart() |
| elif 'foreground' == sys.argv[1]: |
| debug = True |
| handle_args(sys.argv[2:]) |
| main() |
| else: |
| usage() |
| sys.exit(2) |
| sys.exit(0) |
| else: |
| usage() |
| sys.exit(2) |
| |
