<commit_before># vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""AMQP-based RPC.

Queues have consumers and publishers.

No fan-out support yet.

"""

import json
import sys
import time
import traceback
import types
import uuid

from carrot import connection as carrot_connection
from carrot import messaging
import eventlet
from eventlet import greenpool
from eventlet import pools
from eventlet import queue
import greenlet

from nova import context
from nova import exception
from nova import fakerabbit
from nova import flags
from nova.rpc.common import RemoteError, LOG

# Needed for tests
eventlet.monkey_patch()

FLAGS = flags.FLAGS


class Connection(carrot_connection.BrokerConnection):
    """Connection instance object."""

    def __init__(self, *args, **kwargs):
        super(Connection, self).__init__(*args, **kwargs)
        self._rpc_consumers = []
        self._rpc_consumer_thread = None

    @classmethod
    def instance(cls, new=True):
        """Returns the instance."""
        if new or not hasattr(cls, '_instance'):
            params = dict(hostname=FLAGS.rabbit_host,
                          port=FLAGS.rabbit_port,
                          ssl=FLAGS.rabbit_use_ssl,
                          userid=FLAGS.rabbit_userid,
                          password=<PASSWORD>.<PASSWORD>_password,
                          virtual_host=FLAGS.rabbit_virtual_host)

            if FLAGS.fake_rabbit:
                params['backend_cls'] = fakerabbit.Backend

            # NOTE(vish): magic is fun!
            # pylint: disable=W0142
            if new:
                return cls(**params)
            else:
                cls._instance = cls(**params)
        return cls._instance

    @classmethod
    def recreate(cls):
        """Recreates the connection instance.

        This is necessary to recover from some network errors/disconnects.

        """
        try:
            del cls._instance
        except AttributeError, e:
            # The _instance stuff is for testing purposes. Usually we don't use
            # it. So don't freak out if it doesn't exist.
            pass
        return cls.instance()

    def close(self):
        self.cancel_consumer_thread()
        for consumer in self._rpc_consumers:
            try:
                consumer.close()
            except Exception:
                # ignore all errors
                pass
        self._rpc_consumers = []
        super(Connection, self).close()

    def consume_in_thread(self):
        """Consumer from all queues/consumers in a greenthread"""

        consumer_set = ConsumerSet(connection=self,
                consumer_list=self._rpc_consumers)

        def _consumer_thread():
            try:
                consumer_set.wait()
            except greenlet.GreenletExit:
                return
        if self._rpc_consumer_thread is None:
            self._rpc_consumer_thread = eventlet.spawn(_consumer_thread)
        return self._rpc_consumer_thread

    def cancel_consumer_thread(self):
        """Cancel a consumer thread"""
        if self._rpc_consumer_thread is not None:
            self._rpc_consumer_thread.kill()
            try:
                self._rpc_consumer_thread.wait()
            except greenlet.GreenletExit:
                pass
            self._rpc_consumer_thread = None

    def create_consumer(self, topic, proxy, fanout=False):
        """Create a consumer that calls methods in the proxy"""
        if fanout:
            consumer = FanoutAdapterConsumer(
                    connection=self,
                    topic=topic,
                    proxy=proxy)
        else:
            consumer = TopicAdapterConsumer(
                    connection=self,
                    topic=topic,
                    proxy=proxy)
        self._rpc_consumers.append(consumer)


class Pool(pools.Pool):
    """Class that implements a Pool of Connections."""

    # TODO(comstud): Timeout connections not used in a while
    def create(self):
        LOG.debug('Creating new connection')
        return Connection.instance(new=True)

# Create a ConnectionPool to use for RPC calls.  We'll order the
# pool as a stack (LIFO), so that we can potentially loop through and
# timeout old unused connections at some point
ConnectionPool = Pool(
        max_size=FLAGS.rpc_conn_pool_size,
        order_as_stack=True)


class Consumer(messaging.Consumer):
    """Consumer base class.

    Contains methods for connecting the fetch method to async loops.

    """

    def __init__(self, *args, **kwargs):
        connection = kwargs.get('connection')
        max_retries = FLAGS.rabbit_max_retries
        sleep_time = FLAGS.rabbit_retry_interval
        tries = 0
        while True:
            tries += 1
            if tries > 1:
                time.sleep(sleep_time)
                # backoff for next retry attempt.. if there is one
                sleep_time += FLAGS.rabbit_retry_backoff
                if sleep_time > 30:
                    sleep_time = 30
            try:
                super(Consumer, self).__init__(*args, **kwargs)
                self.failed_connection = False
                break
            except Exception as e:  # Catching all because carrot sucks
                self.failed_connection = True
                if max_retries > 0 and tries == max_retries:
                    break<commit_msg>remove unneeded connection= in carrot Consumer init<commit_after># vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""AMQP-based RPC.

Queues have consumers and publishers.

No fan-out support yet.

"""

