<commit_before>import time
import socket

from tornado.iostream import IOStream
from tornado.ioloop import IOLoop

from stormed.util import logger, AmqpStatusError
from stormed.frame import FrameReader, FrameHandler, status
from stormed.channel import Channel
from stormed.method.connection import Close

TORNADO_1_2 = hasattr(IOStream, 'connect')

class Connection(FrameHandler):
    """A "physical" TCP connection to the AMQP server

    heartbeat: int, optional
               the requested time interval in seconds for heartbeat frames.

    Connection.on_error callback, when set, is called in case of
    "hard" AMQP Error. It receives a ConnectionErrorinstance as argument:

        def handle_error(conn_error):
            print conn_error.method
            print conn_error.reply_code

        conn.on_error = handle_error

    Connection.on_disconnect callback, when set, is called in case of
    heartbeat timeout or TCP low level disconnection. It receives no args.
    """

    def __init__(self, host, username='guest', password='<PASSWORD>', vhost='/',
                       port=5672, heartbeat=0, io_loop=None):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.vhost = vhost
        self.heartbeat = heartbeat
        self.last_received_frame = None
        self.frame_max = 0
        self.io_loop = io_loop or IOLoop.instance()
        self.stream = None
        self.channels = {0: self}
        self.last_channel_id = 0
        self.channel_id = 0
        self.on_connect = None
        self.on_disconnect = None
        self.on_error = None
        self._close_callback = None
        self._frame_count = 0
        super(Connection, self).__init__(connection=self)

    def connect(self, callback):
        """open the connection to the server"""
        if self.status is not status.CLOSED:
            raise AmqpStatusError('Connection status is %s' % self.status)
        self.status = status.OPENING
        sock = socket.socket()
        sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
        self.on_connect = callback
        if TORNADO_1_2:
            self.stream = IOStream(sock, io_loop=self.io_loop)
            self.stream.set_close_callback(self.on_closed_stream)
            self.stream.connect((self.host, self.port), self._handshake)<commit_msg>add new a parameter `close_callback' to connect.<commit_after>import time
import socket

from tornado.iostream import IOStream
from tornado.ioloop import IOLoop

from stormed.util import logger, AmqpStatusError
from stormed.frame import FrameReader, FrameHandler, status
from stormed.channel import Channel
from stormed.method.connection import Close

TORNADO_1_2 = hasattr(IOStream, 'connect')

class Connection(FrameHandler):
    """A "physical" TCP connection to the AMQP server

    heartbeat: int, optional
               the requested time interval in seconds for heartbeat frames.

    Connection.on_error callback, when set, is called in case of
    "hard" AMQP Error. It receives a ConnectionErrorinstance as argument:

        def handle_error(conn_error):
            print conn_error.method
            print conn_error.reply_code

        conn.on_error = handle_error

    Connection.on_disconnect callback, when set, is called in case of
    heartbeat timeout or TCP low level disconnection. It receives no args.
    """

    def __init__(self, host, username='guest', password='<PASSWORD>', vhost='/',
                       port=5672, heartbeat=0, io_loop=None):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.vhost = vhost
        self.heartbeat = heartbeat
        self.last_received_frame = None
        self.frame_max = 0
        self.io_loop = io_loop or IOLoop.instance()
        self.stream = None
        self.channels = {0: self}
        self.last_channel_id = 0
        self.channel_id = 0
        self.on_connect = None
        self.on_disconnect = None
        self.on_error = None
        self._close_callback = None
        self._frame_count = 0
        super(Connection, self).__init__(connection=self)

    def connect(self, callback, close_callback=None):
        """open the connection to the server"""
        if self.status is not status.CLOSED:
            raise AmqpStatusError('Connection status is %s' % self.status)
        self.status = status.OPENING
        sock = socket.socket()
        sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
        self.on_connect = callback
        self.on_disconnect = close_callback
        if TORNADO_1_2:
            self.stream = IOStream(sock, io_loop=self.io_loop)
            self.stream.set_close_callback(self.on_closed_stream)
            self.stream.connect((self.host, self.port), self._handshake)
{'fraction_non_alphanumeric': 0.07817656765676567, 'fraction_numerical': 0.006188118811881188, 'mean_word_length': 2.6541070082893747, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '1033536', 'n_tokens_mistral': 1393, 'n_tokens_neox': 1337, 'n_words': 388}