<commit_before>package org.callimachusproject.server.util;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;

public class AsyncPipe {
	public abstract static class Sink implements ReadableByteChannel {
		private boolean closed;

		public boolean isOpen() {
			return !closed;
		}

		public void close() throws IOException {
			closed = true;
		}
	};

	abstract static class Source implements WritableByteChannel {
		private boolean closed;

		public boolean isOpen() {
			return !closed;
		}

		public void close() throws IOException {
			closed = true;
		}
	};

	private final ByteBuffer buf;
	private Throwable error;
	private boolean closed;
	private Runnable action;

	public AsyncPipe() {
		this(65536);
	}

	public AsyncPipe(int capacity) {
		this.buf = ByteBuffer.allocate(capacity);
	}

	public synchronized boolean isOpen() {
		return !closed;
	}

	/**
	 * Source operations will no longer see any data (after internal buffer is
	 * consumed or {@link #sink(ReadableByteChannel)} is called) and
	 * {@link #sink(ReadableByteChannel)} will sink to nil.
	 */
	public synchronized void close() {
		closed = true;
		this.notifyAll();
	}

	/**
	 * Source operations will fail with an IOException with the given throwable as
	 * the cause and the pipe will be closed.
	 */
	public synchronized void fail(Throwable e) {
		error = e;
		close();
	}

	public synchronized boolean hasAvailableCapacity() {
		if (closed || buf.hasRemaining())
			return true;
		try {
			this.wait(500);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
		return closed || buf.hasRemaining();
	}

	public synchronized void onAvailableCapacity(Runnable action) {
		if (closed || buf.hasRemaining()) {
			action.run();
		} else {
			this.action = action;
		}
	}

	/**
	 * Read n bytes from in into the internal buffer
	 */
	public synchronized int sink(ReadableByteChannel in) throws IOException {
		if (closed) {
			buf.clear();
		}
		try {
			return in.read(buf);
		} catch (Error e) {
			fail(e);
			throw e;
		} catch (IOException e) {<commit_msg>Suspend request sooner if not ready<commit_after>package org.callimachusproject.server.util;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;

public class AsyncPipe {
	public abstract static class Sink implements ReadableByteChannel {
		private boolean closed;

		public boolean isOpen() {
			return !closed;
		}

		public void close() throws IOException {
			closed = true;
		}
	};

	abstract static class Source implements WritableByteChannel {
		private boolean closed;

		public boolean isOpen() {
			return !closed;
		}

		public void close() throws IOException {
			closed = true;
		}
	};

	private final ByteBuffer buf;
	private Throwable error;
	private boolean closed;
	private Runnable action;

	public AsyncPipe() {
		this(65536);
	}

	public AsyncPipe(int capacity) {
		this.buf = ByteBuffer.allocate(capacity);
	}

	public synchronized boolean isOpen() {
		return !closed;
	}

	/**
	 * Source operations will no longer see any data (after internal buffer is
	 * consumed or {@link #sink(ReadableByteChannel)} is called) and
	 * {@link #sink(ReadableByteChannel)} will sink to nil.
	 */
	public synchronized void close() {
		closed = true;
		this.notifyAll();
	}

	/**
	 * Source operations will fail with an IOException with the given throwable as
	 * the cause and the pipe will be closed.
	 */
	public synchronized void fail(Throwable e) {
		error = e;
		close();
	}

	public synchronized boolean hasAvailableCapacity() {
		if (closed || buf.hasRemaining())
			return true;
		try {
			this.wait(100);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
		return closed || buf.hasRemaining();
	}

	public synchronized void onAvailableCapacity(Runnable action) {
		if (closed || buf.hasRemaining()) {
			action.run();
		} else {
			this.action = action;
		}
	}

	/**
	 * Read n bytes from in into the internal buffer
	 */
	public synchronized int sink(ReadableByteChannel in) throws IOException {
		if (closed) {
			buf.clear();
		}
		try {
			return in.read(buf);
		} catch (Error e) {
			fail(e);
			throw e;
		} catch (IOException e) {
{'fraction_non_alphanumeric': 0.10190217391304347, 'fraction_numerical': 0.0036231884057971015, 'mean_word_length': 3.729122055674518, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3310278', 'n_tokens_mistral': 1467, 'n_tokens_neox': 1295, 'n_words': 417}