### Question:
NotifyingOutputStream extends OutputStream { @Override public void write(int singleByte) throws IOException { underlyingOutputStream.write(singleByte); countAndCallListener(1); } NotifyingOutputStream(
      OutputStream underlyingOutputStream, Consumer<Long> byteCountListener); @Override void write(int singleByte); @Override void write(byte[] byteArray); @Override void write(byte[] byteArray, int offset, int length); @Override void flush(); @Override void close();  }### Answer:
@Test public void testCallback_correctSequence() throws IOException { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); List<Long> byteCounts = new ArrayList<>(); try (NotifyingOutputStream notifyingOutputStream = new NotifyingOutputStream(byteArrayOutputStream, byteCounts::add)) { notifyingOutputStream.write(0); notifyingOutputStream.write(new byte[] {1, 2, 3}); notifyingOutputStream.write(new byte[] {1, 2, 3, 4, 5}, 3, 2); } Assert.assertEquals(Arrays.asList(1L, 3L, 2L), byteCounts); Assert.assertArrayEquals(new byte[] {0, 1, 2, 3, 4, 5}, byteArrayOutputStream.toByteArray()); }
@Test public void testDelay() throws IOException { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); List<Long> byteCounts = new ArrayList<>(); Queue<Instant> instantQueue = new ArrayDeque<>(); instantQueue.add(Instant.EPOCH); try (ThrottledAccumulatingConsumer byteCounter = new ThrottledAccumulatingConsumer( byteCounts::add, Duration.ofSeconds(3), instantQueue::remove); NotifyingOutputStream notifyingOutputStream = new NotifyingOutputStream(byteArrayOutputStream, byteCounter)) { instantQueue.add(Instant.EPOCH); notifyingOutputStream.write(100); instantQueue.add(Instant.EPOCH); notifyingOutputStream.write(new byte[] {101, 102, 103}); instantQueue.add(Instant.EPOCH.plusSeconds(4)); notifyingOutputStream.write(new byte[] {104, 105, 106}); instantQueue.add(Instant.EPOCH.plusSeconds(10)); notifyingOutputStream.write(new byte[] {107, 108}); instantQueue.add(Instant.EPOCH.plusSeconds(10)); notifyingOutputStream.write(new byte[] {109}); instantQueue.add(Instant.EPOCH.plusSeconds(13)); notifyingOutputStream.write(new byte[] {0, 110}, 1, 1); } Assert.assertEquals(Arrays.asList(7L, 2L, 2L), byteCounts); Assert.assertArrayEquals( new byte[] {100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110}, byteArrayOutputStream.toByteArray()); }