import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
public MappingInput[] findMappingInput()
  {
if (steps==null) return null;

List<MappingInput> list = new ArrayList<MappingInput>();

      // Look in threads and find the MappingInput step thread...
      for (int i=0;i<steps.size();i++)
      {
          StepMetaDataCombi smdc = steps.get(i);
          StepInterface step = smdc.step;
          if (step.getStepID().equalsIgnoreCase("MappingInput")) //$NON-NLS-1$
          {
              list.add((MappingInput)step);
          }
      }
      return list.toArray(new MappingInput[list.size()]);
  }
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
/**
   * Gets the mapping inputs for each step in the transformation.
   *
   * @return an array of MappingInputs
   */
  public MappingInput[] findMappingInput()
  {
if (steps==null) return null;

List<MappingInput> list = new ArrayList<MappingInput>();

      // Look in threads and find the MappingInput step thread...
      for (int i=0;i<steps.size();i++)
      {
          StepMetaDataCombi smdc = steps.get(i);
          StepInterface step = smdc.step;
          if (step.getStepID().equalsIgnoreCase("MappingInput")) //$NON-NLS-1$
          {
              list.add((MappingInput)step);
          }
      }
      return list.toArray(new MappingInput[list.size()]);
  }
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
/**
 * Gets the mapping inputs for each step in the transformation.
 *
 * @return an array of MappingInputs
 */
public MappingInput[] findMappingInput() {
  if ( steps == null ) {
    return null;
  }

  List<MappingInput> list = new ArrayList<>();

  // Look in threads and find the MappingInput step thread...
  for ( int i = 0; i < steps.size(); i++ ) {
    StepMetaDataCombi smdc = steps.get( i );
    StepInterface step = smdc.step;
    if ( step.getStepID().equalsIgnoreCase( "MappingInput" ) ) {
      list.add( (MappingInput) step );
    }
  }
  return list.toArray( new MappingInput[list.size()] );
}
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
@Override
public int rowsetInputSize()
{
    int size = 0;
    for (MappingInput input : data.mappingTrans.findMappingInput())
    {
        for (RowSet rowSet : input.getInputRowSets())
        {
            size += rowSet.size();
        }
    }
    return size;
}
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
@Override
public int rowsetInputSize() {
  int size = 0;
  for ( MappingInput input : getData().getMappingTrans().findMappingInput() ) {
    for ( RowSet rowSet : input.getInputRowSets() ) {
      size += rowSet.size();
    }
  }
  return size;
}
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
@Test
public void testDispose( ) throws Exception {

  MappingMeta meta = Mockito.mock( MappingMeta.class );
  MappingData data = Mockito.mock( MappingData.class );

  Mockito.when( data.getMappingTrans() ).thenReturn( mockHelper.trans );

  MappingInput[] mappingInputs = { Mockito.mock( MappingInput.class ) };
  MappingOutput[] mappingOutputs = { Mockito.mock( MappingOutput.class ) };
  Mockito.when( mockHelper.trans.findMappingInput() ).thenReturn( mappingInputs );
  Mockito.when( mockHelper.trans.findMappingOutput() ).thenReturn( mappingOutputs );

  data.mappingTransMeta = mockHelper.transMeta;
  Mockito.when( data.mappingTransMeta.getTransformationType() ).thenReturn( TransMeta.TransformationType.SingleThreaded );

  data.singleThreadedTransExcecutor = Mockito.mock( SingleThreadedTransExecutor.class );
  Mockito.when( data.singleThreadedTransExcecutor.oneIteration() ).thenReturn( true );

  data.mappingTrans = mockHelper.trans;
  Mockito.when( mockHelper.trans.isFinished() ).thenReturn( false );
  Mapping mapping = Mockito.spy( new Mapping( mockHelper.stepMeta, data, 0, mockHelper.transMeta, mockHelper.trans ) );
  String stepName = "StepName";
  mapping.setStepname( stepName );


  mapping.processRow( meta, data );
  mapping.dispose( meta, data );
  Mockito.verify( mockHelper.trans, Mockito.times( 1 ) ).removeActiveSubTransformation( stepName );

}
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  stepMockHelper =
      new StepMockHelper<SimpleMappingMeta, SimpleMappingData>( "SIMPLE_MAPPING_TEST", SimpleMappingMeta.class,
          SimpleMappingData.class );
  when( stepMockHelper.logChannelInterfaceFactory.create( any(), any( LoggingObjectInterface.class ) ) ).thenReturn(
      stepMockHelper.logChannelInterface );
  when( stepMockHelper.trans.isRunning() ).thenReturn( true );

  // Mock for MappingInput
  MappingInput mpInputMock = mock( MappingInput.class );
  when( mpInputMock.getStepname() ).thenReturn( MAPPING_INPUT_STEP_NAME );

  // Mock for MappingOutput
  MappingOutput mpOutputMock = mock( MappingOutput.class );
  when( mpOutputMock.getStepname() ).thenReturn( MAPPING_OUTPUT_STEP_NAME );

  // Mock for RowDataInputMapper
  RowDataInputMapper rdInputMpMock = mock( RowDataInputMapper.class );
  RowMetaInterface rwMetaInMock = mock( RowMeta.class );
  doReturn( Boolean.TRUE ).when( rdInputMpMock ).putRow( rwMetaInMock, new Object[] { } );

  // Mock for RowProducer
  RowProducer rProducerMock = mock( RowProducer.class );
  when( rProducerMock.putRow( any( RowMetaInterface.class ), any( Object[].class ), anyBoolean() ) )
    .thenReturn( true );

  // Mock for MappingIODefinition
  MappingIODefinition mpIODefMock = mock( MappingIODefinition.class );

  // Set up real SimpleMappingData with some mocked elements
  simpleMpData.mappingInput = mpInputMock;
  simpleMpData.mappingOutput = mpOutputMock;
  simpleMpData.rowDataInputMapper = rdInputMpMock;
  simpleMpData.mappingTrans = stepMockHelper.trans;

  when( stepMockHelper.trans.findStepInterface( MAPPING_OUTPUT_STEP_NAME, 0 ) ).thenReturn( mpOutputMock );
  when( stepMockHelper.trans.addRowProducer( MAPPING_INPUT_STEP_NAME, 0 ) ).thenReturn( rProducerMock );
  when( stepMockHelper.processRowsStepMetaInterface.getInputMapping() ).thenReturn( mpIODefMock );
}
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
@Test
public void testStepShouldStopProcessingInput_IfUnderlyingTransitionIsStopped() throws Exception {

  MappingInput mappingInput = mock( MappingInput.class );
  when( mappingInput.getStepname() ).thenReturn( MAPPING_INPUT_STEP_NAME );
  stepMockHelper.processRowsStepDataInterface.mappingInput = mappingInput;

  RowProducer rowProducer = mock( RowProducer.class );
  when( rowProducer.putRow( any( RowMetaInterface.class ), any( Object[].class ), anyBoolean() ) )
    .thenReturn( true );

  StepInterface stepInterface = mock( StepInterface.class );

  Trans mappingTrans = mock( Trans.class );
  when( mappingTrans.addRowProducer( anyString(), anyInt() ) ).thenReturn( rowProducer );
  when( mappingTrans.findStepInterface( anyString(), anyInt() ) ).thenReturn( stepInterface );
  when( mappingTrans.isFinishedOrStopped() ).thenReturn( Boolean.FALSE ).thenReturn( Boolean.TRUE );
  stepMockHelper.processRowsStepDataInterface.mappingTrans = mappingTrans;

  MappingOutput mappingOutput = mock( MappingOutput.class );
  when( mappingOutput.getStepname() ).thenReturn( MAPPING_OUTPUT_STEP_NAME );
  stepMockHelper.processRowsStepDataInterface.mappingOutput = mappingOutput;


  smp = new SimpleMapping( stepMockHelper.stepMeta, stepMockHelper.stepDataInterface, 0, stepMockHelper.transMeta,
    stepMockHelper.trans );
  smp.init( stepMockHelper.initStepMetaInterface, simpleMpData );
  smp.getInputRowSets().add( stepMockHelper.getMockInputRowSet( new Object[] { } ) );
  smp.getInputRowSets().add( stepMockHelper.getMockInputRowSet( new Object[] { } ) );

  assertTrue(
    smp.processRow( stepMockHelper.processRowsStepMetaInterface, stepMockHelper.processRowsStepDataInterface ) );
  assertFalse(
    smp.processRow( stepMockHelper.processRowsStepMetaInterface, stepMockHelper.processRowsStepDataInterface ) );

}
 

