DefaultTransWiringRowAdapter extends RowAdapter { @Override public void rowWrittenEvent( RowMetaInterface rowMeta, Object[] row ) throws KettleStepException { LogChannelInterface log = serviceTrans.getLogChannel(); try { if ( log.isRowLevel() ) { log.logRowlevel( DataServiceConstants.PASSING_ALONG_ROW + rowMeta.getString( row ) ); } } catch ( KettleValueException e ) { } try { Object[] rowData = rowMeta.cloneRow( row ); while ( !rowProducer.putRowWait( rowMeta, rowData, 1, TimeUnit.SECONDS ) && genTrans.isRunning() ) { if ( log.isRowLevel() ) { log.logRowlevel( DataServiceConstants.ROW_BUFFER_IS_FULL_TRYING_AGAIN ); } } } catch ( KettleValueException e ) { throw new KettleStepException( e ); } } DefaultTransWiringRowAdapter( Trans serviceTrans, Trans genTrans, RowProducer rowProducer ); @Override void rowWrittenEvent( RowMetaInterface rowMeta, Object[] row );  }
@Test public void testRowWrittenEventSuccessLogging() throws KettleStepException { when( logChannelInterface.isRowLevel() ).thenReturn( true ); when( rowProducer.putRowWait( rowMetaInterface, row, 1, TimeUnit.SECONDS ) ).thenReturn( false, true ); defaultTransWiringRowAdapter.rowWrittenEvent( rowMetaInterface, row ); verify( rowProducer, times( 2 ) ).putRowWait( same( rowMetaInterface ), eq( row ), eq( 1L ), eq( TimeUnit.SECONDS ) ); verify( logChannelInterface ).logRowlevel( startsWith( DataServiceConstants.PASSING_ALONG_ROW ) ); verify( logChannelInterface ).logRowlevel( DataServiceConstants.ROW_BUFFER_IS_FULL_TRYING_AGAIN ); }
@Test public void testRowWrittenEventSuccessErrorLogging() throws KettleStepException, KettleValueException { when( logChannelInterface.isRowLevel() ).thenReturn( true ); when( rowMetaInterface.getString( row ) ).thenThrow( new KettleValueException() ); when( rowProducer.putRowWait( rowMetaInterface, row, 1, TimeUnit.SECONDS ) ).thenReturn( false, true ); defaultTransWiringRowAdapter.rowWrittenEvent( rowMetaInterface, row ); verify( rowProducer, times( 2 ) ).putRowWait( same( rowMetaInterface ), eq( row ), eq( 1L ), eq( TimeUnit.SECONDS ) ); verify( logChannelInterface ).logRowlevel( DataServiceConstants.ROW_BUFFER_IS_FULL_TRYING_AGAIN ); }
@Test public void testRowWrittenEventSuccessNoLogging() throws KettleStepException { when( rowProducer.putRowWait( rowMetaInterface, row, 1, TimeUnit.SECONDS ) ).thenReturn( false, true ); defaultTransWiringRowAdapter.rowWrittenEvent( rowMetaInterface, row ); verify( rowProducer, times( 2 ) ).putRowWait( same( rowMetaInterface ), eq( row ), eq( 1L ), eq( TimeUnit.SECONDS ) ); verify( logChannelInterface, never() ).logRowlevel( anyString() ); }
@Test( expected = KettleStepException.class ) public void testRowWrittenEventException() throws KettleValueException, KettleStepException { rowMetaInterface = mock( RowMetaInterface.class ); KettleValueException kettleValueException = new KettleValueException(); when( rowMetaInterface.cloneRow( any( Object[].class ) ) ).thenThrow( kettleValueException ); when( rowProducer.putRowWait( rowMetaInterface, row, 1, TimeUnit.SECONDS ) ).thenReturn( false, true ); try { defaultTransWiringRowAdapter.rowWrittenEvent( rowMetaInterface, row ); } catch ( KettleStepException e ) { assertEquals( kettleValueException, e.getCause() ); throw e; } }