@Test public void testLoopUntilTeesAreClosed() throws Exception { final byte[] data = createMockData(10); InputStream source = new ByteArrayInputStream(data); TeeInputStream in = new TeeInputStream(source, 2) { int loopCount = 0; @Override protected void loop() throws Exception { byte[] temp = new byte[10]; int read; super.loop(); switch (loopCount) { case 0: read = getTees()[0].read(temp); assertEquals(10, read); assertArrayEquals(data, temp); break; case 1: read = getTees()[1].read(temp); assertEquals(10, read); assertArrayEquals(data, temp); break; case 2: getTees()[0].close(); break; case 3: getTees()[1].close(); break; } loopCount++; } }; in.loopUntilTeesAreClosed(); }
public void loopUntilTeesAreClosed() throws Exception { boolean allClosed = false; long lastReport = new Date().getTime(); while (!allClosed) { allClosed = true; for (int i = 0; i < tees.length; i++) { if (!isClosed(i)) { allClosed = false; if (LOGGER.isDebugEnabled() && new Date().getTime() > lastReport + LOOP_REPORT_INTERVAL) { MyInputStream teeWithLowestOffset = findTeeWithLowestTeeOffset(); if (teeWithLowestOffset == null) { LOGGER.debug("All tees are complete"); } else { LOGGER.debug("Waiting for tee: %s (offset: %d)", teeWithLowestOffset.splitName, teeWithLowestOffset.offset); } lastReport = new Date().getTime(); } break; } } loop(); } }
TeeInputStream { public void loopUntilTeesAreClosed() throws Exception { boolean allClosed = false; long lastReport = new Date().getTime(); while (!allClosed) { allClosed = true; for (int i = 0; i < tees.length; i++) { if (!isClosed(i)) { allClosed = false; if (LOGGER.isDebugEnabled() && new Date().getTime() > lastReport + LOOP_REPORT_INTERVAL) { MyInputStream teeWithLowestOffset = findTeeWithLowestTeeOffset(); if (teeWithLowestOffset == null) { LOGGER.debug("All tees are complete"); } else { LOGGER.debug("Waiting for tee: %s (offset: %d)", teeWithLowestOffset.splitName, teeWithLowestOffset.offset); } lastReport = new Date().getTime(); } break; } } loop(); } } }
TeeInputStream { public void loopUntilTeesAreClosed() throws Exception { boolean allClosed = false; long lastReport = new Date().getTime(); while (!allClosed) { allClosed = true; for (int i = 0; i < tees.length; i++) { if (!isClosed(i)) { allClosed = false; if (LOGGER.isDebugEnabled() && new Date().getTime() > lastReport + LOOP_REPORT_INTERVAL) { MyInputStream teeWithLowestOffset = findTeeWithLowestTeeOffset(); if (teeWithLowestOffset == null) { LOGGER.debug("All tees are complete"); } else { LOGGER.debug("Waiting for tee: %s (offset: %d)", teeWithLowestOffset.splitName, teeWithLowestOffset.offset); } lastReport = new Date().getTime(); } break; } } loop(); } } TeeInputStream(InputStream source, String[] splitNames);  TeeInputStream(InputStream source, int splits);  TeeInputStream(InputStream source, int splits, int bufferSize);  TeeInputStream(InputStream source, String[] splitNames, int bufferSize); }
TeeInputStream { public void loopUntilTeesAreClosed() throws Exception { boolean allClosed = false; long lastReport = new Date().getTime(); while (!allClosed) { allClosed = true; for (int i = 0; i < tees.length; i++) { if (!isClosed(i)) { allClosed = false; if (LOGGER.isDebugEnabled() && new Date().getTime() > lastReport + LOOP_REPORT_INTERVAL) { MyInputStream teeWithLowestOffset = findTeeWithLowestTeeOffset(); if (teeWithLowestOffset == null) { LOGGER.debug("All tees are complete"); } else { LOGGER.debug("Waiting for tee: %s (offset: %d)", teeWithLowestOffset.splitName, teeWithLowestOffset.offset); } lastReport = new Date().getTime(); } break; } } loop(); } } TeeInputStream(InputStream source, String[] splitNames);  TeeInputStream(InputStream source, int splits);  TeeInputStream(InputStream source, int splits, int bufferSize);  TeeInputStream(InputStream source, String[] splitNames, int bufferSize); InputStream[] getTees(); void close(); void loopUntilTeesAreClosed(); int getMaxNonblockingReadLength(int teeIndex); }
TeeInputStream { public void loopUntilTeesAreClosed() throws Exception { boolean allClosed = false; long lastReport = new Date().getTime(); while (!allClosed) { allClosed = true; for (int i = 0; i < tees.length; i++) { if (!isClosed(i)) { allClosed = false; if (LOGGER.isDebugEnabled() && new Date().getTime() > lastReport + LOOP_REPORT_INTERVAL) { MyInputStream teeWithLowestOffset = findTeeWithLowestTeeOffset(); if (teeWithLowestOffset == null) { LOGGER.debug("All tees are complete"); } else { LOGGER.debug("Waiting for tee: %s (offset: %d)", teeWithLowestOffset.splitName, teeWithLowestOffset.offset); } lastReport = new Date().getTime(); } break; } } loop(); } } TeeInputStream(InputStream source, String[] splitNames);  TeeInputStream(InputStream source, int splits);  TeeInputStream(InputStream source, int splits, int bufferSize);  TeeInputStream(InputStream source, String[] splitNames, int bufferSize); InputStream[] getTees(); void close(); void loopUntilTeesAreClosed(); int getMaxNonblockingReadLength(int teeIndex); static final int LOOP_REPORT_INTERVAL; }