@Test @SuppressWarnings ( "unchecked" ) public void errorLoggedIfInterruptedInAcceptRows() throws InterruptedException { streamSource.acceptingRowsSemaphore = semaphore; streamSource.logChannel = logChannel; doThrow( new InterruptedException( "interrupt" ) ) .when( semaphore ).acquire(); streamSource.acceptRows( singletonList( "new row" ) ); verify( logChannel ).logError( any() ); verify( semaphore ).release(); }
protected void acceptRows( List<T> rows ) { try { acceptingRowsSemaphore.acquire(); rows.forEach( ( row ) -> { streamStep.incrementLinesInput(); publishProcessor.onNext( row ); } ); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.AcceptRowsInterrupt", Arrays.toString( rows.toArray() ) ) ); } finally { acceptingRowsSemaphore.release(); } }
BlockingQueueStreamSource implements StreamSource<T> { protected void acceptRows( List<T> rows ) { try { acceptingRowsSemaphore.acquire(); rows.forEach( ( row ) -> { streamStep.incrementLinesInput(); publishProcessor.onNext( row ); } ); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.AcceptRowsInterrupt", Arrays.toString( rows.toArray() ) ) ); } finally { acceptingRowsSemaphore.release(); } } }
BlockingQueueStreamSource implements StreamSource<T> { protected void acceptRows( List<T> rows ) { try { acceptingRowsSemaphore.acquire(); rows.forEach( ( row ) -> { streamStep.incrementLinesInput(); publishProcessor.onNext( row ); } ); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.AcceptRowsInterrupt", Arrays.toString( rows.toArray() ) ) ); } finally { acceptingRowsSemaphore.release(); } } protected  BlockingQueueStreamSource( BaseStreamStep streamStep ); }
BlockingQueueStreamSource implements StreamSource<T> { protected void acceptRows( List<T> rows ) { try { acceptingRowsSemaphore.acquire(); rows.forEach( ( row ) -> { streamStep.incrementLinesInput(); publishProcessor.onNext( row ); } ); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.AcceptRowsInterrupt", Arrays.toString( rows.toArray() ) ) ); } finally { acceptingRowsSemaphore.release(); } } 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> { protected void acceptRows( List<T> rows ) { try { acceptingRowsSemaphore.acquire(); rows.forEach( ( row ) -> { streamStep.incrementLinesInput(); publishProcessor.onNext( row ); } ); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.AcceptRowsInterrupt", Arrays.toString( rows.toArray() ) ) ); } finally { acceptingRowsSemaphore.release(); } } protected  BlockingQueueStreamSource( BaseStreamStep streamStep ); @Override Observable<T> observable(); @Override void close(); @Override synchronized void pause(); @Override synchronized void resume(); void error( Throwable throwable );  }