}
  }

  /** Returns once all stream actions complete successfully. */
  fun awaitSuccess() {
    val futureTask =
      results.poll(5, TimeUnit.SECONDS)
        ?: throw AssertionError("no onRequest call received")
    futureTask.get(5, TimeUnit.SECONDS)
  }

        writeSynResetLater(streamId, ErrorCode.PROTOCOL_ERROR)
        return
      }
      currentPushRequests.add(streamId)
    }
    pushQueue.execute("$connectionName[$streamId] onRequest") {
      val cancel = pushObserver.onRequest(streamId, requestHeaders)
      ignoreIoExceptions {
        if (cancel) {
          writer.rstStream(streamId, ErrorCode.CANCEL)
          withLock {

    @Synchronized fun takeEvent(): Any {
      while (events.isEmpty()) {
        wait()
      }
      return events.removeAt(0)
    }

    @Synchronized override fun onRequest(
      streamId: Int,
      requestHeaders: List<Header>,
    ): Boolean {
      assertThat(streamId).isEqualTo(2)
      events.add(requestHeaders)
      notifyAll()
      return false
    }

    val frResponse = client.newCall(frRequest).execute()
    assertThat(frResponse.body.string()).isEqualTo("A")
    val enRequest =
      Request
        .Builder()
        .url(url)
        .header("Accept-Language", "en-US")
        .build()
    val enResponse = client.newCall(enRequest).execute()
    assertThat(enResponse.body.string()).isEqualTo("B")
  }

  @Test