import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * Read the data for this Cache entry from a data input stream
 * @param dis The DataInputStream to read this entry from.
 * @throws KettleFileException if the cache can't be read from disk when it should be able to.  
 * If the cache file doesn't exists, no exception is thrown 
 */
public DBCacheEntry(DataInputStream dis) throws KettleFileException
{
	try
	{
		dbname  = dis.readUTF();
		sql     = dis.readUTF();
	}
	catch(EOFException eof)
	{
		throw new KettleEOFException("End of file reached", eof);
	}
	catch(Exception e)
	{
		throw new KettleFileException("Unable to read cache entry from data input stream", e);
	}
}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
	meta=(CubeInputMeta)smi;
	data=(CubeInputData)sdi;

	try
	{
           Object[] r = data.meta.readData(data.dis);
           putRow(data.meta, r);  // fill the rowset(s). (sleeps if full)
           incrementLinesInput();
		
		if (meta.getRowLimit()>0 && getLinesInput()>=meta.getRowLimit()) // finished!
		{
			setOutputDone();
			return false;
		}
	}
	catch(KettleEOFException eof)
	{
		setOutputDone();
		return false;
	} 
	catch (SocketTimeoutException e) 
	{
		throw new KettleException(e); // shouldn't happen on files
	}

       if (checkFeedback(getLinesInput())) 
       {
       	if(log.isBasic()) logBasic(Messages.getString("CubeInput.Log.LineNumber")+getLinesInput()); //$NON-NLS-1$
       }

	return true;
}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
	meta=(CubeInputMeta)smi;
	data=(CubeInputData)sdi;

	try
	{
           Object[] r = data.meta.readData(data.dis);
           putRow(data.meta, r);  // fill the rowset(s). (sleeps if full)
           incrementLinesInput();
		
		if (meta.getRowLimit()>0 && getLinesInput()>=meta.getRowLimit()) // finished!
		{
			setOutputDone();
			return false;
		}
	}
	catch(KettleEOFException eof)
	{
		setOutputDone();
		return false;
	} 
	catch (SocketTimeoutException e) 
	{
		throw new KettleException(e); // shouldn't happen on files
	}

       if (checkFeedback(getLinesInput())) 
       {
       	if(log.isBasic()) logBasic(BaseMessages.getString(PKG, "CubeInput.Log.LineNumber")+getLinesInput()); //$NON-NLS-1$
       }

	return true;
}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {

    if ( first ) {
      first = false;
      meta = (CubeInputMeta) smi;
      data = (CubeInputData) sdi;
      realRowLimit = Const.toInt( environmentSubstitute( meta.getRowLimit() ), 0 );
    }


    try {
      Object[] r = data.meta.readData( data.dis );
      putRow( data.meta, r ); // fill the rowset(s). (sleeps if full)
      incrementLinesInput();

      if ( realRowLimit > 0 && getLinesInput() >= realRowLimit ) { // finished!
        setOutputDone();
        return false;
      }
    } catch ( KettleEOFException eof ) {
      setOutputDone();
      return false;
    } catch ( SocketTimeoutException e ) {
      throw new KettleException( e ); // shouldn't happen on files
    }

    if ( checkFeedback( getLinesInput() ) ) {
      if ( log.isBasic() ) {
        logBasic( BaseMessages.getString( PKG, "CubeInput.Log.LineNumber" ) + getLinesInput() );
      }
    }

    return true;
  }
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * Read the data for this Cache entry from a data input stream
 *
 * @param dis
 *          The DataInputStream to read this entry from.
 * @throws KettleFileException
 *           if the cache can't be read from disk when it should be able to. If the cache file doesn't exists, no
 *           exception is thrown
 */
