def connect(self):
     """
     Connect to the broker.
     @return: The AMQP connection object.
     @rtype: I{Connection}
     """
     self.__lock()
     try:
         if self.connection is None:
             url = self.url.simple()
             transport = self.url.transport
             log.info('connecting:\n%s', self)
             con = Connection(
                 url=url,
                 tcp_nodelay=True,
                 reconnect=True,
                 transport=transport)
             con.attach()
             log.info('{%s} connected to AMQP', self.id())
             self.connection = con
         else:
             con = self.connection
         return con
     finally:
         self.__unlock()

    def get_qpid_connection(self):
        """Connect to a broker, set and  return the connection object

        Authenticates (if necessary), and sets bkr.common._connection and
        returns it. This method is thread safe.

        """
        self.connection_lock.acquire()
        try:
            global can_use_qpid, _connection
            if not can_use_qpid:
                global qpid_import_error
                raise ImportError(str(qpid_import_error))
            if _connection is None or _connection.get_error():
                connection_params = [[self._broker],
                                     {'reconnect': self._reconnect,
                                      'heartbeat': self._heartbeat_timeout}]
                if self.krb_auth:
                    connection_params[1].update({'sasl_mechanisms' : 'GSSAPI'})
                    # As connections can recover from krb errors, we don't need
                    # to worry about doing this manually.
                    self.do_krb_auth()
                _connection = Connection(*connection_params[0], **connection_params[1])
                _connection.open()
            return _connection
        finally:
            self.connection_lock.release()

class FreshDocWriter(PageWriterBase):
  def _initialize(self):
    self.set_name('FreshDocWriter')
    if get_project_settings()['DEBUG_MODE']:
      self._push_message = self._push_message_debug
    else:
      self._create_client()

  def _create_client(self):
    try:
      self.connection_ = Connection(url='10.100.151.13:5672', heartbeat=4, reconnect=True,
                                    reconnect_limit=10, reconnect_interval=4)
      self.connection_.open()
      self.sender_ = self.connection_.session().sender('leso.exchange.fresh_video')
    except:
      self.connection_ = None
      self.logger_.exception('failed to connect to message queue server.')

  def finalize(self):
    if self.connection_:
      self.connection_.close()
    PageWriterBase.finalize(self)

  def _push_message_debug(self, doc):
    pass

  def _push_message(self, doc):
    doc.video.doc_id = doc.id
    doc.video.id = str(doc.id)
    doc.video.crawl_time = doc.video.create_time = doc.crawl_time
    doc.video.discover_time = doc.discover_time
    doc.video.url = doc.url
    doc.video.domain = doc.domain
    doc.video.domain_id = doc.domain_id
    doc.video.in_links = doc.in_links
    msg_body = thrift_to_str(doc.video)
    if not msg_body:
      return
    self.sender_.send(Message(content=doc.url + '\t' + base64.b64encode(msg_body), durable=True))
    self.logger_.info('send message successfully, %s', doc.url)

  def process_item(self, item):
    if not item:
      return
    doc = item.to_crawldoc()
    if doc.doc_type != CrawlDocType.PAGE_TIME:
      return
    try:
      self._push_message(doc)
    except:
      self.logger_.exception('failed to send message, %s', doc.url)

    def start(self):
        """
        Enable AMQP queueing. This method puts up the event processor and
        sets it to "active".
        """
        self.log.debug("enabling AMQP queueing")

        # Evaluate username
        user = self.env.config.get("amqp.id", default=None)
        if not user:
            user = self.env.uuid

        # Create initial broker connection
        url = "%s:%s" % (self.url['host'], self.url['port'])
        self._conn = Connection.establish(url, reconnect=self.reconnect,
            username=user,
            password=self.env.config.get("amqp.key"),
            transport=self.url['transport'],
            reconnect_interval=self.reconnect_interval,
            reconnect_limit=self.reconnect_limit)

        # Do automatic broker failover if requested
        if self.env.config.get('amqp.failover', default=False):
            auto_fetch_reconnect_urls(self._conn)

        # Create event provider
        self._eventProvider = EventProvider(self.env, self._conn)

 def testConnectError(self):
   try:
     self.conn = Connection.open("localhost", 0)
     assert False, "connect succeeded"
   except ConnectError, e:
     # XXX: should verify that e includes appropriate diagnostic info
     pass

    def checkAuth(self, user, password):
        """
        This function checks a username / password combination using
        the AMQP service' SASL configuration.

        =============== ============
        Parameter       Description
        =============== ============
        user            Username
        password        Password
        =============== ============

        ``Return:`` Bool, success or failure
        """
        # Strip username/password parts of url
        url = "%s:%s" % (self.url['host'], self.url['port'])

        # Don't allow blank authentication
        if user == "" or password == "":
            return False

        try:
            conn = Connection.establish(url, transport=self.url['transport'], username=user, password=password)
            conn.close()
        except ConnectionError as e:
            self.log.debug("AMQP service authentication reports: %s" % str(e))
            return False
        except Exception as e:
            self.log.critical("cannot proceed with authentication")
            self.log.exception(e)
            return False

        return True

 def _create_client(self):
   try:
     self.connection_ = Connection(url='10.100.151.13:5672', heartbeat=4, reconnect=True,
                                   reconnect_limit=10, reconnect_interval=4)
     self.connection_.open()
     self.sender_ = self.connection_.session().sender('leso.exchange.fresh_video')
   except:
     self.connection_ = None
     self.logger_.exception('failed to connect to message queue server.')

