events.add(Event(id, type, data))
    drainCancelQueue(eventSource)
  }

  override fun onClosed(eventSource: EventSource) {
    get().log("[ES] onClosed", Platform.INFO, null)
    events.add(Closed)
    drainCancelQueue(eventSource)
  }

  override fun onFailure(
    eventSource: EventSource,
    t: Throwable?,
    response: Response?,
  ) {
    get().log("[ES] onFailure", Platform.INFO, t)

        this.eventQueue = new ArrayBlockingQueue<>(queueSize);
        this.closed = new AtomicBoolean(false);
        this.updater = new Thread(this::feedConsumer, "simplex-transfer-listener");
        updater.setDaemon(true);
        updater.start();
    }

    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            updater.interrupt();
        }
    }

        assertTrue(ex1.getMessage().contains("Already closed"));
        CIFSException ex2 = assertThrows(CIFSException.class, handle::getOutput, "getOutput after close must throw");
        assertTrue(ex2.getMessage().contains("Already closed"));
    }

    @Test
    @DisplayName("ensureOpen opens SMB2 via UNC path and marks open")

                out <- []byte("response")

                // Return the request for reuse
                grid.PutByteBuffer(req)
            }
        }
        // out is closed by the caller and should never be closed by the handler.
        return nil
    }

    err := manager.RegisterStreamingHandler(grid.HandlerDiskInfo, StreamHandler{
        Handle: handler,
        Subroute: "asubroute",

        ESTABLISHED,
        /** Connection encountered an error */
        ERROR,
        /** Connection is being closed */
        CLOSING,
        /** Connection is closed */
        CLOSED
    }

    /** Remote endpoint address */
    protected final InetSocketAddress remoteAddress;
    /** Local endpoint address */
    protected final InetSocketAddress localAddress;

  }

  public void testCopyToAppendable_doesNotCloseIfWriter() throws IOException {
    TestWriter writer = new TestWriter();
    assertFalse(writer.closed());
    source.copyTo(writer);
    assertFalse(writer.closed());
  }

  public void testClosesOnErrors_copyingToCharSinkThatThrows() {
    for (TestOption option : EnumSet.of(OPEN_THROWS, WRITE_THROWS, CLOSE_THROWS)) {

  }

  @Override
  public String toString() {
    return comparator
        + ":"
        + (lowerBoundType == CLOSED ? '[' : '(')
        + (hasLowerBound ? lowerEndpoint : "-\u221e")
        + ','
        + (hasUpperBound ? upperEndpoint : "\u221e")
        + (upperBoundType == CLOSED ? ']' : ')');
  }

  @Nullable T getLowerEndpoint() {
    return lowerEndpoint;
  }

	crc32     hash.Hash32
	closed    bool
}

func (w *fileWriter) Write(p []byte) (int, error) {
	if w.closed {
		return 0, errors.New("zip: write to closed file")
	}
	if w.raw {
		return w.zipw.Write(p)
	}
	w.crc32.Write(p)
	return w.rawCount.Write(p)
}

func (w *fileWriter) close() error {
	if w.closed {
		return errors.New("zip: file closed twice")
	}
	w.closed = true

    int ERROR_REQ_NOT_ACCEP = 71;
    /** The pipe state is invalid */
    int ERROR_BAD_PIPE = 230;
    /** All pipe instances are busy */
    int ERROR_PIPE_BUSY = 231;
    /** The pipe is being closed */
    int ERROR_NO_DATA = 232;
    /** No process is on the other end of the pipe */
    int ERROR_PIPE_NOT_CONNECTED = 233;
    /** More data is available */
    int ERROR_MORE_DATA = 234;

    int ERROR_REQ_NOT_ACCEP = 71;
    /** The pipe state is invalid */
    int ERROR_BAD_PIPE = 230;
    /** All pipe instances are busy */
    int ERROR_PIPE_BUSY = 231;
    /** The pipe is being closed */
    int ERROR_NO_DATA = 232;
    /** No process is on the other end of the pipe */
    int ERROR_PIPE_NOT_CONNECTED = 233;
    /** More data is available */
    int ERROR_MORE_DATA = 234;