ObservableInputStream extends FilterInputStream { @Override public void close() throws IOException { try { super.close(); } finally { this.observer.close(); } } ObservableInputStream(InputStream in, TransferListener observer); @Override int read(); @Override int read(byte[] b, int off, int len); @Override void close();  }
@Test public void testClose() throws IOException { EventLogHttpTransferListener listener = new EventLogHttpTransferListener(); InputStream in = new ObservableInputStream(new ByteArrayInputStream(new byte[0]), listener); in.close(); assertEquals(of(Closed.INSTANCE), listener.getEventLog()); }
@Test public void testCloseFailed() throws IOException { EventLogHttpTransferListener listener = new EventLogHttpTransferListener(); InputStream in = mock(InputStream.class); doThrow(new IOException("test")).when(in).close(); InputStream oin = new ObservableInputStream(in, listener); try { oin.close(); fail("expected exception not thrown"); } catch (IOException e) { assertEquals("test", e.getMessage()); } assertEquals(of(Closed.INSTANCE), listener.getEventLog()); }
@Test public void testStarted() throws IOException { EventLogHttpTransferListener listener = new EventLogHttpTransferListener(); final URI uri = URI.create("http: final Response response = new Response.Builder().request(new Request.Builder().url(uri.toURL()).build()).protocol(HTTP_1_1).code(200) .build(); InputStream in = new ObservableResourceInputStream<Response>(new ByteArrayInputStream(new byte[0]), listener, response, 1l); in.close(); assertEquals(of(new Started(uri, 1l), Closed.INSTANCE), listener.getEventLog()); }