def migrate(*args, **kwargs):
    """
    Migrate qpid queues:
    - Ensure pulp.task is no longer *exclusive*.
    - Rename agent queues: consumer_id> => pulp.agent.<consumer_id>
    """
    transport = pulp_conf.get('messaging', 'transport')
    if transport != 'qpid':
        # not using qpid
        return

    if not QPID_MESSAGING_AVAILABLE:
        msg = _('Migration 0009 did not run because the python package qpid.messaging is not '
                'installed. Please install qpid.messaging and rerun the migrations. See %s'
                'for more information.')
        msg = msg % QPID_MESSAGING_URL
        _logger.error(msg)
        raise Exception(msg)

    if not QPIDTOOLLIBS_AVAILABLE:
        msg = _('Migration 0009 did not run because the python package qpidtoollibs is not '
                'installed. Please install qpidtoollibs and rerun the migrations. See %s for more '
                'information.')
        msg = msg % QPIDTOOLLIBS_URL
        _logger.error(msg)
        raise Exception(msg)

    url = urlparse(pulp_conf.get('messaging', 'url'))
    connection = Connection(
        host=url.hostname,
        port=url.port,
        transport=url.scheme,
        reconnect=False,
        ssl_certfile=pulp_conf.get('messaging', 'clientcert'),
        ssl_skip_hostname_check=True)

    connection.attach()
    broker = BrokerAgent(connection)
    _migrate_reply_queue(broker)
    _migrate_agent_queues(broker)
    connection.detach()

def migrate(*args, **kwargs):
    """
    Migrate qpid queues:
    - Ensure pulp.task is no longer *exclusive*.
    - Rename agent queues: consumer_id> => pulp.agent.<consumer_id>
    """
    transport = pulp_conf.get('messaging', 'transport')
    if transport != 'qpid':
        # not using qpid
        return

    if not QPID_MESSAGING_AVAILABLE:
        msg = _('Migration 0009 did not run because the python package qpid.messaging is not '
                'installed. Pulp\'s Qpid client dependencies can be installed with the '
                '\"pulp-server-qpid\" package group. See the installation docs for more '
                'information. Alternatively, you may reconfigure Pulp to use RabbitMQ.')
        _logger.error(msg)
        raise Exception(msg)

    if not QPIDTOOLLIBS_AVAILABLE:
        msg = _('Migration 0009 did not run because the python package qpidtoollibs is not '
                'installed. Pulp\'s Qpid client dependencies can be installed with the '
                '\"pulp-server-qpid\" package group. See the installation docs for more '
                'information. Alternatively, you may reconfigure Pulp to use RabbitMQ.')
        _logger.error(msg)
        raise Exception(msg)

    url = urlparse(pulp_conf.get('messaging', 'url'))
    connection = Connection(
        host=url.hostname,
        port=url.port,
        transport=url.scheme,
        reconnect=False,
        ssl_certfile=pulp_conf.get('messaging', 'clientcert'),
        ssl_skip_hostname_check=True)

    connection.attach()
    broker = BrokerAgent(connection)
    _migrate_reply_queue(broker)
    _migrate_agent_queues(broker)
    connection.detach()

 def __init__(self, url, receiver_name, sender_name='pulp.task', asserting=False, **options):
     '''establishes a connection to given url; initializes session, sender and receiver'''
     self.url = url
     self.receiver_name = receiver_name
     self.sender_name = sender_name
     self._asserting = asserting
     self.last_sent = None
     self.last_fetched = None
     self.session = Connection.establish(self.url, **options).session()
     self.receiver = self.session.receiver("%s; {create: always}" % self.receiver_name)
     self.sender = self.session.sender(self.sender_name)
     self._timeout = None

