<commit_before>#!/usr/bin/env python
import sys
import argparse
from os.path import dirname
import gevent
from gevent import monkey
monkey.patch_all()
from socketio.server import SocketIOServer
from socketio import socketio_manage
from kombu import Connection
import logging

sys.path.append(dirname(dirname(dirname(__file__))))

from treeherder.events.consumer import EventsConsumer
from treeherder.events.sockets import EventsNamespace

logger = logging.getLogger("treeherder.events")

class Application(object):
    """wsgi application with socketio enabled"""

    def __init__(self):
        self.buffer = []

    def __call__(self, environ, start_response):
        path = environ['PATH_INFO'].strip('/') or 'index.html'

        if path.startswith("socket.io"):
            socketio_manage(environ, {'/events': EventsNamespace})
        else:
            return not_found(start_response)


def not_found(start_response):
    start_response('404 Not Found', [])
    return ['<h1>Not Found</h1>']


def broadcast_subscribers(body, msg):
    """
    This is the main function where all the magic happens
    It broadcasts the events to the clients subscribed to
    them.
    """
    pkt = dict(type="event", name=body['event'],
               args=body, endpoint='/events')

    logger.log(logging.DEBUG, "emitting event {0} on branch {1}".format(
        body["event"], body["branch"]
    ))

    for session_id, socket in server.sockets.iteritems():
        # loop over all the open connections
        # and send a message when needed
        if "subscriptions" not in socket.session:
            continue

        for branch, events in socket.session['subscriptions'].items():
            if branch == body["branch"] or branch == "*":
                if body["event"] in events or "*" in events:
                    logger.log(logging.DEBUG, "sending packet {0} to {1}".format(
                        pkt, session_id
                    ))
                    socket.send_packet(pkt)
                    break
    msg.ack()


def start_consumer(broker_url):
    with Connection(broker_url) as conn:
        consumer = EventsConsumer(conn)
        consumer.listen_to("events.#", broadcast_subscribers)
        consumer.run()


if __name__ == "__main__":

    parser = argparse.ArgumentParser()
    parser.add_argument("--host",
                        help="interface to bind the server to",
                        default="0.0.0.0")
    parser.add_argument("--port",
                        help="port to bind the server to",
                        default="8005",
                        type=int)
    parser.add_argument("--broker-url",
                        help="url of the broker to use",
                        required=True)
    parser.add_argument('--log-file',
                        default="stdout",
                        help="""the file where the log should be written to.
Default to stdout""")
    parser.add_argument("--log-level",
                        help="minimum level to log",
                        default="DEBUG",
                        choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"])
    args = parser.parse_args()

    # logging system setup
    logger.setLevel(args.log_level)

    if not args.log_file:
        args.log_file = "stdout"
    if args.log_file in ["stdout", "stderr"]:
        log_handler = logging.StreamHandler(getattr(sys, args.log_file))
    else:
        log_handler = logging.FileHandler(args.log_file)
    log_handler.setLevel(args.log_level)
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    log_handler.setFormatter(formatter)
    logger.addHandler(log_handler)

    try:
        logger.info("Starting SocketIOServer")

        server = SocketIOServer((args.host, args.port), Application(),
                                resource="socket.io", policy_server=False)
        logger.log(logging.INFO, "Listening to http://{0}:{1}".format(args.host, args.port))
        logger.log(logging.DEBUG, "writing logs to %s" % args.log_file)
        gevent.spawn(start_consumer, args.broker_url)
        server.serve_forever()
    except KeyboardInterrupt:
        logger.log(logging.INFO, "Socketio server stopped")
        for handler in logger.handlers:
            try:
                handler.close()
            except AttributeError:
                pass<commit_msg>fixup! trying to make the socketio logging work on py26<commit_after>#!/usr/bin/env python
import sys
import argparse
from os.path import dirname
import gevent
from gevent import monkey
monkey.patch_all()
from socketio.server import SocketIOServer
from socketio import socketio_manage
from kombu import Connection
import logging

logger = logging.getLogger("treeherder.events")

sys.path.append(dirname(dirname(dirname(__file__))))

from treeherder.events.consumer import EventsConsumer
from treeherder.events.sockets import EventsNamespace


class Application(object):
    """wsgi application with socketio enabled"""

    def __init__(self):
        self.buffer = []

    def __call__(self, environ, start_response):
        path = environ['PATH_INFO'].strip('/') or 'index.html'

        if path.startswith("socket.io"):
            socketio_manage(environ, {'/events': EventsNamespace})
        else:
            return not_found(start_response)


def not_found(start_response):
    start_response('404 Not Found', [])
    return ['<h1>Not Found</h1>']


def broadcast_subscribers(body, msg):
    """
    This is the main function where all the magic happens
    It broadcasts the events to the clients subscribed to
    them.
    """
    pkt = dict(type="event", name=body['event'],
               args=body, endpoint='/events')

    logger.log(logging.DEBUG, "emitting event {0} on branch {1}".format(
        body["event"], body["branch"]
    ))

    for session_id, socket in server.sockets.iteritems():
        # loop over all the open connections
        # and send a message when needed
        if "subscriptions" not in socket.session:
            continue

        for branch, events in socket.session['subscriptions'].items():
            if branch == body["branch"] or branch == "*":
                if body["event"] in events or "*" in events:
                    logger.log(logging.DEBUG, "sending packet {0} to {1}".format(
                        pkt, session_id
                    ))
                    socket.send_packet(pkt)
                    break
    msg.ack()


def start_consumer(broker_url):
    with Connection(broker_url) as conn:
        consumer = EventsConsumer(conn)
        consumer.listen_to("events.#", broadcast_subscribers)
        consumer.run()


if __name__ == "__main__":

    parser = argparse.ArgumentParser()
    parser.add_argument("--host",
                        help="interface to bind the server to",
                        default="0.0.0.0")
    parser.add_argument("--port",
                        help="port to bind the server to",
                        default="8005",
                        type=int)
    parser.add_argument("--broker-url",
                        help="url of the broker to use",
                        required=True)
    parser.add_argument('--log-file',
                        default="stdout",
                        help="""the file where the log should be written to.
Default to stdout""")
    parser.add_argument("--log-level",
                        help="minimum level to log",
                        default="DEBUG",
                        choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"])
    args = parser.parse_args()

    # logging system setup
    logging.basicConfig(level=args.log_level)

    if not args.log_file:
        args.log_file = "stdout"
    if args.log_file in ["stdout", "stderr"]:
        log_handler = logging.StreamHandler(getattr(sys, args.log_file))
    else:
        log_handler = logging.FileHandler(args.log_file)
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    log_handler.setFormatter(formatter)
    logger.addHandler(log_handler)

    try:
        logger.info("Starting SocketIOServer")

        server = SocketIOServer((args.host, args.port), Application(),
                                resource="socket.io", policy_server=False)
        logger.log(logging.INFO, "Listening to http://{0}:{1}".format(args.host, args.port))
        logger.log(logging.DEBUG, "writing logs to %s" % args.log_file)
        gevent.spawn(start_consumer, args.broker_url)
        server.serve_forever()
    except KeyboardInterrupt:
        logger.log(logging.INFO, "Socketio server stopped")
        for handler in logger.handlers:
            try:
                handler.close()
            except AttributeError:
                pass
{'fraction_non_alphanumeric': 0.11422070534698521, 'fraction_numerical': 0.004550625711035267, 'mean_word_length': 2.297449362340585, 'pattern_counts': {'":': 4, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6324636', 'n_tokens_mistral': 2477, 'n_tokens_neox': 2388, 'n_words': 603}