def setup_data_stream(
            self,
            connection_factory: Callable[[tuple], Connection],
            data_stream_factory: Callable[[Connection], DataStream]=DataStream) -> \
            DataStream:
        '''Create and setup a data stream.

        This function will set up passive and binary mode and handle
        connecting to the data connection.

        Args:
            connection_factory: A coroutine callback that returns a connection
            data_stream_factory: A callback that returns a data stream

        Coroutine.

        Returns:
            DataStream
        '''
        yield from self._control_stream.write_command(Command('TYPE', 'I'))
        reply = yield from self._control_stream.read_reply()

        self.raise_if_not_match('Binary mode', ReplyCodes.command_okay, reply)

        address = yield from self.passive_mode()

        connection = yield from connection_factory(address)

        # TODO: unit test for following line for connections that have
        # the same port over time but within pool cleaning intervals
        connection.reset()

        yield from connection.connect()

        data_stream = data_stream_factory(connection)

        return data_stream
Create and setup a data stream.

        This function will set up passive and binary mode and handle
        connecting to the data connection.

        Args:
            connection_factory: A coroutine callback that returns a connection
            data_stream_factory: A callback that returns a data stream

        Coroutine.

        Returns:
            DataStream