<commit_before>"""
Created by <NAME> <<EMAIL>>

This module encapsulates RabbitMQ messaging functionality and
provides a set of sync. and async. methods for topic-based
communication.
"""

# TODO: Add API for synchronous request / reply calls
# TODO: Add RMQ ack mechanism (cf: http://pika.readthedocs.org/en/latest/examples/asynchronous_publisher_example.html)

import pika
import logging
import threading
import time
import uuid
import os
import json
logging.basicConfig(level=logging.INFO)
logging.getLogger('pika').setLevel(logging.ERROR)
LOG = logging.getLogger("son-mano-base:messaging")
LOG.setLevel(logging.DEBUG)

# if we don't find a broker configuration in our ENV, we use this URL as default
RABBITMQ_URL_FALLBACK = "amqp://guest:guest@localhost:5672/%2F"
# if we don't find a broker configuration in our ENV, we use this exchange as default
RABBITMQ_EXCHANGE_FALLBACK = "son-kernel"


class ManoBrokerConnection(object):
    """
    This class encapsulates a bare RabbitMQ connection setup.
    It provides helper methods to easily publish/subscribe to a given topic.
    It uses the asynchronous adapter implementation of the pika library.
    """

    def __init__(self, app_id, blocking=False):
        self.app_id = app_id

        # do configuration
        self._configs = {}
        self._configs["broker"] = self._load_broker_configuration()
        self.rabbitmq_url = self._configs["broker"]["broker_url"]
        self.rabbitmq_exchange = self._configs["broker"]["exchange"]
        self.rabbitmq_exchange_type = "topic"

        # create additional members
        self._connection = None
        self._channel = None
        self.base_queue = "%s.%s" % (self.rabbitmq_exchange, self.app_id)
        self._connected = False
        self._closing = False
        self._queue_setup_lock = threading.Event()
        # trigger connection setup (without blocking)
        self.setup_connection()

    def _load_broker_configuration(self):
        """
        Tries to get broker configuration from ENV variables.
        Uses fallback values if nothing is found.
        :return: dictionary
        """
        broker_host = os.environ.get("broker_host")
        broker_exchange = os.environ.get("broker_exchange")
        if broker_host is None:
            broker_host = RABBITMQ_URL_FALLBACK
        if broker_exchange is None:
            broker_exchange = RABBITMQ_EXCHANGE_FALLBACK
        bc = dict(broker_url=broker_host, exchange=broker_exchange)
        LOG.info("Broker config: %r" % bc)
        return bc

    def setup_connection(self, blocking=False):
        """
        Setup RabbitMQ connection, channel, exchange etc.
        :param blocking: do not run IO loop in separated thread
        :return: connection
        """
        # setup RabbitMQ connection
        self._connection = self._connect()

        def connection_thread():
            # run connection IO loop
            try:
                self._connection.ioloop.start()
            except Exception as e:
                LOG.exception("Connection lost.")

        if blocking:
            connection_thread()
            return self._connection

        t = threading.Thread(target=connection_thread, args=())
        t.daemon = True
        t.start()

        # FIXME: quick-hack! Basic connection setup should behave synchronous.
        while not self._connected:
            time.sleep(0.001)
        return self._connection

    def stop_connection(self):
        self._closing = True
        self._connection.close()

    def _connect(self):
        # connect to RabbitMQ
        LOG.info("Connecting to RabbitMQ on %r...", self.rabbitmq_url)
        return pika.SelectConnection(parameters=pika.URLParameters(self.rabbitmq_url),
                                     on_open_callback=self._on_connection_open,
                                     on_close_callback=self._on_connection_closed,
                                     on_open_error_callback=self._on_connection_error,
                                     stop_ioloop_on_close=False)

    def _reconnect(self):
        if self._connection is not None:
            self._connection.ioloop.stop()
        if not self._closing:
            self.setup_connection()

    def _setup_channel(self):
        LOG.info("Creating a new channel...")
        self._connection.channel(on_open_callback=self._on_channel_open)

    def _setup_exchange(self):
        LOG.info("Declaring exchange %r...", self.rabbitmq_exchange)<commit_msg>Improved stop_connection() method to get rid of random test behavior.<commit_after>"""
Created by <NAME> <<EMAIL>>

This module encapsulates RabbitMQ messaging functionality and
provides a set of sync. and async. methods for topic-based
communication.
"""