class Qpid(MQ):
    _timeout = 1

    def _conectar(self):
        try:
            logger.debug("Qpid: %s" % self._url.netloc)
            self._conn = Connection(self._url.netloc)
            if not self._conn:
                raise MQError(None, 2)
            self._conn.open()
        except ConnectError:
            raise MQError(cod=2)
        try:
            self._session = self._conn.session()
            self._sender = self._session.sender(self._url.path[1:])
            self._receiver = self._session.receiver(self._url.path[1:])
            logger.info("Connected on queue %s on %s" % (self._url.path[1:], self._url.netloc))
        except ConnectError:
            raise MQError(cod=3)

    def _enviar(self, mensagem):
        logger.debug("Sending a message")
        m = Message(mensagem)
        self._sender.send(m, True, self._timeout)

    def _receber(self, timeout=None):
        logger.debug("Retrieving a message")
        self._mensagem = self._receiver.fetch(timeout)
        return self._mensagem.content

    def _tamanho(self):
        self._receiver.available()

    def _excluir(self):
        logger.debug("Ack last message")
        self._session.acknowledge()

    def _terminar(self):
        self._conn.close(self._timeout)

class SetupTests(Base):

  def testOpen(self):
    # XXX: need to flesh out URL support/syntax
    self.conn = Connection.open(self.broker.host, self.broker.port,
                                reconnect=self.reconnect())
    self.ping(self.conn.session())

  def testConnect(self):
    # XXX: need to flesh out URL support/syntax
    self.conn = Connection(self.broker.host, self.broker.port,
                           reconnect=self.reconnect())
    self.conn.connect()
    self.ping(self.conn.session())

  def testConnectError(self):
    try:
      self.conn = Connection.open("localhost", 0)
      assert False, "connect succeeded"
    except ConnectError, e:
      # XXX: should verify that e includes appropriate diagnostic info
      pass

