@Test @SuppressWarnings ( "unchecked" ) public void errorLoggedIfInterruptedInPause() throws InterruptedException { streamSource.acceptingRowsSemaphore = semaphore; when( semaphore.availablePermits() ).thenReturn( 1 ); streamSource.logChannel = logChannel; doThrow( new InterruptedException( "interrupt" ) ) .when( semaphore ).acquire(); streamSource.pause(); verify( logChannel ).logError( any() ); }
@Override public synchronized void pause() { if ( !paused.getAndSet( true ) ) { try { assert acceptingRowsSemaphore.availablePermits() == 1; acceptingRowsSemaphore.acquire(); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.PauseInterrupt" ) ); } } }
BlockingQueueStreamSource implements StreamSource<T> { @Override public synchronized void pause() { if ( !paused.getAndSet( true ) ) { try { assert acceptingRowsSemaphore.availablePermits() == 1; acceptingRowsSemaphore.acquire(); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.PauseInterrupt" ) ); } } } }
BlockingQueueStreamSource implements StreamSource<T> { @Override public synchronized void pause() { if ( !paused.getAndSet( true ) ) { try { assert acceptingRowsSemaphore.availablePermits() == 1; acceptingRowsSemaphore.acquire(); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.PauseInterrupt" ) ); } } } protected  BlockingQueueStreamSource( BaseStreamStep streamStep ); }
BlockingQueueStreamSource implements StreamSource<T> { @Override public synchronized void pause() { if ( !paused.getAndSet( true ) ) { try { assert acceptingRowsSemaphore.availablePermits() == 1; acceptingRowsSemaphore.acquire(); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.PauseInterrupt" ) ); } } } 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> { @Override public synchronized void pause() { if ( !paused.getAndSet( true ) ) { try { assert acceptingRowsSemaphore.availablePermits() == 1; acceptingRowsSemaphore.acquire(); } catch ( InterruptedException e ) { logChannel.logError( getString( PKG, "BlockingQueueStream.PauseInterrupt" ) ); } } } protected  BlockingQueueStreamSource( BaseStreamStep streamStep ); @Override Observable<T> observable(); @Override void close(); @Override synchronized void pause(); @Override synchronized void resume(); void error( Throwable throwable );  }