) {
      check(!closed) { "closed" }
      checkOffsetAndCount(source.size, 0, byteCount)
      socket.sink.write(source, byteCount)
    }

    override fun flush() {
      if (closed) return // Don't throw; this stream might have been closed on the caller's behalf.
      socket.sink.flush()
    }

    override fun close() {
      if (closed) return
      closed = true
      detachTimeout(timeout)

 *
 * {@snippet :
 * FluentFuture<UserName> userName =
 *     ClosingFuture.submit(
 *             closer -> closer.eventuallyClose(database.newTransaction(), closingExecutor),
 *             executor)
 *         .transformAsync((closer, transaction) -> transaction.queryClosingFuture("..."), executor)
 *         .transform((closer, result) -> result.get("userName"), directExecutor())

      private var closed = false

      override fun close() {
        if (closed) return
        try {
          super.close()
        } finally {
          closedLatch.countDown()
        }
      }
    }.buffer()

  override val sink: BufferedSink =
    object : ForwardingSink(delegate.sink) {
      private var closed = false

      override fun close() {

    }

    @Test
    @DisplayName("Test initial state is CLOSED")
    void testInitialState() {
        assertEquals(SimpleCircuitBreaker.State.CLOSED, circuitBreaker.getState());
        assertTrue(circuitBreaker.allowsRequests());
    }

    @Test
    @DisplayName("Test successful calls in CLOSED state")
    void testSuccessfulCalls() throws Exception {

            if (!holder.closed && !holder.isAlive()) {
                // Resource object has been garbage collected but not closed
                reportLeak(holder);
                activeResources.remove(entry.getKey());
                cleaned++;
            } else if (!holder.closed && holder.getAge() > maxResourceAge * 2) {
                // Very old unclosed resource - attempt to close it

        assertNotNull(auth.getPasswordAsCharArray());
        assertFalse(auth.isClosed());

        // Close the authenticator
        auth.close();

        // Verify it's closed
        assertTrue(auth.isClosed());
        // Cannot check password after closing as it now throws IllegalStateException
        // This is actually good security practice - closed authenticators cannot be accessed

  }

  @Throws(IOException::class)
  fun flush() {
    withLock {
      if (closed) throw IOException("closed")
      sink.flush()
    }
  }

  @Throws(IOException::class)
  fun rstStream(
    streamId: Int,
    errorCode: ErrorCode,
  ) {
    withLock {
      if (closed) throw IOException("closed")
      require(errorCode.httpCode != -1)

      frameHeader(
        streamId = streamId,

 *
 * {@snippet :
 * FluentFuture<UserName> userName =
 *     ClosingFuture.submit(
 *             closer -> closer.eventuallyClose(database.newTransaction(), closingExecutor),
 *             executor)
 *         .transformAsync((closer, transaction) -> transaction.queryClosingFuture("..."), executor)
 *         .transform((closer, result) -> result.get("userName"), directExecutor())

    client.listener.assertFailure(IOException::class.java, "canceled")
    taskFaker.runTasks()
  }

  @Test
  fun serverCloseClosesConnection() {
    server.webSocket!!.close(1000, "Hello!")
    client.processNextFrame() // Read server close, send client close, close connection.
    assertThat(client.closed).isFalse()
    client.listener.assertClosing(1000, "Hello!")

        throw complete(e)!!
      }
    }

    @Throws(IOException::class)
    override fun close() {
      if (closed) return
      closed = true
      if (contentLength != -1L && bytesReceived != contentLength) {
        throw ProtocolException("unexpected end of stream")
      }
      try {
        super.close()
        complete(null)
      } catch (e: IOException) {
        throw complete(e)!!
      }