asynchronous_connection.py

asynchronous_connection.py

@gen.coroutine
def connect(self):
        """
        establishing two connections for publishing and receiving respectively.
        :return: True if establish successfully.
        """
        self._publish_connection = yield self._create_connection(self._parameter)
        self._receive_connection = yield self._create_connection(self._parameter)
        raise gen.Return(True)

def _create_connection(self, parameter):
        self.logger.info("creating connection")
        future = Future()

        def open_callback(unused_connection):
            self.logger.info("created connection")
            future.set_result(unused_connection)

        def open_error_callback(connection, exception):
            self.logger.error("open connection with error: %s" % exception)
            future.set_exception(exception)

        def close_callback(connection, reply_code, reply_text):
            if reply_code not in [self._NORMAL_CLOSE_CODE,]:
                self.logger.error("closing connection: reply code:%s, reply_text: %s. system will exist" % (reply_code, reply_text,))
                sys.exit(self._EXIST_CODE)

        TornadoConnection(parameter,
                          on_open_callback=open_callback,
                          on_open_error_callback=open_error_callback,
                          on_close_callback=close_callback,
                          custom_ioloop=self._io_loop)
        return future