Skip to content

Commit

Permalink
Make readBuffered blocking and add more readBuffered methods, fixes #757
Browse files Browse the repository at this point in the history 
 (#782)
  • Loading branch information
@gnodet
gnodet committed Jan 16, 2023
1 parent 4f57697 commit 77f1cea
Show file tree
Hide file tree
Showing 11 changed files with 230 additions and 99 deletions.
10 changes: 10 additions & 0 deletions reader/src/test/java/org/jline/terminal/impl/AbstractWindowsTerminalTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,16 @@ public void testBracketingPasteHuge() throws Exception {
for (int i = 0; i < 100000; i++) {
str.append("0123456789");
}
str.toString().chars().forEachOrdered(c -> process(terminal, c) );
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
str.setLength(0);
for (int i = 0; i < 100000; i++) {
str.append("0123456789");
}
str.append(LineReaderImpl.BRACKETED_PASTE_END);
str.append("\n");
str.toString().chars().forEachOrdered(c -> process(terminal, c));
Expand Down
5 changes: 0 additions & 5 deletions terminal/src/main/java/org/jline/terminal/impl/AbstractPty.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,11 +86,6 @@ public int read(long timeout, boolean isPeek) throws IOException {
}
}

@Override
public int readBuffered(byte[] b) throws IOException {
return in.read(b);
}

private void setNonBlocking() {
if (current == null
|| current.getControlChar(Attributes.ControlChar.VMIN) != 0
Expand Down
73 changes: 33 additions & 40 deletions terminal/src/main/java/org/jline/utils/NonBlocking.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,13 +95,9 @@ public void close() throws IOException {

@Override
public int read(long timeout, boolean isPeek) throws IOException {
boolean isInfinite = (timeout <= 0L);
while (!bytes.hasRemaining() && (isInfinite || timeout > 0L)) {
long start = 0;
if (!isInfinite) {
start = System.currentTimeMillis();
}
int c = reader.read(timeout);
Timeout t = new Timeout(timeout);
while (!bytes.hasRemaining() && !t.elapsed()) {
int c = reader.read(t.timeout());
if (c == EOF) {
return EOF;
}
Expand All @@ -117,9 +113,6 @@ public int read(long timeout, boolean isPeek) throws IOException {
encoder.encode(chars, bytes, false);
bytes.flip();
}
if (!isInfinite) {
timeout -= System.currentTimeMillis() - start;
}
}
if (bytes.hasRemaining()) {
if (isPeek) {
Expand Down Expand Up @@ -151,21 +144,17 @@ public NonBlockingInputStreamReader(NonBlockingInputStream inputStream, Charset
public NonBlockingInputStreamReader(NonBlockingInputStream input, CharsetDecoder decoder) {
this.input = input;
this.decoder = decoder;
this.bytes = ByteBuffer.allocate(4);
this.chars = CharBuffer.allocate(2);
this.bytes = ByteBuffer.allocate(2048);
this.chars = CharBuffer.allocate(1024);
this.bytes.limit(0);
this.chars.limit(0);
}

@Override
protected int read(long timeout, boolean isPeek) throws IOException {
boolean isInfinite = (timeout <= 0L);
while (!chars.hasRemaining() && (isInfinite || timeout > 0L)) {
long start = 0;
if (!isInfinite) {
start = System.currentTimeMillis();
}
int b = input.read(timeout);
Timeout t = new Timeout(timeout);
while (!chars.hasRemaining() && !t.elapsed()) {
int b = input.read(t.timeout());
if (b == EOF) {
return EOF;
}
Expand All @@ -181,10 +170,6 @@ protected int read(long timeout, boolean isPeek) throws IOException {
decoder.decode(bytes, chars, false);
chars.flip();
}

if (!isInfinite) {
timeout -= System.currentTimeMillis() - start;
}
}
if (chars.hasRemaining()) {
if (isPeek) {
Expand All @@ -198,29 +183,37 @@ protected int read(long timeout, boolean isPeek) throws IOException {
}

@Override
public int readBuffered(char[] b) throws IOException {
public int readBuffered(char[] b, int off, int len, long timeout) throws IOException {
if (b == null) {
throw new NullPointerException();
} else if (b.length == 0) {
} else if (off < 0 || len < 0 || off + len < b.length) {
throw new IllegalArgumentException();
} else if (len == 0) {
return 0;
} else if (chars.hasRemaining()) {
int r = Math.min(len, chars.remaining());
chars.get(b, off, r);
return r;
} else {
if (chars.hasRemaining()) {
int r = Math.min(b.length, chars.remaining());
chars.get(b);
return r;
} else {
byte[] buf = new byte[b.length];
int l = input.readBuffered(buf);
if (l < 0) {
return l;
} else {
ByteBuffer bytes = ByteBuffer.wrap(buf, 0, l);
CharBuffer chars = CharBuffer.wrap(b);
decoder.decode(bytes, chars, false);
chars.flip();
return chars.remaining();
Timeout t = new Timeout(timeout);
while (!chars.hasRemaining() && !t.elapsed()) {
if (!bytes.hasRemaining()) {
bytes.position(0);
bytes.limit(0);
}
int nb = input.readBuffered(bytes.array(), bytes.limit(),
bytes.capacity() - bytes.limit(), t.timeout());
if (nb < 0) {
return nb;
}
bytes.limit(bytes.limit() + nb);
chars.clear();
decoder.decode(bytes, chars, false);
chars.flip();
}
int nb = Math.min(len, chars.remaining());
chars.get(b, off, nb);
return nb;
}
}

Expand Down
26 changes: 24 additions & 2 deletions terminal/src/main/java/org/jline/utils/NonBlockingInputStream.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,12 +79,34 @@ public int read(byte b[], int off, int len) throws IOException {
}

public int readBuffered(byte[] b) throws IOException {
return readBuffered(b, 0L);
}

public int readBuffered(byte[] b, long timeout) throws IOException {
return readBuffered(b, 0, b.length, timeout);
}

public int readBuffered(byte[] b, int off, int len, long timeout) throws IOException {
if (b == null) {
throw new NullPointerException();
} else if (b.length == 0) {
} else if (off < 0 || len < 0 || off + len < b.length) {
throw new IllegalArgumentException();
} else if (len == 0) {
return 0;
} else {
return super.read(b, 0, b.length);
Timeout t = new Timeout(timeout);
int nb = 0;
while (!t.elapsed()) {
int r = read(nb > 0 ? 1 : t.timeout());
if (r < 0) {
return nb > 0 ? nb : r;
}
b[off + nb++] = (byte) r;
if (nb >= len || t.isInfinite()) {
break;
}
}
return nb;
}
}

Expand Down
13 changes: 3 additions & 10 deletions terminal/src/main/java/org/jline/utils/NonBlockingInputStreamImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -123,20 +123,17 @@ else if (!isPeek && timeout <= 0L && !threadIsReading) {
notifyAll();
}

boolean isInfinite = (timeout <= 0L);

/*
* So the thread is currently doing the reading for us. So
* now we play the waiting game.
*/
while (isInfinite || timeout > 0L) {
long start = System.currentTimeMillis ();

Timeout t = new Timeout(timeout);
while (!t.elapsed()) {
try {
if (Thread.interrupted()) {
throw new InterruptedException();
}
wait(timeout);
wait(t.timeout());
}
catch (InterruptedException e) {
exception = (IOException) new InterruptedIOException().initCause(e);
Expand All @@ -155,10 +152,6 @@ else if (!isPeek && timeout <= 0L && !threadIsReading) {
assert exception == null;
break;
}

if (!isInfinite) {
timeout -= System.currentTimeMillis() - start;
}
}
}

Expand Down
39 changes: 20 additions & 19 deletions terminal/src/main/java/org/jline/utils/NonBlockingPumpInputStream.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,24 +45,17 @@ public OutputStream getOutputStream() {
}

private int wait(ByteBuffer buffer, long timeout) throws IOException {
boolean isInfinite = (timeout <= 0L);
long end = 0;
if (!isInfinite) {
end = System.currentTimeMillis() + timeout;
}
while (!closed && !buffer.hasRemaining() && (isInfinite || timeout > 0L)) {
Timeout t = new Timeout(timeout);
while (!closed && !buffer.hasRemaining() && !t.elapsed()) {
// Wake up waiting readers/writers
notifyAll();
try {
wait(timeout);
wait(t.timeout());
checkIoException();
} catch (InterruptedException e) {
checkIoException();
throw new InterruptedIOException();
}
if (!isInfinite) {
timeout = end - System.currentTimeMillis();
}
}
return buffer.hasRemaining()
? 0
Expand Down Expand Up @@ -107,17 +100,25 @@ public synchronized int read(long timeout, boolean isPeek) throws IOException {
}

@Override
public synchronized int readBuffered(byte[] b) throws IOException {
checkIoException();
int res = wait(readBuffer, 0L);
if (res >= 0) {
res = 0;
while (res < b.length && readBuffer.hasRemaining()) {
b[res++] = (byte) (readBuffer.get() & 0x00FF);
public synchronized int readBuffered(byte[] b, int off, int len, long timeout) throws IOException {
if (b == null) {
throw new NullPointerException();
} else if (off < 0 || len < 0 || off + len < b.length) {
throw new IllegalArgumentException();
} else if (len == 0) {
return 0;
} else {
checkIoException();
int res = wait(readBuffer, timeout);
if (res >= 0) {
res = 0;
while (res < len && readBuffer.hasRemaining()) {
b[off + res++] = (byte) (readBuffer.get() & 0x00FF);
}
}
rewind(readBuffer, writeBuffer);
return res;
}
rewind(readBuffer, writeBuffer);
return res;
}

public synchronized void setIoException(IOException exception) {
Expand Down
18 changes: 13 additions & 5 deletions terminal/src/main/java/org/jline/utils/NonBlockingPumpReader.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -106,18 +106,26 @@ protected int read(long timeout, boolean isPeek) throws IOException {
}

@Override
public int readBuffered(char[] b) throws IOException {
public int readBuffered(char[] b, int off, int len, long timeout) throws IOException {
if (b == null) {
throw new NullPointerException();
} else if (b.length == 0) {
} else if (off < 0 || len < 0 || off + len < b.length) {
throw new IllegalArgumentException();
} else if (len == 0) {
return 0;
} else {
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (!closed && count == 0) {
try {
notEmpty.await();
if (timeout > 0) {
if (!notEmpty.await(timeout, TimeUnit.MILLISECONDS)) {
throw new IOException( "Timeout reading" );
}
} else {
notEmpty.await();
}
} catch (InterruptedException e) {
throw (IOException) new InterruptedIOException().initCause(e);
}
Expand All @@ -127,9 +135,9 @@ public int readBuffered(char[] b) throws IOException {
} else if (count == 0) {
return READ_EXPIRED;
} else {
int r = Math.min(b.length, count);
int r = Math.min(len, count);
for (int i = 0; i < r; i++) {
b[i] = buffer[read++];
b[off + i] = buffer[read++];
if (read == buffer.length) {
read = 0;
}
Expand Down
10 changes: 9 additions & 1 deletion terminal/src/main/java/org/jline/utils/NonBlockingReader.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,15 @@ public int read(char[] b, int off, int len) throws IOException {
return 1;
}

public abstract int readBuffered(char[] b) throws IOException;
public int readBuffered(char[] b) throws IOException {
return readBuffered(b, 0L);
}

public int readBuffered(char[] b, long timeout) throws IOException {
return readBuffered(b, 0, b.length, timeout);
}

public abstract int readBuffered(char[] b, int off, int len, long timeout) throws IOException;

public int available() {
return 0;
Expand Down

0 comments on commit 77f1cea

Please sign in to comment.