import org.pentaho.di.trans.steps.mappinginput.MappingInput; //导入依赖的package包/类
/**
    * Process a single row.  In our case, we send one row of data to a piece of transformation.
    * In the transformation, we look up the MappingInput step to send our rows to it.
    * As a consequence, for the time being, there can only be one MappingInput and one MappingOutput step in the Mapping.
    */
public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
	try
	{
		meta=(MappingMeta)smi;
		data=(MappingData)sdi;
		
		// Before we start, let's see if there are loose ends to tie up...
		//
		if (!inputRowSets.isEmpty()) {
			MappingInput[] mappingInputs = data.mappingTrans.findMappingInput();
			for (RowSet rowSet : new ArrayList<RowSet>(inputRowSets)) {
				// Pass this rowset down to a mapping input step in the sub-transformation...
				//
				if (mappingInputs.length==1) {
					// Simple case: only one input mapping.  Move the RowSet over
					//
					inputRowSets.remove(rowSet);
					mappingInputs[0].getInputRowSets().add(rowSet);
				} else {
					// Difficult to see what's going on here.
					// TODO: figure out where this RowSet needs to go and where it comes from.
					//
					throw new KettleException("Unsupported situation detected.  To solve it, insert a dummy step.");
				}
			}
		}
		
		// Start the mapping/sub-transformation threads
        //
        data.mappingTrans.startThreads();
        
        // The transformation still runs in the background and might have some more work to do.
        // Since everything is running in the MappingThreads we don't have to do anything else here but wait...
        //
        data.mappingTrans.waitUntilFinished();
        
        // Set some statistics from the mapping...
        // This will show up in Spoon, etc.
        //
    	Result result = data.mappingTrans.getResult();
    	setErrors(result.getNrErrors());
    	setLinesRead( result.getNrLinesRead() );
    	setLinesWritten( result.getNrLinesWritten() );
    	setLinesInput( result.getNrLinesInput() );
    	setLinesOutput( result.getNrLinesOutput() );
    	setLinesUpdated( result.getNrLinesUpdated() );
    	setLinesRejected( result.getNrLinesRejected() );
    	
    	return false;
	}
	catch(Throwable t)
	{
		// Some unexpected situation occurred.
		// Better to stop the mapping transformation.
		//
		if (data.mappingTrans!=null) data.mappingTrans.stopAll();
		
		// Forward the exception...
		//
		throw new KettleException(t);
	}
}