return EventSource.Factory { request, listener ->
      val actualRequest =
        if (request.header("Accept") == null) {
          request.newBuilder().addHeader("Accept", "text/event-stream").build()
        } else {
          request
        }

      RealEventSource(actualRequest, listener).apply {
        connect(callFactory)
      }
    }
  }

  @JvmStatic

  private var protocol: Protocol?,
  private var source: BufferedSource?,
  private var sink: BufferedSink?,
  private val pingIntervalMillis: Int,
  internal val connectionListener: ConnectionListener,
) : Http2Connection.Listener(), Connection, ExchangeCodec.Carrier {
  private var http2Connection: Http2Connection? = null

  internal val lock: ReentrantLock = ReentrantLock()

  // These properties are guarded by [lock].

  /**

   * [requestBodyOpen], [responseBodyOpen], and [expectMoreExchanges].
   *
   * This will release the connection if it is still held.
   *
   * It will also notify the listener that the call completed; either successfully or
   * unsuccessfully.
   *
   * If the call was canceled or timed out, this will wrap [e] in an exception that provides that

        val http2SocketHandler = Http2SocketHandler(socket, protocol)
        val connection =
          Http2Connection.Builder(false, taskRunner)
            .socket(socket)
            .listener(http2SocketHandler)
            .build()
        connection.start()
        openConnections.add(connection)
        openClientSockets.remove(socket)
        return
      } else if (protocol !== Protocol.HTTP_1_1) {

    peer: MockHttp2Peer,
    pushObserver: PushObserver = IGNORE,
    listener: Http2Connection.Listener = Http2Connection.Listener.REFUSE_INCOMING_STREAMS,
  ): Http2Connection {
    val connection =
      Http2Connection.Builder(true, TaskRunner.INSTANCE)
        .socket(peer.openSocket())
        .pushObserver(pushObserver)
        .listener(listener)
        .build()
    connection.start(sendConnectionPreface = false)

    listener.removeUpToEvent(ConnectionAcquired::class.java)
    listener.removeUpToEvent(ResponseFailed::class.java)
    listener.removeUpToEvent(ConnectionReleased::class.java)
    listener.removeUpToEvent(ConnectionAcquired::class.java)
    listener.removeUpToEvent(ConnectionReleased::class.java)
    listener.removeUpToEvent(CallEnd::class.java)
  }

  @Test
  fun recoverWhenRetryOnConnectionFailureIsTrue_HTTP2() {

    server.processNextFrame()
    client.listener.assertClosing(1000, "Hello!")
    server.listener.assertClosing(1000, "Hi!")
    client.webSocket!!.finishReader()
    server.webSocket!!.finishReader()
    client.listener.assertClosed(1000, "Hello!")
    server.listener.assertClosed(1000, "Hi!")
    taskFaker.runTasks()
    assertThat(client.closed).isTrue()

        .build().also {
          listener.satisfactionFailure(call, it)
        }
    }

    // If we don't need the network, we're done.
    if (networkRequest == null) {
      return cacheResponse!!.newBuilder()
        .cacheResponse(cacheResponse.stripBody())
        .build().also {
          listener.cacheHit(call, it)
        }
    }

    if (cacheResponse != null) {

  val dispatcher = Dispatcher(executor)
  val listener = RecordingEventListener()
  var client =
    clientTestRule.newClientBuilder()
      .dns { throw UnknownHostException() }
      .dispatcher(dispatcher)
      .eventListenerFactory(clientTestRule.wrap(listener))
      .build()

  @BeforeEach
  fun setUp() {
    dispatcher.maxRequests = 20
    dispatcher.maxRequestsPerHost = 10
    listener.forbidLock(dispatcher)
  }

  @Test

    server.close(1000, null)
    serverListener.assertClosed(1000, "goodbye")
  }

  @Test
  fun webSocketsDontTriggerEventListener() {
    val listener = RecordingEventListener()
    client =
      client.newBuilder()
        .eventListenerFactory(clientTestRule.wrap(listener))
        .build()
    webServer.enqueue(
      MockResponse.Builder()
        .webSocketUpgrade(serverListener)
        .build(),
    )