@Test public void testError() { final String exceptionMessage = "Exception raised during acceptRows loop"; streamSource = new BlockingQueueStreamSource<String>( streamStep ) { @Override public void open() { execSvc.submit( () -> { for ( int i = 0; i < 10; i++ ) { acceptRows( singletonList( "new row " + i ) ); try { Thread.sleep( 5 ); } catch ( InterruptedException e ) { fail(); } if ( i == 5 ) { error( new RuntimeException( exceptionMessage ) ); break; } } } ); } }; streamSource.open(); Iterator<String> iterator = streamSource.observable().blockingIterable().iterator(); Future<List<String>> iterLoop = execSvc.submit( () -> { List<String> strings = new ArrayList<>(); do { strings.add( iterator.next() ); } while ( strings.size() < 9 ); return strings; } ); try { iterLoop.get( 50, MILLISECONDS ); fail( "expected exception" ); } catch ( InterruptedException | ExecutionException | TimeoutException e ) { if ( e != null && e.getCause() != null ) { assertThat( e.getCause().getMessage(), equalTo( exceptionMessage ) ); } } }
public void error( Throwable throwable ) { publishProcessor.onError( throwable ); }
BlockingQueueStreamSource implements StreamSource<T> { public void error( Throwable throwable ) { publishProcessor.onError( throwable ); } }
BlockingQueueStreamSource implements StreamSource<T> { public void error( Throwable throwable ) { publishProcessor.onError( throwable ); } protected  BlockingQueueStreamSource( BaseStreamStep streamStep ); }
BlockingQueueStreamSource implements StreamSource<T> { public void error( Throwable throwable ) { publishProcessor.onError( throwable ); } protected  BlockingQueueStreamSource( BaseStreamStep streamStep ); @Override Observable<T> observable(); @Override void close(); @Override synchronized void pause(); @Override synchronized void resume(); void error( Throwable throwable ); }
BlockingQueueStreamSource implements StreamSource<T> { public void error( Throwable throwable ) { publishProcessor.onError( throwable ); } protected  BlockingQueueStreamSource( BaseStreamStep streamStep ); @Override Observable<T> observable(); @Override void close(); @Override synchronized void pause(); @Override synchronized void resume(); void error( Throwable throwable );  }