<commit_before>from twisted.python import log, usage
from twisted.internet.defer import inlineCallbacks, returnValue, Deferred
from twisted.internet import protocol, reactor
from txamqp.client import TwistedDelegate
from txamqp.content import Content
from txamqp.protocol import AMQClient
from vumi.errors import VumiError
import txamqp
import json
import sys

class Options(usage.Options):
    """
    Default options for all workers created
    """
    optParameters = [
        ["hostname", None, "127.0.0.1", "AMQP broker"],
        ["port", None, 5672, "AMQP port", int],
        ["username", None, "vumi", "AMQP username"],
        ["password", None, "<PASSWORD>", "AMQP password"],
        ["vhost", None, "/vumi", "AMQP virtual host"],
        ["specfile", None, "config/amqp-spec-0-8.xml", "AMQP spec file"],
    ]


class Worker(AMQClient):
    """
    The Worker is responsible for starting consumers & publishers
    as needed.
    """
    @inlineCallbacks
    def connectionMade(self):
        AMQClient.connectionMade(self)
        yield self.authenticate(self.factory.username, self.factory.password)
        # authentication was successful
        log.msg("Got an authenticated connection")
        yield self.startWorker()
    
    @inlineCallbacks
    def startWorker(self):
        # I hate camelCasing method but since Twisted has it as a
        # standard I voting to stick with it
        raise VumiError, "You need to subclass Worker and its " \
                             "startWorker method"
    
    @inlineCallbacks
    def get_channel(self, channel_id=None):
        """If channel_id is None a new channel is created"""
        if channel_id:
            channel = self.channels[channel_id]
        else:
            channel_id = self.get_new_channel_id()
            channel = yield self.channel(channel_id)
            yield channel.channel_open()
            self.channels[channel_id] = channel
        returnValue(channel)
    
    def get_new_channel_id(self):
        """
        AMQClient keeps track of channels in a dictionary. The
        channel ids are the keys, get the highest number and up it
        or just return zero for the first channel
        """
        return (max(self.channels) + 1) if self.channels else 0
    
    @inlineCallbacks
    def start_consumer(self, klass, *args, **kwargs):
        channel = yield self.get_channel()
        consumer = klass(*args, **kwargs)
        
        # get the details for AMQP
        exchange_name = consumer.exchange_name
        exchange_type = consumer.exchange_type
        durable = consumer.durable
        queue_name = consumer.queue_name
        routing_key = consumer.routing_key
        
        # declare the exchange, doesn't matter if it already exists
        yield channel.exchange_declare(exchange=exchange_name,
                                        type=exchange_type, durable=durable)
                                                    
        # declare the queue
        yield channel.queue_declare(queue=queue_name, durable=durable)
        # bind it to the exchange with the routing key
        yield channel.queue_bind(queue=queue_name, exchange=exchange_name, 
                                    routing_key=routing_key)
        # register the consumer
        reply = yield channel.basic_consume(queue=queue_name)
        queue = yield self.queue(reply.consumer_tag)
        # start consuming! nom nom nom
        consumer.start(channel, queue)
        # return the newly created & consuming consumer
        returnValue(consumer)
    
    @inlineCallbacks
    def start_publisher(self, klass, *args, **kwargs):
        # much more braindead than start_consumer
        # get a channel
        channel = yield self.get_channel()
        # start the publisher
        publisher = klass(*args, **kwargs)
        # start!
        yield publisher.start(channel)
        # return the publisher
        returnValue(publisher)
    

class Consumer(object):
    
    exchange_name = "vumi"
    exchange_type = "direct"
    durable = False
    
    queue_name = "queue"
    routing_key = "routing_key"
    
    @inlineCallbacks
    def start(self, channel, queue):
        self.channel = channel
        self.queue = queue
        self.keep_consuming = True
        @inlineCallbacks
        def read_messages():
            log.msg("Consumer starting...")
            try:
                while self.keep_consuming:
                    message = yield self.queue.get()
                    self.consume(message)
            except txamqp.queue.Closed, e:
                log.err("Queue has closed", e)
        read_messages()
        yield None
        returnValue(self)
    
    def consume(self, message):
        if self.keep_consuming:
            self.consume_json(json.loads(message.content.body))
        if self.keep_consuming:
            self.ack(message)
    
    def consume_json(self, dictionary):
        "helper method"
        log.msg("Received dict: %s" % dictionary)
    
    def ack(self, message):
        self.channel.basic_ack(message.delivery_tag, True)
    
    @inlineCallbacks
    def stop(self):
        self.keep_consuming = False
    

class Publisher(object):
    exchange_name = "vumi"
    exchange_type = "direct"<commit_msg>ack messages only if handled by smpp<commit_after>from twisted.python import log, usage