public DBCacheEntry( DataInputStream dis ) throws KettleFileException {
  try {
    dbname = dis.readUTF();
    sql = dis.readUTF();
  } catch ( EOFException eof ) {
    throw new KettleEOFException( "End of file reached", eof );
  } catch ( Exception e ) {
    throw new KettleFileException( "Unable to read cache entry from data input stream", e );
  }
}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
@Test
public void testClass() throws IOException, KettleFileException {
  final String dbName = "dbName";
  final String sql = "sql query";
  DBCacheEntry entry = new DBCacheEntry( dbName, sql );
  assertTrue( entry.sameDB( "dbName" ) );
  assertFalse( entry.sameDB( "otherDb" ) );
  assertEquals( dbName.toLowerCase().hashCode() ^ sql.toLowerCase().hashCode(), entry.hashCode() );
  DBCacheEntry otherEntry = new DBCacheEntry();
  assertFalse( otherEntry.sameDB( "otherDb" ) );
  assertEquals( 0, otherEntry.hashCode() );
  assertFalse( entry.equals( otherEntry ) );
  assertFalse( entry.equals( new Object() ) );

  ByteArrayOutputStream baos = new ByteArrayOutputStream();
  DataOutputStream dos = new DataOutputStream( baos );

  dos.writeUTF( dbName );
  dos.writeUTF( sql );

  byte[] bytes = baos.toByteArray();
  InputStream is = new ByteArrayInputStream( bytes );
  DataInputStream dis = new DataInputStream( is );
  DBCacheEntry disEntry = new DBCacheEntry( dis );
  assertTrue( disEntry.equals( entry ) );
  try {
    new DBCacheEntry( dis );
    fail( "Should throw KettleEOFException on EOFException" );
  } catch ( KettleEOFException keofe ) {
    // Ignore
  }

  baos.reset();

  assertTrue( disEntry.write( dos ) );
  assertArrayEquals( bytes, baos.toByteArray() );
}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
private void makeTestExtractDataWithTimestampConversion( RowMetaInterface rowMeta, String str, Date date,
    Timestamp constTimestamp ) throws KettleEOFException, KettleFileException, IOException {
  Object[] rowData = new Object[] { str, date };
  byte[] result = RowMeta.extractData( rowMeta, rowData );
  DataInputStream stream = new DataInputStream( new ByteArrayInputStream( result ) );
  String extractedString = (String) new ValueMetaString().readData( stream );
  Timestamp time = (Timestamp) new ValueMetaTimestamp().readData( stream );
  stream.close();
  assertTrue( str.equals( extractedString ) );
  assertTrue( constTimestamp.equals( time ) );

}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * De-serialize data from an inputstream.  No metadata is read or changed.
 * @param inputStream the input stream to read from 
 * @return a new data object
 * @throws KettleFileException in case a I/O error occurs
 * @throws KettleEOFException When we have read all the data there is to read
 * @throws SocketTimeoutException In case there is a timeout (when set on a socket) during reading
 */
public Object readData(DataInputStream inputStream) throws KettleFileException, KettleEOFException, SocketTimeoutException;
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * @param inputStream
 * @throws KettleFileException
 * @throws KettleEOFException
 * @deprecated
 */
public ValueMeta(DataInputStream inputStream) throws KettleFileException, KettleEOFException {
  super(inputStream);
}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * Read the attributes of this particular value meta object from the specified input stream.
 * Loading the type is not handled here, this should be read from the stream previously!
 * 
 * @param inputStream the input stream to read from
 * @throws KettleFileException In case there was a IO problem
 * @throws KettleEOFException If we reached the end of the stream
 */
public void readMetaData(DataInputStream inputStream) throws KettleFileException, KettleEOFException;
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * @param inputStream
 * @throws KettleFileException
 * @throws KettleEOFException
 * @deprecated
 */
@Deprecated
public ValueMeta( DataInputStream inputStream ) throws KettleFileException, KettleEOFException {
  super( inputStream );
}
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * De-serialize data from an inputstream. No metadata is read or changed.
 *
 * @param inputStream
 *          the input stream to read from
 * @return a new data object
 * @throws KettleFileException
 *           in case a I/O error occurs
 * @throws KettleEOFException
 *           When we have read all the data there is to read
 * @throws SocketTimeoutException
 *           In case there is a timeout (when set on a socket) during reading
 */
Object readData( DataInputStream inputStream ) throws KettleFileException, KettleEOFException,
  SocketTimeoutException;
 

import org.pentaho.di.core.exception.KettleEOFException; //导入依赖的package包/类
/**
 * Read the attributes of this particular value meta object from the specified input stream. Loading the type is not
 * handled here, this should be read from the stream previously!
 *
 * @param inputStream
 *          the input stream to read from
 * @throws KettleFileException
 *           In case there was a IO problem
 * @throws KettleEOFException
 *           If we reached the end of the stream
 */
void readMetaData( DataInputStream inputStream ) throws KettleFileException, KettleEOFException;