This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 029cd1eb40 field support in delay row transform #4276 (#5971)
029cd1eb40 is described below
commit 029cd1eb403d88a1eb4c85e2073cef2facb59e4c
Author: Bart Maertens <[email protected]>
AuthorDate: Fri Nov 14 16:58:07 2025 +0100
field support in delay row transform #4276 (#5971)
* delay row transform accepts numeric values and time scales from fields
instead of only static values and variables. fixes #4276
* minor cleanup
---------
Co-authored-by: Hans Van Akelyen <[email protected]>
---
.../ROOT/pages/pipeline/transforms/delay.adoc | 5 +-
.../0079-delay-rows-default-behavior.hpl | 96 +++++++
.../transforms/0079-delay-rows-from-field.hpl | 270 ++++++++++++++++++++
.../0079-delay-rows-invalid-time-scale.hpl | 107 ++++++++
.../transforms/main-0079-delay-rows.hwf | 180 +++++++++++++
.../hop/pipeline/transforms/delay/Delay.java | 277 +++++++++++++++++----
.../hop/pipeline/transforms/delay/DelayData.java | 23 +-
.../hop/pipeline/transforms/delay/DelayDialog.java | 216 ++++++++++++++--
.../hop/pipeline/transforms/delay/DelayMeta.java | 72 ++++--
.../delay/messages/messages_en_US.properties | 26 ++
.../pipeline/transforms/delay/DelayMetaTest.java | 14 +-
.../transforms/delay/DelayTransformTest.java | 228 +++++++++++++++++
12 files changed, 1416 insertions(+), 98 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/delay.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/delay.adoc
index 2bf3d95fd7..1266dd74ed 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/delay.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/delay.adoc
@@ -46,5 +46,8 @@ Use this transform if you deliberately want to slow down your
pipeline.
|===
|Option|Description
|Transform name|Name of the transform. This name has to be unique in a single
pipeline.
-|Timeout|How long to delay. Select whether this period is measured in
milliseconds, seconds, minutes, or hours.
+|Timeout|How long to delay. You can type a static value, use a variable, or
pick a numeric field from the incoming stream. Values are interpreted together
with the selected time scale. If the selected field is null or empty, the
transform treats it as zero (no delay).
+|Time scale|Select whether the timeout is expressed in milliseconds, seconds,
minutes, or hours.
+|Get time scale from field|Enable this option to pick the time scale from a
field in the incoming stream.
+|Time scale field|Field that contains the time scale value. Accepted values
(case-insensitive) are: `ms`, `millisecond(s)`, `s`, `second(s)`, `m`,
`minute(s)`, `h`, `hour(s)`. Empty values default to seconds.
|===
\ No newline at end of file
diff --git a/integration-tests/transforms/0079-delay-rows-default-behavior.hpl
b/integration-tests/transforms/0079-delay-rows-default-behavior.hpl
new file mode 100644
index 0000000000..922126fa32
--- /dev/null
+++ b/integration-tests/transforms/0079-delay-rows-default-behavior.hpl
@@ -0,0 +1,96 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<pipeline>
+ <info>
+ <name>delay-rows-default-behavior</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2025/11/06 15:35:09.875</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2025/11/06 15:35:09.875</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Delay row</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Delay row</name>
+ <type>Delay</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <scaletime>seconds</scaletime>
+ <scaletime_from_field>N</scaletime_from_field>
+ <timeout>5</timeout>
+ <use_scaletime_fallback>N</use_scaletime_fallback>
+ <use_timeout_fallback>N</use_timeout_fallback>
+ <attributes/>
+ <GUI>
+ <xloc>480</xloc>
+ <yloc>256</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+</fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field>FiveSecondsAgo</last_time_field>
+ <limit>10</limit>
+ <never_ending>N</never_ending>
+ <row_time_field>now</row_time_field>
+ <attributes/>
+ <GUI>
+ <xloc>192</xloc>
+ <yloc>256</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/0079-delay-rows-from-field.hpl
b/integration-tests/transforms/0079-delay-rows-from-field.hpl
new file mode 100644
index 0000000000..631b2e794b
--- /dev/null
+++ b/integration-tests/transforms/0079-delay-rows-from-field.hpl
@@ -0,0 +1,270 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<pipeline>
+ <info>
+ <name>delay-rows-from-field</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2025/11/07 10:42:34.704</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2025/11/07 10:42:34.704</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>generate 10 rows</from>
+ <to>add counter</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>counter = 5?</from>
+ <to>Delay row</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>counter = 5?</from>
+ <to>5 -> null</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>5 -> null</from>
+ <to>Delay row</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>add counter</from>
+ <to>add time scales</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>add time scales</from>
+ <to>counter = 5?</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>add counter</name>
+ <type>Sequence</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <connection/>
+ <counter_name/>
+ <increment_by>1</increment_by>
+ <max_value>999999999</max_value>
+ <schema/>
+ <seqname>SEQ_</seqname>
+ <start_at>1</start_at>
+ <use_counter>Y</use_counter>
+ <use_database>N</use_database>
+ <valuename>counter</valuename>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>256</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Delay row</name>
+ <type>Delay</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <scaletime>seconds</scaletime>
+ <scaletime_field>scale</scaletime_field>
+ <scaletime_from_field>Y</scaletime_from_field>
+ <timeout_field>counter</timeout_field>
+ <attributes/>
+ <GUI>
+ <xloc>736</xloc>
+ <yloc>256</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>generate 10 rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+</fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field>FiveSecondsAgo</last_time_field>
+ <limit>10</limit>
+ <never_ending>N</never_ending>
+ <row_time_field>now</row_time_field>
+ <attributes/>
+ <GUI>
+ <xloc>160</xloc>
+ <yloc>256</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>5 -> null</name>
+ <type>NullIf</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>counter</name>
+ <value>5</value>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>352</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>add time scales</name>
+ <type>ValueMapper</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <field_to_use>counter</field_to_use>
+ <fields>
+ <field>
+ <source_value>1</source_value>
+ <target_value>s</target_value>
+ </field>
+ <field>
+ <source_value>2</source_value>
+ <target_value>second</target_value>
+ </field>
+ <field>
+ <source_value>3</source_value>
+ <target_value>seconds</target_value>
+ </field>
+ <field>
+ <source_value>4</source_value>
+ <target_value>ms</target_value>
+ </field>
+ <field>
+ <source_value>5</source_value>
+ <target_value>milliseconds</target_value>
+ </field>
+ <field>
+ <source_value>6</source_value>
+ <target_value>MS</target_value>
+ </field>
+ <field>
+ <source_value>7</source_value>
+ <target_value>Second</target_value>
+ </field>
+ <field>
+ <source_value>8</source_value>
+ <target_value>minutes</target_value>
+ </field>
+ <field>
+ <source_value>9</source_value>
+ <target_value>sec</target_value>
+ </field>
+ <field>
+ <source_value>10</source_value>
+ <target_value>seconds</target_value>
+ </field>
+ </fields>
+ <non_match_default/>
+ <target_field>scale</target_field>
+ <attributes/>
+ <GUI>
+ <xloc>480</xloc>
+ <yloc>256</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>counter = 5?</name>
+ <type>FilterRows</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <compare>
+ <condition>
+ <conditions>
+</conditions>
+ <function>=</function>
+ <leftvalue>counter</leftvalue>
+ <negated>N</negated>
+ <operator>-</operator>
+ <value>
+ <isnull>N</isnull>
+ <length>-1</length>
+ <mask>####0;-####0</mask>
+ <name>constant</name>
+ <precision>0</precision>
+ <text>5</text>
+ <type>Integer</type>
+ </value>
+ </condition>
+ </compare>
+ <send_false_to>Delay row</send_false_to>
+ <send_true_to>5 -> null</send_true_to>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>256</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git
a/integration-tests/transforms/0079-delay-rows-invalid-time-scale.hpl
b/integration-tests/transforms/0079-delay-rows-invalid-time-scale.hpl
new file mode 100644
index 0000000000..c94e52e595
--- /dev/null
+++ b/integration-tests/transforms/0079-delay-rows-invalid-time-scale.hpl
@@ -0,0 +1,107 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<pipeline>
+ <info>
+ <name>delay-rows-invalid-time-scale</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2025/11/07 11:23:15.376</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2025/11/07 11:23:15.376</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>generate 1 row</from>
+ <to>Delay row</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>generate 1 row</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <currency/>
+ <decimal/>
+ <format/>
+ <group/>
+ <length>-1</length>
+ <name>time_scale</name>
+ <nullif>year</nullif>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <type>String</type>
+ </field>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field>FiveSecondsAgo</last_time_field>
+ <limit>1</limit>
+ <never_ending>N</never_ending>
+ <row_time_field>now</row_time_field>
+ <attributes/>
+ <GUI>
+ <xloc>160</xloc>
+ <yloc>224</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Delay row</name>
+ <type>Delay</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <scaletime>seconds</scaletime>
+ <scaletime_field>time_scale</scaletime_field>
+ <scaletime_from_field>Y</scaletime_from_field>
+ <timeout>1</timeout>
+ <attributes/>
+ <GUI>
+ <xloc>352</xloc>
+ <yloc>224</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/main-0079-delay-rows.hwf
b/integration-tests/transforms/main-0079-delay-rows.hwf
new file mode 100644
index 0000000000..0b2699b62e
--- /dev/null
+++ b/integration-tests/transforms/main-0079-delay-rows.hwf
@@ -0,0 +1,180 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<workflow>
+ <name>main-0079-delay-rows</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2025/11/07 11:51:39.749</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2025/11/07 11:51:39.749</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <doNotWaitOnFirstExecution>N</doNotWaitOnFirstExecution>
+ <hour>12</hour>
+ <intervalMinutes>60</intervalMinutes>
+ <intervalSeconds>0</intervalSeconds>
+ <minutes>0</minutes>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <weekDay>1</weekDay>
+ <parallel>N</parallel>
+ <xloc>64</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0079-delay-rows-default-behavior.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <clear_files>N</clear_files>
+ <clear_rows>N</clear_rows>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+ <filename>${PROJECT_HOME}/0079-delay-rows-default-behavior.hpl</filename>
+ <logext/>
+ <logfile/>
+ <loglevel>Basic</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>320</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0079-delay-rows-from-field.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <clear_files>N</clear_files>
+ <clear_rows>N</clear_rows>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+ <filename>${PROJECT_HOME}/0079-delay-rows-from-field.hpl</filename>
+ <logext/>
+ <logfile/>
+ <loglevel>Basic</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>608</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0079-delay-rows-invalid-time-scale.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <clear_files>N</clear_files>
+ <clear_rows>N</clear_rows>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+
<filename>${PROJECT_HOME}/0079-delay-rows-invalid-time-scale.hpl</filename>
+ <logext/>
+ <logfile/>
+ <loglevel>Basic</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>896</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Success</name>
+ <description/>
+ <type>SUCCESS</type>
+ <attributes/>
+ <parallel>N</parallel>
+ <xloc>1120</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>0079-delay-rows-default-behavior.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>0079-delay-rows-default-behavior.hpl</from>
+ <to>0079-delay-rows-from-field.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0079-delay-rows-from-field.hpl</from>
+ <to>0079-delay-rows-invalid-time-scale.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0079-delay-rows-invalid-time-scale.hpl</from>
+ <to>Success</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/Delay.java
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/Delay.java
index 7c6793f7a1..20960f62d0 100644
---
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/Delay.java
+++
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/Delay.java
@@ -17,8 +17,13 @@
package org.apache.hop.pipeline.transforms.delay;
+import java.util.Locale;
import org.apache.hop.core.Const;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopValueException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.util.Utils;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
@@ -51,80 +56,242 @@ public class Delay extends BaseTransform<DelayMeta,
DelayData> {
return false;
}
+ IRowMeta rowMeta = getInputRowMeta();
+
if (first) {
first = false;
+ initializeMeta(meta, rowMeta);
+ }
+
+ data.timeout = resolveTimeout(meta, r);
+ data.multiple = resolveScaleMultiple(meta, r);
- String msgScale;
- switch (meta.getScaleTimeCode()) {
- case 0:
- msgScale = BaseMessages.getString(PKG,
"DelayDialog.MSScaleTime.Label");
- data.Multiple = 1;
- break;
- case 1:
- msgScale = BaseMessages.getString(PKG,
"DelayDialog.SScaleTime.Label");
- data.Multiple = 1000;
- break;
- case 2:
- msgScale = BaseMessages.getString(PKG,
"DelayDialog.MnScaleTime.Label");
- data.Multiple = 60000;
- break;
- case 3:
- msgScale = BaseMessages.getString(PKG,
"DelayDialog.HrScaleTime.Label");
- data.Multiple = 3600000;
- break;
- default:
- msgScale = "Unknown Scale";
- data.Multiple = 1;
+ if (data.timeout <= 0 || data.multiple <= 0) {
+ putRow(rowMeta, r);
+ if (checkFeedback(getLinesRead()) && isDetailed()) {
+ logDetailed(BaseMessages.getString(PKG, "Delay.Log.LineNumber", "" +
getLinesRead()));
}
+ return true;
+ }
+
+ long delayMillis;
+ try {
+ delayMillis = Math.multiplyExact(data.timeout, data.multiple);
+ } catch (ArithmeticException e) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG,
+ "Delay.Log.DelayOverflow",
+ Long.toString(data.timeout),
+ Long.toString(data.multiple)),
+ e);
+ }
+
+ if (isDebug()) {
+ logDebug(
+ BaseMessages.getString(
+ PKG,
+ "Delay.Log.TimeOutWithScale",
+ String.valueOf(data.timeout),
+ getScaleLabel(data.multiple),
+ String.valueOf(delayMillis)));
+ }
- String timeOut = resolve(meta.getTimeout());
- data.timeout = Const.toInt(timeOut, 0);
+ applyDelay(delayMillis);
- if (isDebug()) {
- logDebug(BaseMessages.getString(PKG, "Delay.Log.TimeOut", "" +
data.timeout, msgScale));
+ if (isDebug()) {
+ logDebug(BaseMessages.getString(PKG, "Delay.WaitTimeIsElapsed.Label"));
+ }
+
+ putRow(rowMeta, r);
+
+ if (checkFeedback(getLinesRead()) && isDetailed()) {
+ logDetailed(BaseMessages.getString(PKG, "Delay.Log.LineNumber", "" +
getLinesRead()));
+ }
+
+ return true;
+ }
+
+ private void initializeMeta(DelayMeta meta, IRowMeta rowMeta) throws
HopException {
+ data.staticTimeout = Const.toLong(resolve(meta.getTimeout()), 0L);
+ data.staticScaleTimeCode = meta.getScaleTimeCode();
+ data.multiple = determineMultiple(data.staticScaleTimeCode);
+ if (!Utils.isEmpty(meta.getTimeoutField())) {
+ data.timeoutFieldIndex = rowMeta.indexOfValue(meta.getTimeoutField());
+ if (data.timeoutFieldIndex < 0) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "Delay.Log.TimeoutFieldNotFound",
meta.getTimeoutField()));
}
+ data.timeoutValueMeta = rowMeta.getValueMeta(data.timeoutFieldIndex);
+ if (!data.timeoutValueMeta.isNumeric()
+ && data.timeoutValueMeta.getType() != IValueMeta.TYPE_STRING) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG,
+ "Delay.Log.TimeoutFieldType",
+ meta.getTimeoutField(),
+ data.timeoutValueMeta.getTypeDesc()));
+ }
+ } else {
+ data.timeoutFieldIndex = -1;
+ data.timeoutValueMeta = null;
}
- if ((data.Multiple < 1000) && (data.timeout > 0)) {
- // handle the milliseconds delays here
- try {
- Thread.sleep(data.timeout);
- } catch (Exception e) {
- // nothing
+ if (meta.isScaleTimeFromField()) {
+ if (Utils.isEmpty(meta.getScaleTimeField())) {
+ throw new HopException(BaseMessages.getString(PKG,
"Delay.Log.ScaleTimeFieldMissing"));
+ }
+ data.scaleTimeFieldIndex =
rowMeta.indexOfValue(meta.getScaleTimeField());
+ if (data.scaleTimeFieldIndex < 0) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "Delay.Log.ScaleTimeFieldNotFound",
meta.getScaleTimeField()));
+ }
+ data.scaleTimeValueMeta = rowMeta.getValueMeta(data.scaleTimeFieldIndex);
+ if (data.scaleTimeValueMeta.getType() != IValueMeta.TYPE_STRING) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG,
+ "Delay.Log.ScaleTimeFieldType",
+ meta.getScaleTimeField(),
+ data.scaleTimeValueMeta.getTypeDesc()));
}
} else {
- // starttime (in seconds ,Minutes or Hours)
- long timeStart = System.currentTimeMillis();
-
- boolean continueLoop = true;
-
- while (continueLoop && !isStopped()) {
- // Update Time value
- long now = System.currentTimeMillis();
-
- // Let's check the limit time
- if (now >= (timeStart + (data.timeout * data.Multiple))) {
- // We have reached the time limit
- continueLoop = false;
- } else {
- try {
- Thread.sleep(1000);
- } catch (Exception e) {
- // handling this exception would be kind of silly.
+ data.scaleTimeFieldIndex = -1;
+ data.scaleTimeValueMeta = null;
+ }
+ }
+
+ private long resolveTimeout(DelayMeta meta, Object[] row) throws
HopException {
+ if (data.timeoutFieldIndex < 0) {
+ return Math.max(0L, data.staticTimeout);
+ }
+
+ Object timeoutValue = row[data.timeoutFieldIndex];
+ if (data.timeoutValueMeta.isNull(timeoutValue)) {
+ return 0L;
+ }
+
+ try {
+ Double number = data.timeoutValueMeta.getNumber(timeoutValue);
+ if (number == null) {
+ String str = data.timeoutValueMeta.getString(timeoutValue);
+ if (Utils.isEmpty(str)) {
+ return 0L;
+ }
+ number = Double.valueOf(str);
+ }
+ long timeout = Math.round(number);
+ if (timeout < 0) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "Delay.Log.TimeoutFieldNegative", meta.getTimeoutField(),
timeout));
+ }
+ return timeout;
+ } catch (NumberFormatException | HopValueException e) {
+ throw new HopException(
+ BaseMessages.getString(PKG, "Delay.Log.TimeoutFieldNotNumeric",
meta.getTimeoutField()),
+ e);
+ }
+ }
+
+ private long resolveScaleMultiple(DelayMeta meta, Object[] row) throws
HopException {
+ int scaleCode = data.staticScaleTimeCode;
+
+ if (data.scaleTimeFieldIndex >= 0) {
+ scaleCode = 1; // default to seconds when using a field
+ Object scaleValue = row[data.scaleTimeFieldIndex];
+ if (!data.scaleTimeValueMeta.isNull(scaleValue)) {
+ String value = data.scaleTimeValueMeta.getString(scaleValue);
+ if (!Utils.isEmpty(value)) {
+ Integer parsedCode = parseScaleTimeCode(value);
+ if (parsedCode == null) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "Delay.Log.ScaleTimeFieldInvalid",
meta.getScaleTimeField(), value));
}
+ scaleCode = parsedCode;
}
}
}
- if (isDebug()) {
- logDebug(BaseMessages.getString(PKG, "Delay.WaitTimeIsElapsed.Label"));
+
+ return determineMultiple(scaleCode);
+ }
+
+ private long determineMultiple(int scaleCode) {
+ switch (scaleCode) {
+ case 0:
+ return 1L;
+ case 1:
+ return 1000L;
+ case 2:
+ return 60000L;
+ case 3:
+ return 3600000L;
+ default:
+ return 1000L;
+ }
+ }
+
+ private String getScaleLabel(long multiple) {
+ if (multiple == 1L) {
+ return BaseMessages.getString(PKG, "DelayDialog.MSScaleTime.Label");
+ }
+ if (multiple == 1000L) {
+ return BaseMessages.getString(PKG, "DelayDialog.SScaleTime.Label");
}
+ if (multiple == 60000L) {
+ return BaseMessages.getString(PKG, "DelayDialog.MnScaleTime.Label");
+ }
+ if (multiple == 3600000L) {
+ return BaseMessages.getString(PKG, "DelayDialog.HrScaleTime.Label");
+ }
+ return BaseMessages.getString(PKG, "Delay.Log.UnknownScale");
+ }
- putRow(getInputRowMeta(), r); // copy row to possible alternate rowset(s).
+ private Integer parseScaleTimeCode(String value) {
+ String normalized = value.trim().toLowerCase(Locale.ROOT);
+ switch (normalized) {
+ case "ms", "msec", "millisecond", "milliseconds":
+ return 0;
+ case "s", "sec", "secs", "second", "seconds":
+ return 1;
+ case "m", "min", "mins", "minute", "minutes":
+ return 2;
+ case "h", "hr", "hour", "hours":
+ return 3;
+ default:
+ return null;
+ }
+ }
- if (checkFeedback(getLinesRead()) && isDetailed()) {
- logDetailed(BaseMessages.getString(PKG, "Delay.Log.LineNumber", "" +
getLinesRead()));
+ private void applyDelay(long delayMillis) {
+ if (delayMillis <= 0) {
+ return;
}
- return true;
+ if (delayMillis < 1000L) {
+ try {
+ Thread.sleep(delayMillis);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return;
+ }
+
+ long endTime = System.currentTimeMillis() + delayMillis;
+ while (!isStopped()) {
+ long remaining = endTime - System.currentTimeMillis();
+ if (remaining <= 0) {
+ break;
+ }
+ long sleepChunk = Math.min(remaining, 1000L);
+ try {
+ Thread.sleep(sleepChunk);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
}
}
diff --git
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayData.java
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayData.java
index f3811db906..ebedbabe91 100644
---
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayData.java
+++
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayData.java
@@ -18,16 +18,27 @@
package org.apache.hop.pipeline.transforms.delay;
import org.apache.hop.pipeline.transform.BaseTransformData;
-import org.apache.hop.pipeline.transform.ITransformData;
@SuppressWarnings("java:S1104")
-public class DelayData extends BaseTransformData implements ITransformData {
- public int Multiple;
- public int timeout;
+public class DelayData extends BaseTransformData {
+ public long multiple;
+ public long timeout;
+ public long staticTimeout;
+ public int staticScaleTimeCode;
+ public int timeoutFieldIndex;
+ public int scaleTimeFieldIndex;
+ public org.apache.hop.core.row.IValueMeta timeoutValueMeta;
+ public org.apache.hop.core.row.IValueMeta scaleTimeValueMeta;
public DelayData() {
super();
- Multiple = 1000;
- timeout = 0;
+ multiple = 1000L;
+ timeout = 0L;
+ staticTimeout = 0L;
+ staticScaleTimeCode = 1;
+ timeoutFieldIndex = -1;
+ scaleTimeFieldIndex = -1;
+ timeoutValueMeta = null;
+ scaleTimeValueMeta = null;
}
}
diff --git
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayDialog.java
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayDialog.java
index 3bf1299326..21c3b09632 100644
---
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayDialog.java
+++
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayDialog.java
@@ -17,17 +17,28 @@
package org.apache.hop.pipeline.transforms.delay;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.util.Utils;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.ui.core.PropsUi;
import org.apache.hop.ui.core.dialog.BaseDialog;
-import org.apache.hop.ui.core.widget.LabelTextVar;
+import org.apache.hop.ui.core.dialog.ErrorDialog;
+import org.apache.hop.ui.core.widget.ComboVar;
import org.apache.hop.ui.pipeline.transform.BaseTransformDialog;
import org.eclipse.swt.SWT;
import org.eclipse.swt.custom.CCombo;
+import org.eclipse.swt.events.FocusAdapter;
+import org.eclipse.swt.events.FocusEvent;
import org.eclipse.swt.events.ModifyListener;
+import org.eclipse.swt.events.SelectionAdapter;
+import org.eclipse.swt.events.SelectionEvent;
+import org.eclipse.swt.graphics.Point;
import org.eclipse.swt.layout.FormAttachment;
import org.eclipse.swt.layout.FormData;
import org.eclipse.swt.layout.FormLayout;
@@ -40,9 +51,13 @@ public class DelayDialog extends BaseTransformDialog {
private static final Class<?> PKG = DelayMeta.class;
private final DelayMeta input;
+ private ComboVar wTimeout;
private CCombo wScaleTime;
-
- private LabelTextVar wTimeout;
+ private Button wScaleTimeFromField;
+ private Label wlScaleTimeField;
+ private ComboVar wScaleTimeField;
+ private String[] numericFieldNames = new String[0];
+ private String[] stringFieldNames = new String[0];
public DelayDialog(
Shell parent, IVariables variables, DelayMeta transformMeta,
PipelineMeta pipelineMeta) {
@@ -90,24 +105,38 @@ public class DelayDialog extends BaseTransformDialog {
fdTransformName.right = new FormAttachment(100, 0);
wTransformName.setLayoutData(fdTransformName);
- // Timeout line
- wTimeout =
- new LabelTextVar(
- variables,
- shell,
- BaseMessages.getString(PKG, "DelayDialog.Timeout.Label"),
- BaseMessages.getString(PKG, "DelayDialog.Timeout.Tooltip"));
+ // Timeout label and combo
+ Label wlTimeout = new Label(shell, SWT.RIGHT);
+ wlTimeout.setText(BaseMessages.getString(PKG,
"DelayDialog.Timeout.Label"));
+ wlTimeout.setToolTipText(BaseMessages.getString(PKG,
"DelayDialog.Timeout.Tooltip"));
+ PropsUi.setLook(wlTimeout);
+ FormData fdlTimeout = new FormData();
+ fdlTimeout.left = new FormAttachment(0, 0);
+ fdlTimeout.right = new FormAttachment(middle, -margin);
+ fdlTimeout.top = new FormAttachment(wTransformName, margin);
+ wlTimeout.setLayoutData(fdlTimeout);
+
+ wTimeout = new ComboVar(variables, shell, SWT.SINGLE | SWT.LEFT |
SWT.BORDER);
+ wTimeout.setToolTipText(BaseMessages.getString(PKG,
"DelayDialog.Timeout.Tooltip"));
PropsUi.setLook(wTimeout);
- wTimeout.addModifyListener(lsMod);
+ wTimeout.addModifyListener(
+ e -> {
+ input.setChanged();
+ wTimeout.setToolTipText(variables.resolve(wTimeout.getText()));
+ });
+ wTimeout.addFocusListener(
+ new FocusAdapter() {
+ @Override
+ public void focusGained(FocusEvent e) {
+ refreshTimeoutFieldItems();
+ }
+ });
FormData fdTimeout = new FormData();
- fdTimeout.left = new FormAttachment(0, 0);
+ fdTimeout.left = new FormAttachment(middle, 0);
fdTimeout.top = new FormAttachment(wTransformName, margin);
fdTimeout.right = new FormAttachment(100, 0);
wTimeout.setLayoutData(fdTimeout);
- // Whenever something changes, set the tooltip to the expanded version:
- wTimeout.addModifyListener(e ->
wTimeout.setToolTipText(variables.resolve(wTimeout.getText())));
-
wScaleTime = new CCombo(shell, SWT.SINGLE | SWT.READ_ONLY | SWT.BORDER);
wScaleTime.add(BaseMessages.getString(PKG,
"DelayDialog.MSScaleTime.Label"));
wScaleTime.add(BaseMessages.getString(PKG,
"DelayDialog.SScaleTime.Label"));
@@ -122,13 +151,59 @@ public class DelayDialog extends BaseTransformDialog {
wScaleTime.setLayoutData(fdScaleTime);
wScaleTime.addModifyListener(lsMod);
+ wScaleTimeFromField = new Button(shell, SWT.CHECK);
+ wScaleTimeFromField.setText(
+ BaseMessages.getString(PKG, "DelayDialog.ScaleTimeFromField.Label"));
+ PropsUi.setLook(wScaleTimeFromField);
+ FormData fdScaleTimeFromField = new FormData();
+ fdScaleTimeFromField.left = new FormAttachment(middle, 0);
+ fdScaleTimeFromField.top = new FormAttachment(wScaleTime, margin);
+ wScaleTimeFromField.setLayoutData(fdScaleTimeFromField);
+ wScaleTimeFromField.addSelectionListener(
+ new SelectionAdapter() {
+ @Override
+ public void widgetSelected(SelectionEvent selectionEvent) {
+ input.setChanged();
+ enableScaleTimeControls();
+ }
+ });
+
+ wlScaleTimeField = new Label(shell, SWT.RIGHT);
+ wlScaleTimeField.setText(BaseMessages.getString(PKG,
"DelayDialog.ScaleTimeField.Label"));
+ wlScaleTimeField.setToolTipText(
+ BaseMessages.getString(PKG, "DelayDialog.ScaleTimeField.Tooltip"));
+ PropsUi.setLook(wlScaleTimeField);
+ FormData fdlScaleTimeField = new FormData();
+ fdlScaleTimeField.left = new FormAttachment(0, 0);
+ fdlScaleTimeField.right = new FormAttachment(middle, -margin);
+ fdlScaleTimeField.top = new FormAttachment(wScaleTimeFromField, margin);
+ wlScaleTimeField.setLayoutData(fdlScaleTimeField);
+
+ wScaleTimeField = new ComboVar(variables, shell, SWT.SINGLE | SWT.LEFT |
SWT.BORDER);
+ wScaleTimeField.setToolTipText(
+ BaseMessages.getString(PKG, "DelayDialog.ScaleTimeField.Tooltip"));
+ PropsUi.setLook(wScaleTimeField);
+ wScaleTimeField.addModifyListener(lsMod);
+ wScaleTimeField.addFocusListener(
+ new FocusAdapter() {
+ @Override
+ public void focusGained(FocusEvent e) {
+ refreshScaleTimeFieldItems();
+ }
+ });
+ FormData fdScaleTimeField = new FormData();
+ fdScaleTimeField.left = new FormAttachment(middle, 0);
+ fdScaleTimeField.top = new FormAttachment(wScaleTimeFromField, margin);
+ fdScaleTimeField.right = new FormAttachment(100, 0);
+ wScaleTimeField.setLayoutData(fdScaleTimeField);
+
// Some buttons
wOk = new Button(shell, SWT.PUSH);
wOk.setText(BaseMessages.getString(PKG, "System.Button.OK"));
wCancel = new Button(shell, SWT.PUSH);
wCancel.setText(BaseMessages.getString(PKG, "System.Button.Cancel"));
- setButtonPositions(new Button[] {wOk, wCancel}, margin, wScaleTime);
+ setButtonPositions(new Button[] {wOk, wCancel}, margin, wScaleTimeField);
// Add listeners
wCancel.addListener(SWT.Selection, e -> cancel());
@@ -143,10 +218,24 @@ public class DelayDialog extends BaseTransformDialog {
/** Copy information from the meta-data input to the dialog fields. */
public void getData() {
- if (input.getTimeout() != null) {
+ loadAvailableFields();
+ wTimeout.setItems(numericFieldNames);
+ wScaleTimeField.setItems(stringFieldNames);
+
+ if (!Utils.isEmpty(input.getTimeoutField())) {
+ wTimeout.setText(input.getTimeoutField());
+ } else if (input.getTimeout() != null) {
wTimeout.setText(input.getTimeout());
}
+ wTimeout.setToolTipText(variables.resolve(wTimeout.getText()));
+
wScaleTime.select(input.getScaleTimeCode());
+ wScaleTimeFromField.setSelection(input.isScaleTimeFromField());
+ if (input.getScaleTimeField() != null) {
+ wScaleTimeField.setText(input.getScaleTimeField());
+ }
+
+ enableScaleTimeControls();
wTransformName.selectAll();
wTransformName.setFocus();
}
@@ -162,8 +251,99 @@ public class DelayDialog extends BaseTransformDialog {
return;
}
transformName = wTransformName.getText(); // return value
- input.setTimeout(wTimeout.getText());
+ String timeoutField = determineTimeoutField();
+ if (timeoutField != null) {
+ input.setTimeoutField(timeoutField);
+ input.setTimeout(null);
+ } else {
+ input.setTimeoutField(null);
+ input.setTimeout(wTimeout.getText());
+ }
input.setScaleTimeCode(wScaleTime.getSelectionIndex());
+ input.setScaleTimeFromField(wScaleTimeFromField.getSelection());
+ input.setScaleTimeField(
+ wScaleTimeFromField.getSelection() &&
!Utils.isEmpty(wScaleTimeField.getText())
+ ? wScaleTimeField.getText()
+ : null);
dispose();
}
+
+ private String determineTimeoutField() {
+ loadAvailableFields();
+ String candidate = wTimeout.getText();
+ if (Utils.isEmpty(candidate)) {
+ return null;
+ }
+ for (String field : numericFieldNames) {
+ if (candidate.equals(field)) {
+ return field;
+ }
+ }
+ return null;
+ }
+
+ private void enableScaleTimeControls() {
+ boolean useField = wScaleTimeFromField.getSelection();
+ wScaleTime.setEnabled(!useField);
+ wScaleTimeField.setEnabled(useField);
+ wlScaleTimeField.setEnabled(useField);
+ }
+
+ private void loadAvailableFields() {
+ try {
+ IRowMeta prevFields = pipelineMeta.getPrevTransformFields(variables,
transformName);
+ if (prevFields == null) {
+ numericFieldNames = new String[0];
+ stringFieldNames = new String[0];
+ return;
+ }
+
+ List<String> numeric = new ArrayList<>();
+ List<String> strings = new ArrayList<>();
+ for (IValueMeta valueMeta : prevFields.getValueMetaList()) {
+ switch (valueMeta.getType()) {
+ case IValueMeta.TYPE_INTEGER, IValueMeta.TYPE_NUMBER,
IValueMeta.TYPE_BIGNUMBER:
+ numeric.add(valueMeta.getName());
+ break;
+ case IValueMeta.TYPE_STRING:
+ strings.add(valueMeta.getName());
+ break;
+ default:
+ break;
+ }
+ }
+ numericFieldNames = numeric.toArray(new String[0]);
+ stringFieldNames = strings.toArray(new String[0]);
+ } catch (HopException e) {
+ new ErrorDialog(
+ shell,
+ BaseMessages.getString(PKG, "DelayDialog.UnableToGetFields.Title"),
+ BaseMessages.getString(PKG, "DelayDialog.UnableToGetFields.Message"),
+ e);
+ numericFieldNames = new String[0];
+ stringFieldNames = new String[0];
+ }
+ }
+
+ private void refreshTimeoutFieldItems() {
+ loadAvailableFields();
+ String previous = wTimeout.getText();
+ wTimeout.setItems(numericFieldNames);
+ if (!Utils.isEmpty(previous)) {
+ wTimeout.setText(previous);
+ wTimeout.getCComboWidget().setSelection(new Point(previous.length(),
previous.length()));
+ }
+ }
+
+ private void refreshScaleTimeFieldItems() {
+ loadAvailableFields();
+ String previous = wScaleTimeField.getText();
+ wScaleTimeField.setItems(stringFieldNames);
+ if (!Utils.isEmpty(previous)) {
+ wScaleTimeField.setText(previous);
+ wScaleTimeField
+ .getCComboWidget()
+ .setSelection(new Point(previous.length(), previous.length()));
+ }
+ }
}
diff --git
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayMeta.java
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayMeta.java
index 9236d3f2ec..1746712067 100644
---
a/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayMeta.java
+++
b/plugins/transforms/delay/src/main/java/org/apache/hop/pipeline/transforms/delay/DelayMeta.java
@@ -18,6 +18,8 @@
package org.apache.hop.pipeline.transforms.delay;
import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
import org.apache.hop.core.CheckResult;
import org.apache.hop.core.ICheckResult;
import org.apache.hop.core.annotations.Transform;
@@ -40,15 +42,32 @@ import org.apache.hop.pipeline.transform.TransformMeta;
categoryDescription =
"i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Utility",
keywords = "i18n::DelayMeta.keyword",
documentationUrl = "/pipeline/transforms/delay.html")
+@Getter
+@Setter
public class DelayMeta extends BaseTransformMeta<Delay, DelayData> {
private static final Class<?> PKG = DelayMeta.class;
@HopMetadataProperty(key = "timeout", injectionKeyDescription =
"Delay.Injection.Timeout")
private String timeout;
+ @HopMetadataProperty(
+ key = "timeout_field",
+ injectionKeyDescription = "Delay.Injection.TimeoutField")
+ private String timeoutField;
+
@HopMetadataProperty(key = "scaletime", injectionKeyDescription =
"Delay.Injection.Scaletime")
private String scaletime;
+ @HopMetadataProperty(
+ key = "scaletime_from_field",
+ injectionKeyDescription = "Delay.Injection.ScaleTimeFromField")
+ private boolean scaleTimeFromField;
+
+ @HopMetadataProperty(
+ key = "scaletime_field",
+ injectionKeyDescription = "Delay.Injection.ScaleTimeField")
+ private String scaleTimeField;
+
private static final String DEFAULT_SCALE_TIME = "seconds";
private static final String[] SCALE_TIME_CODE = {
@@ -64,20 +83,8 @@ public class DelayMeta extends BaseTransformMeta<Delay,
DelayData> {
return super.clone();
}
- public String getScaletime() {
- return scaletime;
- }
-
- public void setScaletime(String scaletime) {
- this.scaletime = scaletime;
- }
-
- public String getTimeout() {
- return timeout;
- }
-
- public void setTimeout(String timeout) {
- this.timeout = timeout;
+ public void setTimeoutField(String timeoutField) {
+ this.timeoutField = Utils.isEmpty(timeoutField) ? null : timeoutField;
}
public void setScaleTimeCode(int scaleTimeIndex) {
@@ -118,10 +125,17 @@ public class DelayMeta extends BaseTransformMeta<Delay,
DelayData> {
return retval;
}
+ public void setScaleTimeField(String scaleTimeField) {
+ this.scaleTimeField = Utils.isEmpty(scaleTimeField) ? null :
scaleTimeField;
+ }
+
@Override
public void setDefault() {
timeout = "1"; // default one second
scaletime = DEFAULT_SCALE_TIME; // defaults to "seconds"
+ timeoutField = null;
+ scaleTimeFromField = false;
+ scaleTimeField = null;
}
@Override
@@ -150,7 +164,7 @@ public class DelayMeta extends BaseTransformMeta<Delay,
DelayData> {
CheckResult cr;
String errorMessage = "";
- if (Utils.isEmpty(timeout)) {
+ if (Utils.isEmpty(timeout) && Utils.isEmpty(timeoutField)) {
errorMessage = BaseMessages.getString(PKG,
"DelayMeta.CheckResult.TimeOutMissing");
cr = new CheckResult(ICheckResult.TYPE_RESULT_ERROR, errorMessage,
transformMeta);
} else {
@@ -191,5 +205,33 @@ public class DelayMeta extends BaseTransformMeta<Delay,
DelayData> {
transformMeta);
}
remarks.add(cr);
+
+ if (prev != null && !prev.isEmpty()) {
+ if (!Utils.isEmpty(timeoutField) && prev.indexOfValue(timeoutField) < 0)
{
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG, "DelayMeta.CheckResult.TimeoutFieldNotFound",
timeoutField),
+ transformMeta));
+ }
+
+ if (scaleTimeFromField) {
+ if (Utils.isEmpty(scaleTimeField)) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(PKG,
"DelayMeta.CheckResult.ScaleTimeFieldMissing"),
+ transformMeta));
+ } else if (prev.indexOfValue(scaleTimeField) < 0) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG, "DelayMeta.CheckResult.ScaleTimeFieldNotFound",
scaleTimeField),
+ transformMeta));
+ }
+ }
+ }
}
}
diff --git
a/plugins/transforms/delay/src/main/resources/org/apache/hop/pipeline/transforms/delay/messages/messages_en_US.properties
b/plugins/transforms/delay/src/main/resources/org/apache/hop/pipeline/transforms/delay/messages/messages_en_US.properties
index f5f5af0c9a..e5ae4032b2 100644
---
a/plugins/transforms/delay/src/main/resources/org/apache/hop/pipeline/transforms/delay/messages/messages_en_US.properties
+++
b/plugins/transforms/delay/src/main/resources/org/apache/hop/pipeline/transforms/delay/messages/messages_en_US.properties
@@ -18,8 +18,24 @@
Delay.Description=Output each input row after a delay
Delay.Injection.Scaletime=Unit of waittime (milliseconds/seconds/minutes/hours)
Delay.Injection.Timeout=How long should we wait
+Delay.Injection.TimeoutField=Field containing timeout value
+Delay.Injection.ScaleTimeFromField=Use a field to determine the time scale
+Delay.Injection.ScaleTimeField=Field containing time scale value
Delay.Log.LineNumber=Linenr {0}
Delay.Log.TimeOut=Let''s wait now for {0} {1} ...
+Delay.Log.TimeOutWithScale=Waiting {0} {1} (total {2} ms)
+Delay.Log.UnknownScale=Unknown scale
+Delay.Log.DelayOverflow=Calculated delay is too large. Timeout: {0},
multiplier: {1}
+Delay.Log.TimeoutFieldNotFound=Timeout field ''{0}'' was not found in the
input stream.
+Delay.Log.TimeoutFieldType=Timeout field ''{0}'' must be numeric or string,
but was ''{1}''.
+Delay.Log.TimeoutFieldNull=Timeout field ''{0}'' contains an empty or null
value.
+Delay.Log.TimeoutFieldNegative=Timeout field ''{0}'' is negative ({1}).
+Delay.Log.TimeoutFieldNotNumeric=Timeout field ''{0}'' could not be converted
to a number.
+Delay.Log.ScaleTimeFieldMissing=Please specify the time scale field name when
enabling "Get time scale from field".
+Delay.Log.ScaleTimeFieldNotFound=Time scale field ''{0}'' was not found in the
input stream.
+Delay.Log.ScaleTimeFieldType=Time scale field ''{0}'' must be a string, but
was ''{1}''.
+Delay.Log.ScaleTimeFieldNull=Time scale field ''{0}'' contains an empty or
null value.
+Delay.Log.ScaleTimeFieldInvalid=Value ''{1}'' from time scale field ''{0}'' is
not recognised. Expected milliseconds/ms, seconds/s, minutes/m, or hours/h.
Delay.Name=Delay row
Delay.WaitTimeIsElapsed.Label=Wait time is reached.
DelayDialog.HrScaleTime.Label=Hours
@@ -29,11 +45,21 @@ DelayDialog.Shell.Title=Delay row
DelayDialog.SScaleTime.Label=Seconds
DelayDialog.Timeout.Label=Timeout
DelayDialog.Timeout.Tooltip=Units of time to wait
+DelayDialog.TimeoutField.Label=Timeout field
+DelayDialog.TimeoutField.Tooltip=Select a numeric field from the input stream
to determine the timeout.
DelayDialog.TransformName.Label=Transform name
+DelayDialog.ScaleTimeFromField.Label=Get time scale from field
+DelayDialog.ScaleTimeField.Label=Time scale field
+DelayDialog.ScaleTimeField.Tooltip=Select a field from the input stream that
contains values such as milliseconds, seconds, minutes or hours.
+DelayDialog.UnableToGetFields.Title=Unable to retrieve fields
+DelayDialog.UnableToGetFields.Message=Unable to retrieve fields from previous
transforms
DelayMeta.CheckResult.NoInputReceivedFromOtherTransforms=No input received
from other transforms\!
DelayMeta.CheckResult.NotReceivingFields=Not receiving any fields from
previous transforms\!
DelayMeta.CheckResult.TimeOutMissing=Timeout is missing!
DelayMeta.CheckResult.TimeOutOk=Timeout is specified!
DelayMeta.CheckResult.TransformRecevingData=Transform is connected to previous
one, receiving {0} fields
DelayMeta.CheckResult.TransformRecevingData2=Transform is receiving info from
other transforms.
+DelayMeta.CheckResult.TimeoutFieldNotFound=Timeout field ''{0}'' was not found
in the input stream.
+DelayMeta.CheckResult.ScaleTimeFieldMissing=Time scale field is required when
"Get time scale from field" is enabled.
+DelayMeta.CheckResult.ScaleTimeFieldNotFound=Time scale field ''{0}'' was not
found in the input stream.
DelayMeta.keyword=delay,wait,hold,slow
diff --git
a/plugins/transforms/delay/src/test/java/org/apache/hop/pipeline/transforms/delay/DelayMetaTest.java
b/plugins/transforms/delay/src/test/java/org/apache/hop/pipeline/transforms/delay/DelayMetaTest.java
index f05f212e3f..0a43c44447 100644
---
a/plugins/transforms/delay/src/test/java/org/apache/hop/pipeline/transforms/delay/DelayMetaTest.java
+++
b/plugins/transforms/delay/src/test/java/org/apache/hop/pipeline/transforms/delay/DelayMetaTest.java
@@ -33,18 +33,26 @@ class DelayMetaTest {
@Test
void testTransformMeta() throws HopException {
- List<String> attributes = Arrays.asList("timeout", "scaletime");
+ List<String> attributes =
+ Arrays.asList(
+ "timeout", "timeoutField", "scaletime", "scaleTimeFromField",
"scaleTimeField");
Map<String, String> getterMap = new HashMap<>();
getterMap.put("timeout", "getTimeout");
+ getterMap.put("timeoutField", "getTimeoutField");
getterMap.put("scaletime", "getScaletime");
+ getterMap.put("scaleTimeFromField", "isScaleTimeFromField");
+ getterMap.put("scaleTimeField", "getScaleTimeField");
Map<String, String> setterMap = new HashMap<>();
setterMap.put("timeout", "setTimeout");
+ setterMap.put("timeoutField", "setTimeoutField");
setterMap.put("scaletime", "setScaletime");
+ setterMap.put("scaleTimeFromField", "setScaleTimeFromField");
+ setterMap.put("scaleTimeField", "setScaleTimeField");
- LoadSaveTester loadSaveTester =
- new LoadSaveTester(DelayMeta.class, attributes, getterMap, setterMap);
+ LoadSaveTester<DelayMeta> loadSaveTester =
+ new LoadSaveTester<>(DelayMeta.class, attributes, getterMap,
setterMap);
loadSaveTester.testSerialization();
}
}
diff --git
a/plugins/transforms/delay/src/test/java/org/apache/hop/pipeline/transforms/delay/DelayTransformTest.java
b/plugins/transforms/delay/src/test/java/org/apache/hop/pipeline/transforms/delay/DelayTransformTest.java
new file mode 100644
index 0000000000..06b3da6c9c
--- /dev/null
+++
b/plugins/transforms/delay/src/test/java/org/apache/hop/pipeline/transforms/delay/DelayTransformTest.java
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.delay;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hop.core.BlockingRowSet;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class DelayTransformTest {
+
+ @RegisterExtension
+ static RestoreHopEngineEnvironmentExtension env = new
RestoreHopEngineEnvironmentExtension();
+
+ private TransformMockHelper<DelayMeta, DelayData> mockHelper;
+
+ @BeforeEach
+ void setUp() {
+ mockHelper = new TransformMockHelper<>("Delay", DelayMeta.class,
DelayData.class);
+ when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(mockHelper.iLogChannel);
+ when(mockHelper.pipeline.isRunning()).thenReturn(true);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockHelper.cleanUp();
+ }
+
+ @Test
+ void processRowTreatsNullTimeoutFieldAsZero() throws Exception {
+ DelayMeta meta = new DelayMeta();
+ meta.setTimeout("5");
+ meta.setTimeoutField("timeoutField");
+
+ DelayData data = new DelayData();
+ Delay delay = createTransform(meta, data);
+ assertTrue(delay.init());
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("timeoutField"));
+ Object[] rowData = new Object[] {null};
+
+ BlockingRowSet input = createInputRowSet(rowMeta, rowData);
+ BlockingRowSet output = createOutputRowSet(rowMeta);
+
+ delay.addRowSetToInputRowSets(input);
+ delay.addRowSetToOutputRowSets(output);
+ delay.setInputRowMeta(rowMeta);
+
+ assertTrue(delay.processRow());
+ Object[] result = output.getRowWait(1, TimeUnit.SECONDS);
+ assertArrayEquals(rowData, result);
+ assertEquals(0L, data.timeout);
+
+ assertFalse(delay.processRow());
+ }
+
+ @Test
+ void processRowUsesStringTimeoutField() throws Exception {
+ DelayMeta meta = new DelayMeta();
+ meta.setTimeoutField("timeoutField");
+ meta.setScaleTimeCode(0); // milliseconds
+
+ DelayData data = new DelayData();
+ Delay delay = createTransform(meta, data);
+ assertTrue(delay.init());
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("timeoutField"));
+ Object[] rowData = new Object[] {"2"};
+
+ BlockingRowSet input = createInputRowSet(rowMeta, rowData);
+ BlockingRowSet output = createOutputRowSet(rowMeta);
+
+ delay.addRowSetToInputRowSets(input);
+ delay.addRowSetToOutputRowSets(output);
+ delay.setInputRowMeta(rowMeta);
+
+ assertTrue(delay.processRow());
+ Object[] result = output.getRowWait(1, TimeUnit.SECONDS);
+ assertArrayEquals(rowData, result);
+ assertEquals(2L, data.timeout);
+
+ assertFalse(delay.processRow());
+ }
+
+ @Test
+ void processRowTreatsEmptyScaleFieldAsSeconds() throws Exception {
+ DelayMeta meta = new DelayMeta();
+ meta.setTimeoutField("timeout");
+ meta.setScaleTimeFromField(true);
+ meta.setScaleTimeField("scale");
+
+ DelayData data = new DelayData();
+ Delay delay = createTransform(meta, data);
+ assertTrue(delay.init());
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("timeout"));
+ rowMeta.addValueMeta(new ValueMetaString("scale"));
+ Object[] rowData = new Object[] {null, ""};
+
+ BlockingRowSet input = createInputRowSet(rowMeta, rowData);
+ BlockingRowSet output = createOutputRowSet(rowMeta);
+
+ delay.addRowSetToInputRowSets(input);
+ delay.addRowSetToOutputRowSets(output);
+ delay.setInputRowMeta(rowMeta);
+
+ assertTrue(delay.processRow());
+ Object[] result = output.getRowWait(1, TimeUnit.SECONDS);
+ assertArrayEquals(rowData, result);
+ assertEquals(0L, data.timeout);
+ assertEquals(1000L, data.multiple); // seconds
+
+ assertFalse(delay.processRow());
+ }
+
+ @Test
+ void processRowWithUnknownScaleThrows() throws Exception {
+ DelayMeta meta = new DelayMeta();
+ meta.setTimeoutField("timeout");
+ meta.setScaleTimeFromField(true);
+ meta.setScaleTimeField("scale");
+
+ DelayData data = new DelayData();
+ Delay delay = createTransform(meta, data);
+ assertTrue(delay.init());
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("timeout"));
+ rowMeta.addValueMeta(new ValueMetaString("scale"));
+ Object[] rowData = new Object[] {1L, "weeks"};
+
+ BlockingRowSet input = createInputRowSet(rowMeta, rowData);
+ delay.addRowSetToInputRowSets(input);
+ delay.setInputRowMeta(rowMeta);
+
+ HopException exception = assertThrows(HopException.class,
delay::processRow);
+ String expectedMessage =
+ BaseMessages.getString(
+ DelayMeta.class, "Delay.Log.ScaleTimeFieldInvalid",
meta.getScaleTimeField(), "weeks");
+ assertEquals(expectedMessage.trim(), exception.getMessage().trim());
+ }
+
+ @Test
+ void processRowWithNegativeTimeoutThrows() throws Exception {
+ DelayMeta meta = new DelayMeta();
+ meta.setTimeoutField("timeout");
+
+ DelayData data = new DelayData();
+ Delay delay = createTransform(meta, data);
+ assertTrue(delay.init());
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("timeout"));
+ Object[] rowData = new Object[] {-1L};
+
+ BlockingRowSet input = createInputRowSet(rowMeta, rowData);
+ delay.addRowSetToInputRowSets(input);
+ delay.setInputRowMeta(rowMeta);
+
+ HopException exception = assertThrows(HopException.class,
delay::processRow);
+ String expectedMessage =
+ BaseMessages.getString(
+ DelayMeta.class, "Delay.Log.TimeoutFieldNegative",
meta.getTimeoutField(), -1L);
+ assertEquals(expectedMessage.trim(), exception.getMessage().trim());
+ }
+
+ private Delay createTransform(DelayMeta meta, DelayData data) {
+ when(mockHelper.transformMeta.getTransform()).thenReturn(meta);
+ return new Delay(
+ mockHelper.transformMeta, meta, data, 0, mockHelper.pipelineMeta,
mockHelper.pipeline);
+ }
+
+ private BlockingRowSet createInputRowSet(IRowMeta rowMeta, Object[]... rows)
{
+ BlockingRowSet rowSet = new BlockingRowSet(10);
+ rowSet.setThreadNameFromToCopy("previous", 0, "Delay", 0);
+ rowSet.setRowMeta(rowMeta);
+ for (Object[] row : rows) {
+ rowSet.putRow(rowMeta, row);
+ }
+ rowSet.setDone();
+ return rowSet;
+ }
+
+ private BlockingRowSet createOutputRowSet(IRowMeta rowMeta) {
+ BlockingRowSet rowSet = new BlockingRowSet(10);
+ rowSet.setThreadNameFromToCopy("Delay", 0, "next", 0);
+ rowSet.setRowMeta(rowMeta);
+ return rowSet;
+ }
+}