# TODO: Add API for synchronous request / reply calls
# TODO: Add RMQ ack mechanism (cf: http://pika.readthedocs.org/en/latest/examples/asynchronous_publisher_example.html)

import pika
import logging
import threading
import time
import uuid
import os
import json
logging.basicConfig(level=logging.INFO)
logging.getLogger('pika').setLevel(logging.ERROR)
LOG = logging.getLogger("son-mano-base:messaging")
LOG.setLevel(logging.DEBUG)

# if we don't find a broker configuration in our ENV, we use this URL as default
RABBITMQ_URL_FALLBACK = "amqp://guest:guest@localhost:5672/%2F"
# if we don't find a broker configuration in our ENV, we use this exchange as default
RABBITMQ_EXCHANGE_FALLBACK = "son-kernel"


class ManoBrokerConnection(object):
    """
    This class encapsulates a bare RabbitMQ connection setup.
    It provides helper methods to easily publish/subscribe to a given topic.
    It uses the asynchronous adapter implementation of the pika library.
    """

    def __init__(self, app_id, blocking=False):
        self.app_id = app_id

        # do configuration
        self._configs = {}
        self._configs["broker"] = self._load_broker_configuration()
        self.rabbitmq_url = self._configs["broker"]["broker_url"]
        self.rabbitmq_exchange = self._configs["broker"]["exchange"]
        self.rabbitmq_exchange_type = "topic"

        # create additional members
        self._connection = None
        self._channel = None
        self.base_queue = "%s.%s" % (self.rabbitmq_exchange, self.app_id)
        self._connected = False
        self._closing = False
        self._queue_setup_lock = threading.Event()
        # trigger connection setup (without blocking)
        self.setup_connection()

    def _load_broker_configuration(self):
        """
        Tries to get broker configuration from ENV variables.
        Uses fallback values if nothing is found.
        :return: dictionary
        """
        broker_host = os.environ.get("broker_host")
        broker_exchange = os.environ.get("broker_exchange")
        if broker_host is None:
            broker_host = RABBITMQ_URL_FALLBACK
        if broker_exchange is None:
            broker_exchange = RABBITMQ_EXCHANGE_FALLBACK
        bc = dict(broker_url=broker_host, exchange=broker_exchange)
        LOG.info("Broker config: %r" % bc)
        return bc

    def setup_connection(self, blocking=False):
        """
        Setup RabbitMQ connection, channel, exchange etc.
        :param blocking: do not run IO loop in separated thread
        :return: connection
        """
        # setup RabbitMQ connection
        self._connection = self._connect()

        def connection_thread():
            # run connection IO loop
            try:
                self._connection.ioloop.start()
            except Exception as e:
                LOG.exception("Connection lost.")

        if blocking:
            connection_thread()
            return self._connection

        t = threading.Thread(target=connection_thread, args=())
        t.daemon = True
        t.start()

        # FIXME: quick-hack! Basic connection setup should behave synchronous.
        while not self._connected:
            time.sleep(0.001)
        return self._connection

    def stop_connection(self):
        self._closing = True
        try:
            self._connection.close()
        except pika.exceptions.ConnectionClosed:
            pass  # get rid of send_frame error on connection close

    def _connect(self):
        # connect to RabbitMQ
        LOG.info("Connecting to RabbitMQ on %r...", self.rabbitmq_url)
        return pika.SelectConnection(parameters=pika.URLParameters(self.rabbitmq_url),
                                     on_open_callback=self._on_connection_open,
                                     on_close_callback=self._on_connection_closed,
                                     on_open_error_callback=self._on_connection_error,
                                     stop_ioloop_on_close=False)

    def _reconnect(self):
        if self._connection is not None:
            self._connection.ioloop.stop()
        if not self._closing:
            self.setup_connection()

    def _setup_channel(self):
        LOG.info("Creating a new channel...")
        self._connection.channel(on_open_callback=self._on_channel_open)

    def _setup_exchange(self):
        LOG.info("Declaring exchange %r...", self.rabbitmq_exchange)
{'fraction_non_alphanumeric': 0.08455001089561996, 'fraction_numerical': 0.0019612115929396384, 'mean_word_length': 2.531742978068488, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '807739', 'n_tokens_mistral': 2611, 'n_tokens_neox': 2473, 'n_words': 660}