'Returns the ResponseFuture before calling send_request() on it'

def _create_response_future(self, query, parameters, trace, custom_payload, timeout, execution_profile=EXEC_PROFILE_DEFAULT, paging_state=None):

    prepared_statement = None
    if isinstance(query, six.string_types):
        query = SimpleStatement(query)
    elif isinstance(query, PreparedStatement):
        query = query.bind(parameters)
    if (self.cluster._config_mode == _ConfigMode.LEGACY):
        if (execution_profile is not EXEC_PROFILE_DEFAULT):
            raise ValueError('Cannot    specify    execution_profile    while    using    legacy    parameters.')
        if (timeout is _NOT_SET):
            timeout = self.default_timeout
        cl = (query.consistency_level if (query.consistency_level is not None) else self.default_consistency_level)
        serial_cl = (query.serial_consistency_level if (query.serial_consistency_level is not None) else self.default_serial_consistency_level)
        retry_policy = (query.retry_policy or self.cluster.default_retry_policy)
        row_factory = self.row_factory
        load_balancing_policy = self.cluster.load_balancing_policy
        spec_exec_policy = None
    else:
        execution_profile = self._get_execution_profile(execution_profile)
        if (timeout is _NOT_SET):
            timeout = execution_profile.request_timeout
        cl = (query.consistency_level if (query.consistency_level is not None) else execution_profile.consistency_level)
        serial_cl = (query.serial_consistency_level if (query.serial_consistency_level is not None) else execution_profile.serial_consistency_level)
        retry_policy = (query.retry_policy or execution_profile.retry_policy)
        row_factory = execution_profile.row_factory
        load_balancing_policy = execution_profile.load_balancing_policy
        spec_exec_policy = execution_profile.speculative_execution_policy
    fetch_size = query.fetch_size
    if ((fetch_size is FETCH_SIZE_UNSET) and (self._protocol_version >= 2)):
        fetch_size = self.default_fetch_size
    elif (self._protocol_version == 1):
        fetch_size = None
    start_time = time.time()
    if ((self._protocol_version >= 3) and self.use_client_timestamp):
        timestamp = self.cluster.timestamp_generator()
    else:
        timestamp = None
    if isinstance(query, SimpleStatement):
        query_string = query.query_string
        if parameters:
            query_string = bind_params(query_string, parameters, self.encoder)
        message = QueryMessage(query_string, cl, serial_cl, fetch_size, timestamp=timestamp)
    elif isinstance(query, BoundStatement):
        prepared_statement = query.prepared_statement
        message = ExecuteMessage(prepared_statement.query_id, query.values, cl, serial_cl, fetch_size, timestamp=timestamp, skip_meta=bool(prepared_statement.result_metadata))
    elif isinstance(query, BatchStatement):
        if (self._protocol_version < 2):
            raise UnsupportedOperation('BatchStatement    execution    is    only    supported    with    protocol    version    2    or    higher    (supported    in    Cassandra    2.0    and    higher).        Consider    setting    Cluster.protocol_version    to    2    to    support    this    operation.')
        message = BatchMessage(query.batch_type, query._statements_and_parameters, cl, serial_cl, timestamp)
    message.tracing = trace
    message.update_custom_payload(query.custom_payload)
    message.update_custom_payload(custom_payload)
    message.allow_beta_protocol_version = self.cluster.allow_beta_protocol_version
    message.paging_state = paging_state
    spec_exec_plan = (spec_exec_policy.new_plan((query.keyspace or self.keyspace), query) if (query.is_idempotent and spec_exec_policy) else None)
    return ResponseFuture(self, message, query, timeout, metrics=self._metrics, prepared_statement=prepared_statement, retry_policy=retry_policy, row_factory=row_factory, load_balancer=load_balancing_policy, start_time=start_time, speculative_execution_plan=spec_exec_plan)
