return socket.isHealthy(source)
    }

    return true
  }

  /** Refuse incoming streams. */
  @Throws(IOException::class)
  override fun onStream(stream: Http2Stream) {
    stream.close(ErrorCode.REFUSED_STREAM, null)
  }

  /** When settings are received, adjust the allocation limit. */
  override fun onSettings(
    connection: Http2Connection,

 * layer primitives: connections, requests, responses, and streams.
 *
 * This class supports [asynchronous canceling][cancel]. This is intended to have the smallest
 * blast radius possible. If an HTTP/2 stream is active, canceling will cancel that stream but not
 * the other streams sharing its connection. But if the TLS handshake is still in progress then
 * canceling may break the entire connection.

          writeBytesTotal >= writeBytesMaximum ||
          stream.writeBytesTotal >= stream.writeBytesMaximum
        if (stream.isOpen) {
          streams[streamId] = stream
        }
      }
      if (associatedStreamId == 0) {
        writer.headers(outFinished, streamId, requestHeaders)
      } else {
        require(!client) { "client streams shouldn't have associated stream IDs" }
        // HTTP/2 has a PUSH_PROMISE frame.

    }
  }

  fun initReaderAndWriter(
    name: String,
    streams: Streams,
  ) {
    val extensions = this.extensions!!
    synchronized(this) {
      this.name = name
      this.streams = streams
      this.writer =
        WebSocketWriter(
          isClient = streams.client,
          sink = streams.sink,
          random = random,
          perMessageDeflate = extensions.perMessageDeflate,

      sink.writeInt(payload2)
      sink.flush()
    }
  }

  /**
   * Tell the peer to stop creating streams and that we last processed `lastGoodStreamId`, or zero
   * if no streams were processed.
   *
   * @param lastGoodStreamId the last stream ID processed, or zero if no streams were processed.
   * @param errorCode reason for closing the connection.

    connection: Http2Connection,
    amount: Int,
  ) {
    val settings = Settings()
    settings[Settings.MAX_CONCURRENT_STREAMS] = amount
    connection.readerRunnable.applyAndAckSettings(true, settings)
    assertThat(connection.peerSettings[Settings.MAX_CONCURRENT_STREAMS]).isEqualTo(amount)
    taskFaker.runTasks()
  }

  /** Use a helper method so there's no hidden reference remaining on the stack.  */

      check(!isLocallyInitiated) { "locally-initiated streams shouldn't have headers yet" }
      headersQueue += headers
    } else {
      check(isLocallyInitiated) { "remotely-initiated streams should have headers" }
    }
  }

  /**
   * Returns true if this stream is open. A stream is open until either:
   *
   *  * A `SYN_RESET` frame abnormally terminates the stream.

    client.listener.assertTextMessage(message)
  }

  /** One peer's streams, listener, and web socket in the test.  */
  private class TestStreams(
    client: Boolean,
    private val taskFaker: TaskFaker,
    private val sourcePipe: Pipe,
    private val sinkPipe: Pipe,
  ) : RealWebSocket.Streams(client, sourcePipe.source.buffer(), sinkPipe.sink.buffer()) {

      sleepNanos(response.headersDelayNanos)
      stream.writeHeaders(response.toHttp2Headers(), outFinished, flushHeaders)

      if (trailers.size > 0) {
        stream.enqueueTrailers(trailers)
      }
      pushPromises(stream, request, response.pushPromises)
      if (body != null) {
        sleepNanos(bodyDelayNanos)
        val responseBodySink =
          stream.getSink().withThrottlingAndSocketPolicy(

    server.enqueue(
      MockResponse(socketPolicy = ResetStreamAtStart(ErrorCode.REFUSED_STREAM.httpCode)),
    )
    server.enqueue(
      MockResponse(socketPolicy = ResetStreamAtStart(ErrorCode.REFUSED_STREAM.httpCode)),
    )
    server.enqueue(
      MockResponse(socketPolicy = ResetStreamAtStart(ErrorCode.REFUSED_STREAM.httpCode)),
    )

    val request = Request(server.url("/"))