class QpidConnection():

    port = None
    ip = None
    data_queue = None
    cmd_queue = None
    conn = None
    session = None
    sender = None
    receiver = None

    def __init__(self, ip, port, data_queue, cmd_queue):
            self.ip = ip;
            self.port = port
            self.data_queue = data_queue
            self.cmd_queue = cmd_queue

    def start(self):
        try:
            url = str(self.ip)
            url += str(':')
            url += str(self.port)
            self.conn = Connection(url)
            self.conn.open()
            self.session = self.conn.session()
            self.sender = self.session.sender(self.cmd_queue)
            self.receiver = self.session.receiver(self.data_queue)
            return 1
        except MessagingError as m:
            print(m)
            return 0

    def stop(self):
        try:
            self.conn.close()
        except MessagingError as m:
            print(m)
            return 0

 def start(self):
     try:
         url = str(self.ip)
         url += str(':')
         url += str(self.port)
         self.conn = Connection(url)
         self.conn.open()
         self.session = self.conn.session()
         self.sender = self.session.sender(self.cmd_queue)
         self.receiver = self.session.receiver(self.data_queue)
         return 1
     except MessagingError as m:
         print(m)
         return 0

 def open(self):
     """
     Open a connection to the broker.
     """
     if self.is_open():
         # already open
         return
     connector = Connector.find(self.url)
     Connection.add_transports()
     domain = self.ssl_domain(connector)
     log.info('open: %s', connector)
     impl = RealConnection(
         host=connector.host,
         port=connector.port,
         tcp_nodelay=True,
         transport=connector.url.scheme,
         username=connector.userid,
         password=connector.password,
         heartbeat=10,
         **domain)
     impl.open()
     self._impl = impl
     log.info('opened: %s', self.url)

 def _conectar(self):
     try:
         logger.debug("Qpid: %s" % self._url.netloc)
         self._conn = Connection(self._url.netloc)
         if not self._conn:
             raise MQError(None, 2)
         self._conn.open()
     except ConnectError:
         raise MQError(cod=2)
     try:
         self._session = self._conn.session()
         self._sender = self._session.sender(self._url.path[1:])
         self._receiver = self._session.receiver(self._url.path[1:])
         logger.info("Connected on queue %s on %s" % (self._url.path[1:], self._url.netloc))
     except ConnectError:
         raise MQError(cod=3)

    def __init__(self, url, domain="org.clacks", xquery=".", callback=None):

        # Build connection
        url = parseURL(url)

        _url = "%s:%s" % (url['host'], url['port'])
        self.__conn = Connection.establish(_url, reconnect=True,
            username=url['user'],
            password=url['password'],
            transport=url['transport'],
            reconnect_interval=3,
            reconnect_limit=0)

        # Do automatic broker failover if requested
        #TODO: configure reconnect
        #auto_fetch_reconnect_urls(self.__conn)

        # Assemble subscription query
        queue = 'event-listener-%s' % uuid4()
        address = """%s; {
            create: always,
            delete:always,
            node: {
                durable: False,
                x-declare: {
                    exclusive: True,
                    auto-delete: True }
            },
            link: {
                x-bindings: [
                        {
                            exchange: '%s',
                            queue: %s,
                            key: event,
                            arguments: { xquery: %r}
                        }
                    ]
                }
            }""" % (queue, domain, queue, xquery)

        # Add processor for core.event queue
        self.__callback = callback
        self.__eventWorker = AMQPStandaloneWorker(
                        self.__conn,
                        r_address=address,
                        workers=1,
                        callback=self.__eventProcessor)

    def start(self):
        """
        Enable AMQP queueing. This method puts up the event processor and
        sets it to "active".
        """
        self.log.debug("enabling AMQP queueing")

        # Evaluate username
        user = self.config.get("amqp.id", default=None)
        if not user:
            user = self.env.uuid
        password = self.config.get("amqp.key")

        # Create initial broker connection
        url = "%s:%s" % (self.url['host'], self.url['port'])
        self._conn = Connection.establish(url, reconnect=self.reconnect,
            username=user,
            password=password,
            transport=self.url['transport'],
            reconnect_interval=self.reconnect_interval,
            reconnect_limit=self.reconnect_limit)

        # Do automatic broker failover if requested
        if self.config.get('amqp.failover', False):
            auto_fetch_reconnect_urls(self._conn)

        # Create event exchange
        socket = connect(self.url['host'], self.url['port'])
        if self.url['scheme'][-1] == 's':
            socket = ssl(socket)
        user = self.config.get("amqp.id", default=None)
        if not user:
            user = self.env.uuid
        connection = DirectConnection(sock=socket,
                username=user,
                password=self.config.get("amqp.key"))
        connection.start()
        session = connection.session(str(uuid4()))
        # pylint: disable=E1103
        session.exchange_declare(exchange=self.env.domain, type="xml")
        connection.close()

        # Create event provider
        self._eventProvider = EventProvider(self.env, self.getConnection())

    def start(self, is_reconnect=False):
        """
        Initiate the Qpid connection and start up any listeners.

        :param: is_reconnect True if this method is called as part of a
                             reconnect attempt, False otherwise
        :raise: ConnectionError if a connection cannot be established
        """
        # If the Qpid broker URL is not specified (or just the hostname is not
        # specified) then we can't make a connection.
        if not self.url or self.url.startswith(':'):
            log(self.log, 'warn', _('Qpid broker not specified, cannot start '
                                    'connection.'))
            return

        if not self._is_connected:
            self.conn = Connection(self.url, username=self.username,
                                   password=self.password,
                                   transport=self.transport)
            try:
                self.conn.open()
            except ConnectionError as e:
                log(self.log, 'critical', _('Cannot connect to Qpid message '
                                            'broker: %s') % (e.message))
                # close this connection when encounter connection error
                # otherwise, it will leave an ESTABLISHED connection
                # to qpid server forever.
                if self.conn is not None:
                    self.conn.close()
                raise e

            self._is_connected = True

            if is_reconnect and self.reconnect_handler:
                self.reconnect_handler()

            for listener in self._listeners:
                listener._start(self.conn)

            log(self.log, 'info', _('Connected to Qpid message broker: '
                                    '%[email protected]%s') % (self.username, self.url))

    def __init__(self, url, domain="org.gosa", xquery=".", callback=None):

        # Build connection
        url = parseURL(url)
        self.__conn = Connection(url['url'], transport=url['transport'], reconnect=True)
        self.__conn.open()

        # Assemble subscription query
        queue = 'event-listener-%s' % uuid4()
        address = """%s; {
            create: always,
            delete:always,
            node: {
                durable: False,
                x-declare: {
                    exclusive: True,
                    auto-delete: True }
            },
            link: {
                x-bindings: [
                        {
                            exchange: '%s',
                            queue: %s,
                            key: event,
                            arguments: { xquery: %r}
                        }
                    ]
                }
            }""" % (queue, domain, queue, xquery)

        # Add processor for core.event queue
        self.__callback = callback
        self.__eventWorker = AMQPStandaloneWorker(
                        self.__conn,
                        r_address=address,
                        workers=1,
                        callback=self.__eventProcessor)