Sends the rows asynchronously in MetaInject transform
package org.apache.hop.pipeline.transforms.metainject;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.vfs2.FileObject;
import org.apache.hop.core.Const;
@@ -28,6 +30,7 @@ import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.injection.bean.BeanInjectionInfo;
import org.apache.hop.core.injection.bean.BeanInjector;
import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.util.ExecutorUtil;
import org.apache.hop.core.util.Utils;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.core.vfs.HopVfs;
@@ -158,10 +161,10 @@ public class MetaInject extends BaseTransform<MetaInjectMeta, MetaInjectData> {
// See if we need to stream some data over...
//
- RowProducer rowProducer = null;
- if (data.streaming) {
- rowProducer = injectPipeline.addRowProducer(data.streamingTargetTransformName, 0);
- }
+ RowProducer rowProducer =
+ data.streaming
+ ? injectPipeline.addRowProducer(data.streamingTargetTransformName, 0)
+ : null;
// Finally, add the mapping transformation to the active sub-transformations
// map in the parent transformation
@@ -189,6 +192,7 @@ public class MetaInject extends BaseTransform<MetaInjectMeta, MetaInjectData> {
injectPipeline.startThreads();
+ Future<HopTransformException> steamingFuture = null;
if (data.streaming) {
// Deplete all the rows from the parent transformation into the modified transformation
//
@@ -199,12 +203,7 @@ public class MetaInject extends BaseTransform<MetaInjectMeta, MetaInjectData> {
+ data.streamingSourceTransformName
+ "' to stream data from");
}
- Object[] row = getRowFrom(rowSet);
- while (row != null && !isStopped()) {
- rowProducer.putRow(rowSet.getRowMeta(), row);
- row = getRowFrom(rowSet);
- }
- rowProducer.finished();
+ steamingFuture = ExecutorUtil.getExecutor().submit(() -> putRows(rowSet, rowProducer));
}
// Wait until the child transformation finished processing...
@@ -221,6 +220,15 @@ public class MetaInject extends BaseTransform<MetaInjectMeta, MetaInjectData> {
}
copyResult(injectPipeline);
waitUntilFinished(injectPipeline);
+ try {
+ if (steamingFuture != null && steamingFuture.get() != null) {
+ throw steamingFuture.get();
+ }
+ } catch (ExecutionException ignore) {
+ // Ignore
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
}
// let the transformation complete it's execution to allow for any customizations to MDI to
@@ -240,6 +248,20 @@ public class MetaInject extends BaseTransform<MetaInjectMeta, MetaInjectData> {
return false;
}
+ HopTransformException putRows(IRowSet rowSet, RowProducer rowProducer) {
+ try {
+ Object[] row;
+ while (!isStopped() && (row = getRowFrom(rowSet)) != null) {
+ rowProducer.putRow(rowSet.getRowMeta(), row);
+ }
+ } catch (HopTransformException e) {
+ return e;
+ } finally {
+ rowProducer.finished();
+ }
+ return null;
+ }
+
void waitUntilFinished(Pipeline injectTrans) {
injectTrans.waitUntilFinished();
}
