PipeChannelInputStream extends InputStream { @Override public int read() throws IOException { byte[] b1 = new byte[1]; int n = this.read(b1); if (n == 1) { return b1[0] & 0xff; } return -1; } PipeChannelInputStream(PipeChannel pipeChannel, Duration readTimeout); @Override int read(); @Override int read(@Nonnull byte[] bs, int off, int len); boolean isOpen(); void shutdown(); @Override int available(); @Override void close();  }
@Test(timeout = 10000) public void simpleRead() throws IOException { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(2)); String msg = "Hello world!"; testChannel.printToReadBuffer(msg); byte[] buffer = new byte[4096]; int n = input.read(buffer); assertThat(new String(buffer, 0, n, StandardCharsets.UTF_8)).isEqualTo(msg); mustBlock(input::read); }
@Test(timeout = 10000) public void readSmallBuffer() throws IOException { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(2)); String msg = "Hello world!"; byte[] msgBytes = msg.getBytes(StandardCharsets.UTF_8); byte[] buffer = new byte[5]; assertThat(buffer.length).isLessThan(msgBytes.length); assertThat(msgBytes.length % buffer.length).isNotZero(); testChannel.putInReadBuffer(msgBytes); ByteArrayOutputStream baos = new ByteArrayOutputStream(); int n; while (baos.size() < msgBytes.length && (n = input.read(buffer)) != -1) { baos.write(buffer, 0, n); } assertThat(baos.toByteArray()).isEqualTo(msgBytes); mustBlock(input::read); }
@Test(timeout = 10000) public void readSingleByte() throws IOException { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(2)); String msg = "Hello world!"; byte[] msgBytes = msg.getBytes(StandardCharsets.UTF_8); testChannel.putInReadBuffer(msgBytes); ByteArrayOutputStream baos = new ByteArrayOutputStream(); int b; while (baos.size() < msgBytes.length && (b = input.read()) != -1) { baos.write(b); } assertThat(baos.toByteArray()).isEqualTo(msgBytes); mustBlock(input::read); }
@Test(timeout = 10000) public void readWithOffsetUndLength() throws IOException { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(2)); byte[] msgBytes = new byte[]{1, 2, 3}; byte[] buffer = new byte[msgBytes.length + 4]; testChannel.putInReadBuffer(msgBytes); int n = input.read(buffer, 2, buffer.length - 2); assertThat(n).isEqualTo(msgBytes.length); assertThat(buffer).containsExactly((byte) 0, (byte) 0, (byte) 1, (byte) 2, (byte) 3, (byte) 0, (byte) 0); Arrays.fill(buffer, (byte) 0); testChannel.exhaustReadBuffer(); testChannel.putInReadBuffer(msgBytes); n = input.read(buffer, 0, 2); assertThat(n).isEqualTo(2); assertThat(buffer).containsExactly((byte) 1, (byte) 2, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0); Arrays.fill(buffer, (byte) 0); testChannel.exhaustReadBuffer(); testChannel.putInReadBuffer(msgBytes); n = input.read(buffer, 2, 2); assertThat(n).isEqualTo(2); assertThat(buffer).containsExactly((byte) 0, (byte) 0, (byte) 1, (byte) 2, (byte) 0, (byte) 0, (byte) 0); }
@Test(timeout = 10000) public void readTimeout() throws IOException { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(1)); assertThat(Stopwatch.measure(() -> { try { input.read(); fail("Read operation must not complete."); } catch (SocketTimeoutException e) { } catch (IOException e) { fail("Read failed.", e); } }).toMillis()).isCloseTo(1000, Offset.offset(150L)); }
@Test(timeout = 10000) public void infiniteReadTimeout() throws IOException { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(0)); Future<Void> future = runAsync(() -> { try { input.read(); fail("Read operation must not complete."); } catch (SocketTimeoutException e) { fail("Unexpected timeout.", e); } catch (IOException e) { fail("Read failed.", e); } }); try { future.get(2, TimeUnit.SECONDS); } catch (InterruptedException | ExecutionException e) { fail("Unexpected failure.", e); } catch (TimeoutException e) { } }
@Test(timeout = 10000) public void targetChannelClosed() throws Exception { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(1)); CompletableFuture<Void> futur = runAsync(() -> { int n = input.read(); assertThat(n).isEqualTo(-1); }); testChannel.shutdownOutputStreams(); futur.get(500, TimeUnit.MILLISECONDS); }
@Test(timeout = 10000) public void invalidReadParameter() throws IOException { testChannel = new TestPipeChannel(); PipeChannelInputStream input = new PipeChannelInputStream(testChannel, Duration.ofSeconds(1)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> input.read(new byte[10], -1, 5)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> input.read(new byte[10], 0, -1)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> input.read(new byte[10], 9, 2)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> input.read(new byte[10], 11, 0)); assertThatExceptionOfType(NullPointerException.class) .isThrownBy(() -> input.read(null, 0, 0)); }