<commit_before>/*
 * Copyright (c) 2007 Pentaho Corporation.  All rights reserved. 
 * This software was developed by Pentaho Corporation and is provided under the terms 
 * of the GNU Lesser General Public License, Version 2.1. You may not use 
 * this file except in compliance with the license. If you need a copy of the license, 
 * please go to http://www.gnu.org/licenses/lgpl-2.1.txt. The Original Code is Pentaho 
 * Data Integration.  The Initial Developer is Pentaho Corporation.
 *
 * Software distributed under the GNU Lesser Public License is distributed on an "AS IS" 
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or  implied. Please refer to 
 * the license for the specific language governing your rights and limitations.
*/
package org.pentaho.di.trans;

import org.pentaho.di.core.RowSet;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.trans.step.StepInterface;

/**
 * Allows you to "Inject" rows into a step.
 * 
 * @author Matt
 *
 */
public class RowProducer
{
    private RowSet rowSet;
    private StepInterface stepInterface;
    
    public RowProducer(StepInterface stepInterface, RowSet rowSet)
    {
        this.stepInterface = stepInterface;
        this.rowSet = rowSet;
    }
    
    public void putRow(RowMetaInterface rowMeta, Object[] row)
    {
        rowSet.putRow(rowMeta, row);
    }
    
    public void finished()
    {
        rowSet.setDone();
    }

    /**
     * @return Returns the rowSet.
     */
    public RowSet getRowSet()
    {
        return rowSet;
    }

    /**
     * @param rowSet The rowSet to set.
     */
    public void setRowSet(RowSet rowSet)
    {
        this.rowSet = rowSet;
    }

    /**
     * @return Returns the stepInterface.
     */
    public StepInterface getStepInterface()
    {
        return stepInterface;
    }

    /**
     * @param stepInterface The stepInterface to set.
     */
    public void setStepInterface(StepInterface stepInterface)
    {
        this.stepInterface = stepInterface;
    }
    
    
}
<commit_msg>[PDI-5226] RowProducer now blocks for putRow() if the buffer is full.  Also exposed RowSet.putRow(..) and RowSet.putRowWait(..) through RowProducer.<commit_after>/*
 * Copyright (c) 2007 Pentaho Corporation.  All rights reserved. 
 * This software was developed by Pentaho Corporation and is provided under the terms 
 * of the GNU Lesser General Public License, Version 2.1. You may not use 
 * this file except in compliance with the license. If you need a copy of the license, 
 * please go to http://www.gnu.org/licenses/lgpl-2.1.txt. The Original Code is Pentaho 
 * Data Integration.  The Initial Developer is Pentaho Corporation.
 *
 * Software distributed under the GNU Lesser Public License is distributed on an "AS IS" 
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or  implied. Please refer to 
 * the license for the specific language governing your rights and limitations.
*/
package org.pentaho.di.trans;

import java.util.concurrent.TimeUnit;

import org.pentaho.di.core.RowSet;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.trans.step.StepInterface;

/**
 * Allows you to "Inject" rows into a step.
 * 
 * @author Matt
 *
 */
public class RowProducer
{
    private RowSet rowSet;
    private StepInterface stepInterface;
    
    public RowProducer(StepInterface stepInterface, RowSet rowSet)
    {
        this.stepInterface = stepInterface;
        this.rowSet = rowSet;
    }
    
    /**
     * Puts a row into the underlying row set.  This will block until the row is successfully added.
     * 
     * @see #putRow(RowMetaInterface, Object[], boolean) putRow(RowMetaInterface, Object[], true)
     */
    public void putRow(RowMetaInterface rowMeta, Object[] row)
    {
      putRow(rowMeta, row, true);
    }
    
    /**
     * Puts a row on to the underlying row set, optionally blocking until the row can be successfully put.
     * 
     * @return true if the row was successfully added to the rowset and false if this buffer was full.  If {@code block} is true this will always return true. 
     * @see RowSet#putRow(RowMetaInterface, Object[])
     */
    public boolean putRow(RowMetaInterface rowMeta, Object[] row, boolean block) {
      if (block) {
        while (!rowSet.putRowWait(rowMeta, row, Long.MAX_VALUE, TimeUnit.DAYS));
        return true;
      }
      return rowSet.putRow(rowMeta, row);
    }
    
    /**
     * @see RowSet#putRowWait(RowMetaInterface, Object[], long, TimeUnit)
     */
    public boolean putRowWait(RowMetaInterface rowMeta, Object[] rowData, long time, TimeUnit tu)
    {
      return rowSet.putRowWait(rowMeta, rowData, time, tu);
    }
    
    public void finished()
    {
        rowSet.setDone();
    }

    /**
     * @return Returns the rowSet.
     */
    public RowSet getRowSet()
    {
        return rowSet;
    }

    /**
     * @param rowSet The rowSet to set.
     */
    public void setRowSet(RowSet rowSet)
    {
        this.rowSet = rowSet;
    }

    /**
     * @return Returns the stepInterface.
     */
    public StepInterface getStepInterface()
    {
        return stepInterface;
    }

    /**
     * @param stepInterface The stepInterface to set.
     */
    public void setStepInterface(StepInterface stepInterface)
    {
        this.stepInterface = stepInterface;
    }
    
    
}

{'fraction_non_alphanumeric': 0.0847457627118644, 'fraction_numerical': 0.003606202668589975, 'mean_word_length': 2.3496376811594204, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4999961', 'n_tokens_mistral': 1597, 'n_tokens_neox': 1556, 'n_words': 525}