from twisted.internet.defer import inlineCallbacks, returnValue, Deferred
from twisted.internet import protocol, reactor
from txamqp.client import TwistedDelegate
from txamqp.content import Content
from txamqp.protocol import AMQClient
from vumi.errors import VumiError
import txamqp
import json
import sys

class Options(usage.Options):
    """
    Default options for all workers created
    """
    optParameters = [
        ["hostname", None, "127.0.0.1", "AMQP broker"],
        ["port", None, 5672, "AMQP port", int],
        ["username", None, "vumi", "AMQP username"],
        ["password", None, "<PASSWORD>", "AMQP password"],
        ["vhost", None, "/vumi", "AMQP virtual host"],
        ["specfile", None, "config/amqp-spec-0-8.xml", "AMQP spec file"],
    ]


class Worker(AMQClient):
    """
    The Worker is responsible for starting consumers & publishers
    as needed.
    """
    @inlineCallbacks
    def connectionMade(self):
        AMQClient.connectionMade(self)
        yield self.authenticate(self.factory.username, self.factory.password)
        # authentication was successful
        log.msg("Got an authenticated connection")
        yield self.startWorker()
    
    @inlineCallbacks
    def startWorker(self):
        # I hate camelCasing method but since Twisted has it as a
        # standard I voting to stick with it
        raise VumiError, "You need to subclass Worker and its " \
                             "startWorker method"
    
    @inlineCallbacks
    def get_channel(self, channel_id=None):
        """If channel_id is None a new channel is created"""
        if channel_id:
            channel = self.channels[channel_id]
        else:
            channel_id = self.get_new_channel_id()
            channel = yield self.channel(channel_id)
            yield channel.channel_open()
            self.channels[channel_id] = channel
        returnValue(channel)
    
    def get_new_channel_id(self):
        """
        AMQClient keeps track of channels in a dictionary. The
        channel ids are the keys, get the highest number and up it
        or just return zero for the first channel
        """
        return (max(self.channels) + 1) if self.channels else 0
    
    @inlineCallbacks
    def start_consumer(self, klass, *args, **kwargs):
        channel = yield self.get_channel()
        consumer = klass(*args, **kwargs)
        
        # get the details for AMQP
        exchange_name = consumer.exchange_name
        exchange_type = consumer.exchange_type
        durable = consumer.durable
        queue_name = consumer.queue_name
        routing_key = consumer.routing_key
        
        # declare the exchange, doesn't matter if it already exists
        yield channel.exchange_declare(exchange=exchange_name,
                                        type=exchange_type, durable=durable)
                                                    
        # declare the queue
        yield channel.queue_declare(queue=queue_name, durable=durable)
        # bind it to the exchange with the routing key
        yield channel.queue_bind(queue=queue_name, exchange=exchange_name, 
                                    routing_key=routing_key)
        # register the consumer
        reply = yield channel.basic_consume(queue=queue_name)
        queue = yield self.queue(reply.consumer_tag)
        # start consuming! nom nom nom
        consumer.start(channel, queue)
        # return the newly created & consuming consumer
        returnValue(consumer)
    
    @inlineCallbacks
    def start_publisher(self, klass, *args, **kwargs):
        # much more braindead than start_consumer
        # get a channel
        channel = yield self.get_channel()
        # start the publisher
        publisher = klass(*args, **kwargs)
        # start!
        yield publisher.start(channel)
        # return the publisher
        returnValue(publisher)
    

class Consumer(object):
    
    exchange_name = "vumi"
    exchange_type = "direct"
    durable = False
    
    queue_name = "queue"
    routing_key = "routing_key"
    
    @inlineCallbacks
    def start(self, channel, queue):
        self.channel = channel
        self.queue = queue
        self.keep_consuming = True
        @inlineCallbacks
        def read_messages():
            log.msg("Consumer starting...")
            try:
                while self.keep_consuming:
                    message = yield self.queue.get()
                    self.consume(message)
            except txamqp.queue.Closed, e:
                log.err("Queue has closed", e)
        read_messages()
        yield None
        returnValue(self)
    
    def consume(self, message):
        self.consume_json(json.loads(message.content.body))
        self.ack(message)
    
    def consume_json(self, dictionary):
        "helper method"
        log.msg("Received dict: %s" % dictionary)
    
    def ack(self, message):
        self.channel.basic_ack(message.delivery_tag, True)
    
    @inlineCallbacks
    def stop(self):
        self.keep_consuming = False
        returnValue(self.keep_consuming)
    

class Publisher(object):
    exchange_name = "vumi"
    exchange_type = "direct"
{'fraction_non_alphanumeric': 0.08432934926958831, 'fraction_numerical': 0.0026560424966799467, 'mean_word_length': 2.197755535335153, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3667523', 'n_tokens_mistral': 2882, 'n_tokens_neox': 2796, 'n_words': 798}