return ChunkedSource(url)
  }

  private fun newUnknownLengthSource(url: HttpUrl): Source {
    check(state == STATE_OPEN_RESPONSE_BODY) { "state: $state" }
    state = STATE_READING_RESPONSE_BODY
    carrier.noNewExchanges()
    return UnknownLengthSource(url)
  }

  /**
   * Sets the delegate of `timeout` to [Timeout.NONE] and resets its underlying timeout

          noNewExchangesEvent = !connection.noNewExchanges
          connection.noNewExchanges = true
          call.releaseConnectionNoEvents()
        }
      if (toClose != null) {
        toClose.closeQuietly()
        connectionListener.connectionClosed(connection)
      } else if (noNewExchangesEvent) {
        connectionListener.noNewExchanges(connection)
      }
    }
    return null
  }

    assertThat(server2.requestCount).isEqualTo(1)

    assertThat(clientTestRule.recordedConnectionEventTypes()).containsExactly(
      "ConnectStart",
      "ConnectEnd",
      "ConnectionAcquired",
      "NoNewExchanges",
      "ConnectionReleased",
      "ConnectionClosed",
      "ConnectStart",
      "ConnectEnd",
      "ConnectionAcquired",
      "ConnectionReleased",
    )
  }

  @Test

            // Everything else wants a fresh connection.
            noNewExchangesEvent = !noNewExchanges
            noNewExchanges = true
            routeFailureCount++
          }
        }
      } else if (!isMultiplexed || e is ConnectionShutdownException) {
        noNewExchangesEvent = !noNewExchanges
        noNewExchanges = true

        // If this route hasn't completed a call, avoid it for new connections.

      if (connection.noNewExchanges) {
        assertThat(eventSequence).matchesPredicate { deque ->
          deque.any { it is NoNewExchanges && it.connection == connection }
        }
      }
    }

    logEvent(ConnectionEvent.ConnectionReleased(System.nanoTime(), connection, call))
  }

   */
  interface Carrier {
    val route: Route

    fun trackFailure(
      call: RealCall,
      e: IOException?,
    )

    fun noNewExchanges()

    fun cancel()
  }

  companion object {
    /**
     * The timeout to use while discarding a stream of input data. Since this is used for connection

    val toClose: Socket? =
      candidate.withLock {
        when {
          !healthy -> {
            noNewExchangesEvent = !candidate.noNewExchanges
            candidate.noNewExchanges = true
            call.releaseConnectionNoEvents()
          }
          candidate.noNewExchanges || !sameHostAndPort(candidate.route().address.url) -> {
            call.releaseConnectionNoEvents()
          }
          else -> null

   */
  open fun connectionReleased(
    connection: Connection,
    call: Call,
  ) {}

  /**
   * Invoked when a connection is marked for no new exchanges.
   */
  open fun noNewExchanges(connection: Connection) {}

  companion object {
    val NONE: ConnectionListener = object : ConnectionListener() {}
  }

    val call: Call,
  ) : ConnectionEvent() {
    override fun closes(event: ConnectionEvent): Boolean =
      event is ConnectionAcquired && connection == event.connection && call == event.call
  }

  data class NoNewExchanges(
    override val timestampNs: Long,
    override val connection: Connection,
  ) : ConnectionEvent()

    return connection
  }

  override fun trackFailure(
    call: RealCall,
    e: IOException?,
  ) {
    // Do nothing.
  }

  override fun noNewExchanges() {
    // Do nothing.
  }

  override fun cancel() {
    canceled = true
    // Close the raw socket so we don't end up doing synchronous I/O.
    rawSocket?.closeQuietly()
  }