class 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 );

}