def _handle_stream(self, stream, address):
        """Handle a new connection as a tornado.iostream.IOStream instance."""
        try:
            assert get_thread_ident() == self.ioloop_thread_id
            stream.set_close_callback(partial(self._stream_closed_callback,
                                              stream))
            # Our message packets are small, don't delay sending them.
            stream.set_nodelay(True)
            stream.max_write_buffer_size = self.MAX_WRITE_BUFFER_SIZE

            # Abuse IOStream object slightly by adding 'address' and 'closing'
            # attributes. Use nasty prefix to prevent naming collisions.
            stream.KATCPServer_address = address
            # Flag to indicate that no more write should be accepted so that
            # we can flush the write buffer when closing a connection
            stream.KATCPServer_closing = False

            client_conn = self.client_connection_factory(self, stream)
            self._connections[stream] = client_conn
            try:
                yield gen.maybe_future(self._device.on_client_connect(client_conn))
            except Exception:
                # If on_client_connect fails there is no reason to continue
                # trying to handle this connection. Try and send exception info
                # to the client and disconnect
                e_type, e_value, trace = sys.exc_info()
                reason = "\n".join(traceback.format_exception(
                    e_type, e_value, trace, self._tb_limit))
                log_msg = 'Device error initialising connection {0}'.format(reason)
                self._logger.error(log_msg)
                stream.write(str(Message.inform('log', log_msg)))
                stream.close(exc_info=True)
            else:
                self._line_read_loop(stream, client_conn)
        except Exception:
            self._logger.error('Unhandled exception trying '
                               'to handle new connection', exc_info=True)
Handle a new connection as a tornado.iostream.IOStream instance.