@Test public void testTransferInterrupt() throws Exception { final TransferStatus status = new TransferStatus(); final CyclicBarrier lock = new CyclicBarrier(2); final CyclicBarrier exit = new CyclicBarrier(2); status.setLength(432768L); new Thread(new Runnable() { @Override public void run() { try { new StreamCopier(status, new StreamProgress() { @Override public void progress(final long bytes) { status.progress(bytes); try { lock.await(); exit.await(); } catch(InterruptedException | BrokenBarrierException e) { fail(e.getMessage()); } } @Override public void setComplete() { } }).transfer(new NullInputStream(status.getLength()), new NullOutputStream()); } catch(BackgroundException e) { assertTrue(e instanceof ConnectionCanceledException); } } }).start(); lock.await(); status.setCanceled(); exit.await(); assertFalse(status.isComplete()); try { status.validate(); fail(); } catch(ConnectionCanceledException e) { } assertEquals(32768L, status.getOffset()); }
public void transfer(final InputStream in, final OutputStream out) throws BackgroundException { try { try { if(offset > 0) { skip(in, offset); } final byte[] buffer = new byte[chunksize]; long total = 0; int len = chunksize; if(limit > 0 && limit < chunksize) { len = limit.intValue(); } while(len > 0) { cancel.validate(); final int read = in.read(buffer, 0, len); if(-1 == read) { if(log.isDebugEnabled()) { log.debug(String.format("End of file reached with %d bytes read from stream", total)); } progress.setComplete(); break; } else { listener.recv(read); out.write(buffer, 0, read); progress.progress(read); listener.sent(read); total += read; } if(limit > 0) { len = (int) Math.min(limit - total, chunksize); } if(limit == total) { if(log.isDebugEnabled()) { log.debug(String.format("Limit %d reached reading from stream", limit)); } progress.setComplete(); } } final StreamCloser c = new DefaultStreamCloser(); c.close(out); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { final StreamCloser c = new DefaultStreamCloser(); c.close(in); } } catch(ConnectionCanceledException e) { throw e; } catch(Exception e) { final long sent = listener.getSent(); progress.progress(-sent); listener.sent(-sent); final long recv = listener.getRecv(); listener.recv(-recv); throw e; } cancel.validate(); }
StreamCopier { public void transfer(final InputStream in, final OutputStream out) throws BackgroundException { try { try { if(offset > 0) { skip(in, offset); } final byte[] buffer = new byte[chunksize]; long total = 0; int len = chunksize; if(limit > 0 && limit < chunksize) { len = limit.intValue(); } while(len > 0) { cancel.validate(); final int read = in.read(buffer, 0, len); if(-1 == read) { if(log.isDebugEnabled()) { log.debug(String.format("End of file reached with %d bytes read from stream", total)); } progress.setComplete(); break; } else { listener.recv(read); out.write(buffer, 0, read); progress.progress(read); listener.sent(read); total += read; } if(limit > 0) { len = (int) Math.min(limit - total, chunksize); } if(limit == total) { if(log.isDebugEnabled()) { log.debug(String.format("Limit %d reached reading from stream", limit)); } progress.setComplete(); } } final StreamCloser c = new DefaultStreamCloser(); c.close(out); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { final StreamCloser c = new DefaultStreamCloser(); c.close(in); } } catch(ConnectionCanceledException e) { throw e; } catch(Exception e) { final long sent = listener.getSent(); progress.progress(-sent); listener.sent(-sent); final long recv = listener.getRecv(); listener.recv(-recv); throw e; } cancel.validate(); } }
StreamCopier { public void transfer(final InputStream in, final OutputStream out) throws BackgroundException { try { try { if(offset > 0) { skip(in, offset); } final byte[] buffer = new byte[chunksize]; long total = 0; int len = chunksize; if(limit > 0 && limit < chunksize) { len = limit.intValue(); } while(len > 0) { cancel.validate(); final int read = in.read(buffer, 0, len); if(-1 == read) { if(log.isDebugEnabled()) { log.debug(String.format("End of file reached with %d bytes read from stream", total)); } progress.setComplete(); break; } else { listener.recv(read); out.write(buffer, 0, read); progress.progress(read); listener.sent(read); total += read; } if(limit > 0) { len = (int) Math.min(limit - total, chunksize); } if(limit == total) { if(log.isDebugEnabled()) { log.debug(String.format("Limit %d reached reading from stream", limit)); } progress.setComplete(); } } final StreamCloser c = new DefaultStreamCloser(); c.close(out); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { final StreamCloser c = new DefaultStreamCloser(); c.close(in); } } catch(ConnectionCanceledException e) { throw e; } catch(Exception e) { final long sent = listener.getSent(); progress.progress(-sent); listener.sent(-sent); final long recv = listener.getRecv(); listener.recv(-recv); throw e; } cancel.validate(); } StreamCopier(final StreamCancelation cancel, final StreamProgress progress); }
StreamCopier { public void transfer(final InputStream in, final OutputStream out) throws BackgroundException { try { try { if(offset > 0) { skip(in, offset); } final byte[] buffer = new byte[chunksize]; long total = 0; int len = chunksize; if(limit > 0 && limit < chunksize) { len = limit.intValue(); } while(len > 0) { cancel.validate(); final int read = in.read(buffer, 0, len); if(-1 == read) { if(log.isDebugEnabled()) { log.debug(String.format("End of file reached with %d bytes read from stream", total)); } progress.setComplete(); break; } else { listener.recv(read); out.write(buffer, 0, read); progress.progress(read); listener.sent(read); total += read; } if(limit > 0) { len = (int) Math.min(limit - total, chunksize); } if(limit == total) { if(log.isDebugEnabled()) { log.debug(String.format("Limit %d reached reading from stream", limit)); } progress.setComplete(); } } final StreamCloser c = new DefaultStreamCloser(); c.close(out); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { final StreamCloser c = new DefaultStreamCloser(); c.close(in); } } catch(ConnectionCanceledException e) { throw e; } catch(Exception e) { final long sent = listener.getSent(); progress.progress(-sent); listener.sent(-sent); final long recv = listener.getRecv(); listener.recv(-recv); throw e; } cancel.validate(); } StreamCopier(final StreamCancelation cancel, final StreamProgress progress); StreamCopier withChunksize(final Integer chunksize); StreamCopier withListener(final StreamListener listener); StreamCopier withLimit(final Long limit); StreamCopier withOffset(final Long offset); void transfer(final InputStream in, final OutputStream out); static InputStream skip(final InputStream in, final long offset); }
StreamCopier { public void transfer(final InputStream in, final OutputStream out) throws BackgroundException { try { try { if(offset > 0) { skip(in, offset); } final byte[] buffer = new byte[chunksize]; long total = 0; int len = chunksize; if(limit > 0 && limit < chunksize) { len = limit.intValue(); } while(len > 0) { cancel.validate(); final int read = in.read(buffer, 0, len); if(-1 == read) { if(log.isDebugEnabled()) { log.debug(String.format("End of file reached with %d bytes read from stream", total)); } progress.setComplete(); break; } else { listener.recv(read); out.write(buffer, 0, read); progress.progress(read); listener.sent(read); total += read; } if(limit > 0) { len = (int) Math.min(limit - total, chunksize); } if(limit == total) { if(log.isDebugEnabled()) { log.debug(String.format("Limit %d reached reading from stream", limit)); } progress.setComplete(); } } final StreamCloser c = new DefaultStreamCloser(); c.close(out); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { final StreamCloser c = new DefaultStreamCloser(); c.close(in); } } catch(ConnectionCanceledException e) { throw e; } catch(Exception e) { final long sent = listener.getSent(); progress.progress(-sent); listener.sent(-sent); final long recv = listener.getRecv(); listener.recv(-recv); throw e; } cancel.validate(); } StreamCopier(final StreamCancelation cancel, final StreamProgress progress); StreamCopier withChunksize(final Integer chunksize); StreamCopier withListener(final StreamListener listener); StreamCopier withLimit(final Long limit); StreamCopier withOffset(final Long offset); void transfer(final InputStream in, final OutputStream out); static InputStream skip(final InputStream in, final long offset);  }