import org.pentaho.di.core.RowSet; //导入依赖的package包/类
/**
 * This adds a row producer to the transformation that just got set up.
 * Preferable run this BEFORE execute() but after prepareExcution()
 *
 * @param stepname The step to produce rows for
 * @param copynr The copynr of the step to produce row for (normally 0 unless you have multiple copies running)
 * @throws KettleException in case the thread/step to produce rows for could not be found.
 */
public RowProducer addRowProducer(String stepname, int copynr) throws KettleException
{
    StepInterface stepInterface = getStepInterface(stepname, copynr);
    if (stepInterface==null)
    {
        throw new KettleException("Unable to find thread with name "+stepname+" and copy number "+copynr);
    }

    // We are going to add an extra RowSet to this stepInterface.
    RowSet rowSet = new RowSet(transMeta.getSizeRowset());

    // Add this rowset to the list of active rowsets for the selected step
    stepInterface.getInputRowSets().add(rowSet);

    return new RowProducer(stepInterface, rowSet);
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public void setConnectorSteps(StepInterface[] sourceSteps, List<MappingValueRename> valueRenames, String mappingStepname) {
	
       for (int i=0;i<sourceSteps.length;i++) {
       	
       	// We don't want to add the mapping-to-mapping rowset
       	//
       	if (!sourceSteps[i].isMapping()) {
	        // OK, before we leave, make sure there is a rowset that covers the path to this target step.
	        // We need to create a new RowSet and add it to the Input RowSets of the target step
        	//
	        RowSet rowSet = new RowSet(getTransMeta().getSizeRowset());
	        
	        // This is always a single copy, both for source and target...
	        //
	        rowSet.setThreadNameFromToCopy(sourceSteps[i].getStepname(), 0, mappingStepname, 0);
	        
	        // Make sure to connect it to both sides...
	        //
	        sourceSteps[i].getOutputRowSets().add(rowSet);
	        getInputRowSets().add(rowSet);
       	}
       }
       data.valueRenames = valueRenames;
       
	data.sourceSteps = sourceSteps;
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public void identifyErrorOutput() {
	if (stepMeta.isDoingErrorHandling()) {
           StepErrorMeta stepErrorMeta = stepMeta.getStepErrorMeta();
           boolean stop=false;
           for (int rowsetNr=0;rowsetNr<outputRowSets.size() && !stop;rowsetNr++)
           {
               RowSet outputRowSet = outputRowSets.get(rowsetNr);
               if (outputRowSet.getDestinationStepName().equalsIgnoreCase(stepErrorMeta.getTargetStep().getName()))
               {
                   // This is the rowset to move!
               	//
                   errorRowSet = outputRowSet;
                   outputRowSets.remove(rowsetNr);
                   stop=true;
               }
           }
	}
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
/**
    * Opens socket connections to the remote input steps of this step.
    * <br>This method should be used by steps that don't call getRow() first in which it is executed automatically.
    * <br><b>This method should be called before any data is read from previous steps.</b>
    * <br>This action is executed only once.
    * @throws KettleStepException
    */
   protected void openRemoteInputStepSocketsOnce() throws KettleStepException {
       if (!remoteInputSteps.isEmpty()) {
       	if (!remoteInputStepsInitialized) {
       		// Loop over the remote steps and open client sockets to them 
       		// Just be careful in case we're dealing with a partitioned clustered step.
       		// A partitioned clustered step has only one. (see dispatch())
       		// 
       		for (RemoteStep remoteStep : remoteInputSteps) {
       			try {
					RowSet rowSet = remoteStep.openReaderSocket(this);
					inputRowSets.add(rowSet);
				} catch (Exception e) {
					throw new KettleStepException("Error opening reader socket to remote step '"+remoteStep+"'", e);
				}
       		}
       		remoteInputStepsInitialized = true;
       	}
       }
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public RowSet findOutputRowSet(String targetStep) throws KettleStepException {
	
   	// Check to see that "targetStep" only runs in a single copy
   	// Otherwise you'll see problems during execution.
   	//
   	StepMeta targetStepMeta = transMeta.findStep(targetStep);
   	if (targetStepMeta==null) {
   		throw new KettleStepException(Messages.getString("BaseStep.Exception.TargetStepToWriteToDoesntExist", targetStep));
   	}
   	
   	if (targetStepMeta.getCopies()>1) {
   		throw new KettleStepException(Messages.getString("BaseStep.Exception.TargetStepToWriteToCantRunInMultipleCopies", targetStep, Integer.toString(targetStepMeta.getCopies())));
   	}
   	

	return findOutputRowSet(getStepname(), getCopy(), targetStep, 0);
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public boolean canProcessOneRow() {
	switch(inputRowSets.size()) {
	case 0: return false;
	case 1:
		RowSet set = inputRowSets.get(0);
		if (set.isDone()) return false;
		return set.size()>0;
	default: 
		boolean allDone=true;
		for (RowSet rowSet : inputRowSets) {
			if (!rowSet.isDone()) allDone=false;
			if (rowSet.size()>0) {
				return true;
			}
		}
		return !allDone;
	}
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
@Override
public void batchComplete() throws KettleException {
  RowSet rowSet = getInputRowSets().get(0);
  int repeats = 0;
  for (int i=0;i<data.cache.length;i++) {
    if (repeats==0) repeats=1;
    if (data.cache[i]!=null) {
      repeats *= data.cache[i].size();
    }
  }
  while (rowSet.size()>0 && !isStopped()) {
    processRow(meta, data);
  }
  // The last row needs to be written too to the account of the number of input rows.
  //
  for (int i=0;i<repeats;i++) {
    processRow(meta, data);
  }
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public RowSet findRowSet(String rowsetname)
{
	// Start with the transformation.
	for (int i=0;i<rowsets.size();i++)
	{
		//log.logDetailed("DIS: looking for RowSet ["+rowsetname+"] in nr "+i+" of "+threads.size()+" threads...");
		RowSet rs=rowsets.get(i);
		if (rs.getName().equalsIgnoreCase(rowsetname)) return rs;
	}

	return null;
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public boolean init(StepMetaInterface smi, StepDataInterface sdi)
 {
     meta=(XMLJoinMeta)smi;
     data=(XMLJoinData)sdi;
     if(!super.init(smi, sdi))
         return false;
     
     

    try {
 	   setSerializer(TransformerFactory.newInstance().newTransformer());
 	   if(meta.getEncoding()!=null) {
        	getSerializer().setOutputProperty(OutputKeys.ENCODING, meta.getEncoding());
        }
        
        if(meta.isOmitXMLHeader()) {
            getSerializer().setOutputProperty(OutputKeys.OMIT_XML_DECLARATION, "yes");
        }
        
 	   this.setSafeModeEnabled(false);
 	   // See if a main step is supplied: in that case move the corresponding rowset to position 0
for (int i=0;i<inputRowSets.size();i++)
{
    RowSet rs = (RowSet) inputRowSets.get(i);
    if (rs.getOriginStepName().equalsIgnoreCase(meta.getTargetXMLstep()))
    {
        // swap this one and position 0...
                // That means, the main stream is always stream 0 --> easy!
                //
        RowSet zero = (RowSet)inputRowSets.get(0);
        inputRowSets.set(0, rs);
        inputRowSets.set(i, zero);
    }
}
     } catch (Exception e) {
         return false;
     }

     return true;
 }
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
@Override
public int rowsetOutputSize()
{
    int size = 0;
    for (MappingOutput output : data.mappingTrans.findMappingOutput())
    {
        for (RowSet rowSet : output.getOutputRowSets())
        {
            size += rowSet.size();
        }
    }
    return size;
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
protected RowSet get(Object value)
{
    String valueStr = (String)value;
    for (String key : list)
    {
        if (valueStr.contains(key))
            return map.get(key);
    }
    return null;
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public void setConnectorSteps(StepInterface[] targetSteps, List<MappingValueRename> inputValueRenames, List<MappingValueRename> outputValueRenames)
{
    for (int i=0;i<targetSteps.length;i++) {
     	
     // OK, before we leave, make sure there is a rowset that covers the path to this target step.
     // We need to create a new RowSet and add it to the Input RowSets of the target step
    	//
     RowSet rowSet = new RowSet(getTransMeta().getSizeRowset());
     
     // This is always a single copy, but for source and target...
     //
     rowSet.setThreadNameFromToCopy(getStepname(), 0, targetSteps[i].getStepname(), 0);
     
     // Make sure to connect it to both sides...
     //
     getOutputRowSets().add(rowSet);
     
     // Add the row set to the target step as input.
     // This will appropriately drain the buffer as data comes in.
     // However, as an exception, we can't attach it to another mapping step.
     // We need to attach it to the appropriate mapping input step.
     // The main problem is that we can't do it here since we don't know that the other step has initialized properly yet.
     // This method is called during init() and we can't tell for sure it's done already.
     // As such, we'll simply grab the remaining row sets at the Mapping#processRow() level and assign them to a Mapping Input step. 
     //
    	targetSteps[i].getInputRowSets().add(rowSet);
    }
    
    data.inputValueRenames = inputValueRenames;
    data.outputValueRenames = outputValueRenames;
    data.targetSteps = targetSteps;
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
private synchronized RowMetaAndData readStartDate() throws KettleException
  {
if (log.isDetailed()) logDetailed("Reading from step [" + meta.getLookupStepname() + "]");

      RowMetaInterface parametersMeta = new RowMeta();
      Object[] parametersData = new Object[] {};

      RowSet rowSet = findInputRowSet(meta.getLookupStepname());
      if (rowSet!=null) 
      {
       Object[] rowData = getRowFrom(rowSet); // rows are originating from "lookup_from"
       while (rowData!=null)
       {
           parametersData = RowDataUtil.addRowData(parametersData, parametersMeta.size(), rowData);
           parametersMeta.addRowMeta(rowSet.getRowMeta());
           
           rowData = getRowFrom(rowSet); // take all input rows if needed!
       }
       
       if (parametersMeta.size()==0)
       {
           throw new KettleException("Expected to read parameters from step ["+meta.getLookupStepname()+"] but none were found.");
       }
      }
      else
      {
          throw new KettleException("Unable to find rowset to read from, perhaps step ["+meta.getLookupStepname()+"] doesn't exist. (or perhaps you are trying a preview?)");
      }
	
      RowMetaAndData parameters = new RowMetaAndData(parametersMeta, parametersData);

      return parameters;
  }
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public boolean outputIsDone()
{
    int nrstopped = 0;

    for (RowSet rs : outputRowSets)
    {
        if (rs.isDone()) nrstopped++;
    }
    return nrstopped >= outputRowSets.size();
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
/**
 * @param rowSet
 * @param rowData
 */
public RowSetRow(RowSet rowSet, RowMetaInterface rowMeta, Object[] rowData) {
	super();
	this.rowSet = rowSet;
	this.rowMeta = rowMeta;
	this.rowData = rowData;
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public RowSet findInputRowSet(String sourceStep) throws KettleStepException {
	// Check to see that "sourceStep" only runs in a single copy
	// Otherwise you'll see problems during execution.
	//
	StepMeta sourceStepMeta = transMeta.findStep(sourceStep);
	if (sourceStepMeta==null) {
		throw new KettleStepException(Messages.getString("BaseStep.Exception.SourceStepToReadFromDoesntExist", sourceStep));
	}
	
	if (sourceStepMeta.getCopies()>1) {
		throw new KettleStepException(Messages.getString("BaseStep.Exception.SourceStepToReadFromCantRunInMultipleCopies", sourceStep, Integer.toString(sourceStepMeta.getCopies())));
	}
	
	return findInputRowSet(sourceStep, 0, getStepname(), getCopy());
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public void setOutputDone()
{
    if (log.isDebug()) logDebug(Messages.getString("BaseStep.Log.OutputDone", String.valueOf(outputRowSets.size()))); //$NON-NLS-1$ //$NON-NLS-2$
    synchronized(outputRowSets)
    {
        for (int i = 0; i < outputRowSets.size(); i++)
        {
            RowSet rs = outputRowSets.get(i);
            rs.setDone();
        }
        if (errorRowSet!=null) errorRowSet.setDone();
    }
}
 

import org.pentaho.di.core.RowSet; //导入依赖的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.core.RowSet; //导入依赖的package包/类
public void setOutputDone()
{
    if (log.isDebug()) logDebug(BaseMessages.getString(PKG, "BaseStep.Log.OutputDone", String.valueOf(outputRowSets.size()))); //$NON-NLS-1$ //$NON-NLS-2$
    synchronized(outputRowSets)
    {
        for (int i = 0; i < outputRowSets.size(); i++)
        {
            RowSet rs = outputRowSets.get(i);
            rs.setDone();
        }
        if (errorRowSet!=null) errorRowSet.setDone();
    }
}
 

import org.pentaho.di.core.RowSet; //导入依赖的package包/类
public RowSet findInfoRowSet(String tag) throws KettleException {
	if (tag==null) return null;
	String stepname = data.infoMap.get(tag);
	if (Const.isEmpty(stepname)) {
		throw new KettleException(BaseMessages.getString(PKG, "TransformClassBase.Exception.UnableToFindInfoStepNameForTag", tag));
	}
	RowSet rowSet = findInputRowSet(stepname);
	if (rowSet==null) {
		throw new KettleException(BaseMessages.getString(PKG, "TransformClassBase.Exception.UnableToFindInfoRowSetForStep", stepname));
	}
	return rowSet;
}