import json
import sys
import time
import traceback
import types
import uuid

from carrot import connection as carrot_connection
from carrot import messaging
import eventlet
from eventlet import greenpool
from eventlet import pools
from eventlet import queue
import greenlet

from nova import context
from nova import exception
from nova import fakerabbit
from nova import flags
from nova.rpc.common import RemoteError, LOG

# Needed for tests
eventlet.monkey_patch()

FLAGS = flags.FLAGS


class Connection(carrot_connection.BrokerConnection):
    """Connection instance object."""

    def __init__(self, *args, **kwargs):
        super(Connection, self).__init__(*args, **kwargs)
        self._rpc_consumers = []
        self._rpc_consumer_thread = None

    @classmethod
    def instance(cls, new=True):
        """Returns the instance."""
        if new or not hasattr(cls, '_instance'):
            params = dict(hostname=FLAGS.rabbit_host,
                          port=FLAGS.rabbit_port,
                          ssl=FLAGS.rabbit_use_ssl,
                          userid=FLAGS.rabbit_userid,
                          password=FLAGS.rabbit_password,
                          virtual_host=FLAGS.rabbit_virtual_host)

            if FLAGS.fake_rabbit:
                params['backend_cls'] = fakerabbit.Backend

            # NOTE(vish): magic is fun!
            # pylint: disable=W0142
            if new:
                return cls(**params)
            else:
                cls._instance = cls(**params)
        return cls._instance

    @classmethod
    def recreate(cls):
        """Recreates the connection instance.

        This is necessary to recover from some network errors/disconnects.

        """
        try:
            del cls._instance
        except AttributeError, e:
            # The _instance stuff is for testing purposes. Usually we don't use
            # it. So don't freak out if it doesn't exist.
            pass
        return cls.instance()

    def close(self):
        self.cancel_consumer_thread()
        for consumer in self._rpc_consumers:
            try:
                consumer.close()
            except Exception:
                # ignore all errors
                pass
        self._rpc_consumers = []
        super(Connection, self).close()

    def consume_in_thread(self):
        """Consumer from all queues/consumers in a greenthread"""

        consumer_set = ConsumerSet(connection=self,
                consumer_list=self._rpc_consumers)

        def _consumer_thread():
            try:
                consumer_set.wait()
            except greenlet.GreenletExit:
                return
        if self._rpc_consumer_thread is None:
            self._rpc_consumer_thread = eventlet.spawn(_consumer_thread)
        return self._rpc_consumer_thread

    def cancel_consumer_thread(self):
        """Cancel a consumer thread"""
        if self._rpc_consumer_thread is not None:
            self._rpc_consumer_thread.kill()
            try:
                self._rpc_consumer_thread.wait()
            except greenlet.GreenletExit:
                pass
            self._rpc_consumer_thread = None

    def create_consumer(self, topic, proxy, fanout=False):
        """Create a consumer that calls methods in the proxy"""
        if fanout:
            consumer = FanoutAdapterConsumer(
                    connection=self,
                    topic=topic,
                    proxy=proxy)
        else:
            consumer = TopicAdapterConsumer(
                    connection=self,
                    topic=topic,
                    proxy=proxy)
        self._rpc_consumers.append(consumer)


class Pool(pools.Pool):
    """Class that implements a Pool of Connections."""

    # TODO(comstud): Timeout connections not used in a while
    def create(self):
        LOG.debug('Creating new connection')
        return Connection.instance(new=True)

# Create a ConnectionPool to use for RPC calls.  We'll order the
# pool as a stack (LIFO), so that we can potentially loop through and
# timeout old unused connections at some point
ConnectionPool = Pool(
        max_size=FLAGS.rpc_conn_pool_size,
        order_as_stack=True)


class Consumer(messaging.Consumer):
    """Consumer base class.

    Contains methods for connecting the fetch method to async loops.

    """

    def __init__(self, *args, **kwargs):
        max_retries = FLAGS.rabbit_max_retries
        sleep_time = FLAGS.rabbit_retry_interval
        tries = 0
        while True:
            tries += 1
            if tries > 1:
                time.sleep(sleep_time)
                # backoff for next retry attempt.. if there is one
                sleep_time += FLAGS.rabbit_retry_backoff
                if sleep_time > 30:
                    sleep_time = 30
            try:
                super(Consumer, self).__init__(*args, **kwargs)
                self.failed_connection = False
                break
            except Exception as e:  # Catching all because carrot sucks
                self.failed_connection = True
                if max_retries > 0 and tries == max_retries:
                    break
{'fraction_non_alphanumeric': 0.07015672437843604, 'fraction_numerical': 0.0037745138262082548, 'mean_word_length': 2.0754478930103457, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, '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': '320567', 'n_tokens_mistral': 3246, 'n_tokens_neox': 3038, 'n_words': 1013}