PipeChannelOutputStream extends OutputStream { @Override public void write(int b) throws IOException { if (shutdown) { throw new IOException("Stream has been shut down."); } byte[] b1 = new byte[1]; b1[0] = (byte) b; this.write(b1); } PipeChannelOutputStream(@Nonnull PipeChannel pipeChannel, Duration writeTimeout); @Override void write(int b); @Override void write(@NotNull byte[] bs, int off, int len); @Override void close();  }
@Test(timeout = 10000) public void simpleWrite() throws IOException { testChannel = new TestPipeChannel(4096); PipeChannelOutputStream output = new PipeChannelOutputStream(testChannel, Duration.ofSeconds(2)); String msg = "Hello world!"; output.write(msg.getBytes(StandardCharsets.UTF_8)); assertThat(testChannel.readWriteBufferContent()).isEqualTo(msg); }
@Test(timeout = 10000) public void writeSingleByte() throws IOException { testChannel = new TestPipeChannel(4096); PipeChannelOutputStream output = new PipeChannelOutputStream(testChannel, Duration.ofSeconds(2)); String msg = "Hello world!"; for (byte b : msg.getBytes(StandardCharsets.UTF_8)) { output.write(Byte.toUnsignedInt(b)); } assertThat(testChannel.readWriteBufferContent()).isEqualTo(msg); }
@Test(timeout = 10000) public void writeWithOffsetUndLength() throws IOException { testChannel = new TestPipeChannel(4096); PipeChannelOutputStream output = new PipeChannelOutputStream(testChannel, Duration.ofSeconds(2)); byte[] msgBytes = new byte[]{1, 2, 3}; output.write(msgBytes, 0, msgBytes.length - 1); assertThat(testChannel.getWriteBufferContent()).containsExactly((byte) 1, (byte) 2); output.write(msgBytes, 2, msgBytes.length - 2); assertThat(testChannel.getWriteBufferContent()).containsExactly((byte) 3); output.write(msgBytes, 1, 1); assertThat(testChannel.getWriteBufferContent()).containsExactly((byte) 2); }
@Test(timeout = 10000) public void writeTimeout() throws IOException { int writeBufferSize = 4; byte[] msgBytes = "Hello world!".getBytes(StandardCharsets.UTF_8); assertThat(msgBytes.length).isGreaterThan(writeBufferSize); testChannel = new TestPipeChannel(writeBufferSize); PipeChannelOutputStream output = new PipeChannelOutputStream(testChannel, Duration.ofSeconds(1)); assertThat(Stopwatch.measure(() -> { try { output.write(msgBytes); fail("Write operation must not complete."); } catch (WriteTimeoutIOException e) { } catch (IOException e) { fail("Unexpected exception.", e); } }).toMillis()).isCloseTo(1000, Offset.offset(150L)); }
@Test(timeout = 10000) public void infiniteWriteTimeout() throws IOException { int writeBufferSize = 4; byte[] msgBytes = "Hello world!".getBytes(StandardCharsets.UTF_8); assertThat(msgBytes.length).isGreaterThan(writeBufferSize); testChannel = new TestPipeChannel(writeBufferSize); PipeChannelOutputStream output = new PipeChannelOutputStream(testChannel, Duration.ofSeconds(0)); Future<Void> future = runAsync(() -> { try { output.write(msgBytes); fail("Write 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 { int writeBufferSize = 4; byte[] msgBytes = "Hello world!".getBytes(StandardCharsets.UTF_8); assertThat(msgBytes.length).isGreaterThan(writeBufferSize); testChannel = new TestPipeChannel(writeBufferSize); PipeChannelOutputStream output = new PipeChannelOutputStream(testChannel, Duration.ofSeconds(1)); CompletableFuture<Void> futur = runAsync(() -> { try { output.write(msgBytes); } catch (IOException e) { } }); testChannel.shutdownOutputStreams(); futur.get(500, TimeUnit.MILLISECONDS); }
@Test(timeout = 10000) public void invalidWriteParameter() throws IOException { testChannel = new TestPipeChannel(); PipeChannelOutputStream output = new PipeChannelOutputStream(testChannel, Duration.ofSeconds(1)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> output.write(new byte[10], -1, 5)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> output.write(new byte[10], 0, -1)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> output.write(new byte[10], 9, 2)); assertThatExceptionOfType(IndexOutOfBoundsException.class) .isThrownBy(() -> output.write(new byte[10], 11, 0)); assertThatExceptionOfType(NullPointerException.class) .isThrownBy(() -> output.write(null, 0, 0)); }