From a9b3982b536432c3b268e03cf731f7e8dabecb19 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tobias=20B=C3=BClte?= Date: Tue, 24 Sep 2024 13:20:45 +0200 Subject: [PATCH 1/8] Add draft for object sleeper #495 --- .../flowcontrol/ObjectSleeper.java | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java diff --git a/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java new file mode 100644 index 000000000..02824b669 --- /dev/null +++ b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java @@ -0,0 +1,72 @@ +/* +* Copyright 2024 hbz +* +* Licensed 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.metafacture.strings; + +import org.metafacture.framework.FluxCommand; +import org.metafacture.framework.ObjectReceiver; +import org.metafacture.framework.annotations.Description; +import org.metafacture.framework.annotations.In; +import org.metafacture.framework.annotations.Out; +import org.metafacture.framework.helpers.DefaultObjectPipe; + +/** + * Lets the process between objects sleep for a specific ms. +* +* @author Tobias Bülte +*/ +@Description("Lets the process between objects sleep for a specific ms.") +@In(Object.class) +@Out(Object.class) +@FluxCommand("object-sleep") +public final class ObjectSleeper extends DefaultObjectPipe> { + + public static final long DEFAULT_SLEEP_TIME = 1000; + + private long sleepTime = DEFAULT_SLEEP_TIME; + + /** + * Creates an instance of {@link ObjectSleeper}. + */ + public ObjectSleeper() { + } + + + /** + * Sets the time in ms for the sleep phase. + * + * @param sleepTime the time to sleep + */ + public void setSleepTime(final int sleepTime) { + this.sleepTime = sleepTime; + } + + /** + * Gets the time in ms for the sleep phase. + * + * @return the time to sleep + */ + public long getSleepTime() { + return sleepTime; + } + + @Override + public void process(final T obj) { + Thread.sleep(sleepTime); + getReceiver().process(obj); + } + +} From 489a74d1c450c53d1232331541d8e5d6478c3fcb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tobias=20B=C3=BClte?= Date: Thu, 17 Oct 2024 12:15:25 +0200 Subject: [PATCH 2/8] Adjust ObjectSleeper.java #495 --- .../flowcontrol/ObjectSleeper.java | 28 ++++++++++++------- .../main/resources/flux-commands.properties | 1 + 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java index 02824b669..4526684c1 100644 --- a/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java +++ b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java @@ -14,25 +14,28 @@ * limitations under the License. */ -package org.metafacture.strings; +package org.metafacture.flowcontrol; import org.metafacture.framework.FluxCommand; +import org.metafacture.framework.MetafactureException; import org.metafacture.framework.ObjectReceiver; import org.metafacture.framework.annotations.Description; import org.metafacture.framework.annotations.In; import org.metafacture.framework.annotations.Out; import org.metafacture.framework.helpers.DefaultObjectPipe; + /** * Lets the process between objects sleep for a specific ms. -* -* @author Tobias Bülte -*/ + * + * @param object type + * @author Tobias Bülte + */ @Description("Lets the process between objects sleep for a specific ms.") @In(Object.class) @Out(Object.class) -@FluxCommand("object-sleep") -public final class ObjectSleeper extends DefaultObjectPipe> { +@FluxCommand("sleep") +public final class ObjectSleeper extends DefaultObjectPipe> { public static final long DEFAULT_SLEEP_TIME = 1000; @@ -44,8 +47,7 @@ public final class ObjectSleeper extends DefaultObjectPipe> public ObjectSleeper() { } - - /** + /** * Sets the time in ms for the sleep phase. * * @param sleepTime the time to sleep @@ -54,7 +56,7 @@ public void setSleepTime(final int sleepTime) { this.sleepTime = sleepTime; } - /** + /** * Gets the time in ms for the sleep phase. * * @return the time to sleep @@ -65,7 +67,13 @@ public long getSleepTime() { @Override public void process(final T obj) { - Thread.sleep(sleepTime); + try { + Thread.sleep(sleepTime); + } + catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MetafactureException(e.getMessage(), e); + } getReceiver().process(obj); } diff --git a/metafacture-flowcontrol/src/main/resources/flux-commands.properties b/metafacture-flowcontrol/src/main/resources/flux-commands.properties index beefcf19a..ec36f0791 100644 --- a/metafacture-flowcontrol/src/main/resources/flux-commands.properties +++ b/metafacture-flowcontrol/src/main/resources/flux-commands.properties @@ -21,3 +21,4 @@ reset-object-batch org.metafacture.flowcontrol.ObjectBatchResetter defer-stream org.metafacture.flowcontrol.StreamDeferrer catch-stream-exception org.metafacture.flowcontrol.StreamExceptionCatcher thread-object-tee org.metafacture.flowcontrol.ObjectThreader +sleep org.metafacture.flowcontrol.ObjectSleeper From f069a04098547d89284ad851eecaa70af56feb6c Mon Sep 17 00:00:00 2001 From: Jens Wille Date: Wed, 6 Nov 2024 13:56:18 +0100 Subject: [PATCH 3/8] Extract `ObjectSleeper.sleep()` method to make it reusable (e.g. in metafacture-fix). (#559) --- .../org/metafacture/flowcontrol/ObjectSleeper.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java index 4526684c1..304d7f379 100644 --- a/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java +++ b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java @@ -65,8 +65,10 @@ public long getSleepTime() { return sleepTime; } - @Override - public void process(final T obj) { + /** + * Sleeps for the specified amount of time. + */ + public void sleep() { try { Thread.sleep(sleepTime); } @@ -74,6 +76,11 @@ public void process(final T obj) { Thread.currentThread().interrupt(); throw new MetafactureException(e.getMessage(), e); } + } + + @Override + public void process(final T obj) { + sleep(); getReceiver().process(obj); } From 18b6ae0cc0446c4cbe0dc6ed6127b21d69914e26 Mon Sep 17 00:00:00 2001 From: Jens Wille Date: Wed, 6 Nov 2024 14:52:55 +0100 Subject: [PATCH 4/8] Add support for time units in `ObjectSleeper` (similar to Catmandu). (#559) See also https://metacpan.org/pod/Catmandu::Fix::sleep (defaults to `SECONDS`, doesn't support `NANOSECONDS` and `DAYS`). --- .../flowcontrol/ObjectSleeper.java | 61 +++++++++++++++++-- 1 file changed, 56 insertions(+), 5 deletions(-) diff --git a/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java index 304d7f379..e42a355f3 100644 --- a/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java +++ b/metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectSleeper.java @@ -24,21 +24,26 @@ import org.metafacture.framework.annotations.Out; import org.metafacture.framework.helpers.DefaultObjectPipe; +import java.util.concurrent.TimeUnit; /** - * Lets the process between objects sleep for a specific ms. + * Lets the process sleep for a specific amount of time between objects. * * @param object type * @author Tobias Bülte */ -@Description("Lets the process between objects sleep for a specific ms.") +@Description("Lets the process sleep for a specific amount of time between objects.") @In(Object.class) @Out(Object.class) @FluxCommand("sleep") public final class ObjectSleeper extends DefaultObjectPipe> { + public static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS; public static final long DEFAULT_SLEEP_TIME = 1000; + private static final String TIME_UNIT_SUFFIX = "S"; + + private TimeUnit timeUnit = DEFAULT_TIME_UNIT; private long sleepTime = DEFAULT_SLEEP_TIME; /** @@ -48,16 +53,29 @@ public ObjectSleeper() { } /** - * Sets the time in ms for the sleep phase. + * Sets the amount of time for the sleep phase (measured in {@link + * #setTimeUnit time unit}). * * @param sleepTime the time to sleep */ public void setSleepTime(final int sleepTime) { + // NOTE: ConfigurableClass.convertValue() doesn't support long. + setSleepTime((long) sleepTime); + } + + /** + * Sets the amount of time for the sleep phase (measured in {@link + * #setTimeUnit time unit}). + * + * @param sleepTime the time to sleep + */ + public void setSleepTime(final long sleepTime) { this.sleepTime = sleepTime; } /** - * Gets the time in ms for the sleep phase. + * Gets the amount of time for the sleep phase (measured in {@link + * #setTimeUnit time unit}). * * @return the time to sleep */ @@ -65,12 +83,45 @@ public long getSleepTime() { return sleepTime; } + /** + * Sets the time unit for the sleep phase. See {@link TimeUnit available + * time units}, case-insensitive, trailing "s" optional. + * + * @param timeUnit the time unit + */ + public void setTimeUnit(final String timeUnit) { + // NOTE: Adds NANOSECONDS and DAYS over Catmandu's supported time units. + + final String timeUnitName = timeUnit.toUpperCase(); + final String timeUnitSuffix = timeUnitName.endsWith(TIME_UNIT_SUFFIX) ? "" : TIME_UNIT_SUFFIX; + + setTimeUnit(TimeUnit.valueOf(timeUnitName + timeUnitSuffix)); + } + + /** + * Sets the time unit for the sleep phase. + * + * @param timeUnit the time unit + */ + public void setTimeUnit(final TimeUnit timeUnit) { + this.timeUnit = timeUnit; + } + + /** + * Gets the time unit for the sleep phase. + * + * @return the time unit + */ + public TimeUnit getTimeUnit() { + return timeUnit; + } + /** * Sleeps for the specified amount of time. */ public void sleep() { try { - Thread.sleep(sleepTime); + timeUnit.sleep(sleepTime); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); From 4f11c696bec1de7760dcfdaf354bb3cfe1ceeb11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tobias=20B=C3=BClte?= Date: Fri, 15 Nov 2024 09:09:26 +0100 Subject: [PATCH 5/8] Draft test Add a flux command for waiting a specific time #495 --- .../flowcontrol/ObjectSleeperTest.java | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java diff --git a/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java new file mode 100644 index 000000000..1b6aea191 --- /dev/null +++ b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java @@ -0,0 +1,73 @@ +/* + * Copyright 2016 Christoph Böhme + * + * Licensed 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.metafacture.flowcontrol; + + import static org.mockito.ArgumentMatchers.anyString; + import static org.mockito.Mockito.doThrow; + + import org.junit.Before; + import org.junit.Test; + import org.metafacture.framework.MetafactureException; + import org.metafacture.framework.ObjectReceiver; + import org.mockito.Mock; + import org.mockito.MockitoAnnotations; + import java.time.Duration; + import java.time.Instant; + + /** + * Tests for class {@link ObjectSleeper}. + * + * @author Tobias Bülte + * + */ + public final class ObjectExceptionSleeperTest { + + @Mock + private ObjectReceiver sleepTimer; + + @Before + public void setup() { + MockitoAnnotations.initMocks(this); + } + + @After + public void cleanup() { + bulk.closeStream(); + } + + + @Test + public void shouldTestIfClockedTimeExceedsDuration() { + long sleepTime = 10; + + objectSleeper = new ObjectSleeper(); + objectSleeper.setSleepTime(sleepTime); + Instant start = Instant.now(); + sleepTimer.objectSleeper(); + Instant end = Instant.now(); + + Duration timeElapsed = Duration.between(start, end); + + if (timeElampse > sleepTime) { + exception.expect(MetafactureException.class); + exception.expectMessage("Process did not sleep enough."); + } + + } + + + } From b86dce35fc59bbe74a7e1eafbece88e60c761f5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tobias=20B=C3=BClte?= Date: Mon, 25 Nov 2024 14:40:05 +0100 Subject: [PATCH 6/8] Incorporate some suggestions by @fsteeg --- .../flowcontrol/ObjectSleeperTest.java | 69 +++++++++---------- 1 file changed, 31 insertions(+), 38 deletions(-) diff --git a/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java index 1b6aea191..5dc35f3f5 100644 --- a/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java +++ b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java @@ -14,27 +14,28 @@ * limitations under the License. */ - package org.metafacture.flowcontrol; +package org.metafacture.flowcontrol; - import static org.mockito.ArgumentMatchers.anyString; - import static org.mockito.Mockito.doThrow; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; - import org.junit.Before; - import org.junit.Test; - import org.metafacture.framework.MetafactureException; - import org.metafacture.framework.ObjectReceiver; - import org.mockito.Mock; - import org.mockito.MockitoAnnotations; - import java.time.Duration; - import java.time.Instant; +import org.junit.Before; +import org.junit.Test; +import org.metafacture.framework.MetafactureException; +import org.metafacture.framework.ObjectReceiver; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import java.time.Duration; +import java.time.Instant; - /** - * Tests for class {@link ObjectSleeper}. - * - * @author Tobias Bülte - * - */ - public final class ObjectExceptionSleeperTest { +/** + * Tests for class {@link ObjectSleeper}. +* +* @author Tobias Bülte +* +*/ +public final class ObjectSleeperTest { @Mock private ObjectReceiver sleepTimer; @@ -44,30 +45,22 @@ public void setup() { MockitoAnnotations.initMocks(this); } - @After - public void cleanup() { - bulk.closeStream(); - } - + @Test + public void shouldTestIfClockedTimeExceedsDuration() { + long sleepTime = 10; - @Test - public void shouldTestIfClockedTimeExceedsDuration() { - long sleepTime = 10; + ObjectSleeper objectSleeper = new ObjectSleeper(); + objectSleeper.setSleepTime(sleepTime); + Instant start = Instant.now(); + sleepTimer.objectSleeper; + Instant end = Instant.now(); - objectSleeper = new ObjectSleeper(); - objectSleeper.setSleepTime(sleepTime); - Instant start = Instant.now(); - sleepTimer.objectSleeper(); - Instant end = Instant.now(); - Duration timeElapsed = Duration.between(start, end); + Duration timeElapsed = Duration.between(start, end); - if (timeElampse > sleepTime) { - exception.expect(MetafactureException.class); - exception.expectMessage("Process did not sleep enough."); - } + assertTrue(timeElapsed.toSeconds() >= sleepTime); - } + } - } +} From 03bc0099d7172cb539027c0bd62b6e27aa8c2dd2 Mon Sep 17 00:00:00 2001 From: Fabian Steeg Date: Wed, 27 Nov 2024 16:39:48 +0100 Subject: [PATCH 7/8] Tweak test to fix compiler error and warnings, wait 100 ms (#495) --- .../flowcontrol/ObjectSleeperTest.java | 21 +++++++++---------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java index 5dc35f3f5..ba3ae0982 100644 --- a/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java +++ b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 Christoph Böhme + * Copyright 2024 Tobias Bülte, hbz * * Licensed under the Apache License, Version 2.0 the "License"; * you may not use this file except in compliance with the License. @@ -17,15 +17,14 @@ package org.metafacture.flowcontrol; import static org.junit.Assert.assertTrue; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.doThrow; + +import org.metafacture.framework.ObjectReceiver; import org.junit.Before; import org.junit.Test; -import org.metafacture.framework.MetafactureException; -import org.metafacture.framework.ObjectReceiver; import org.mockito.Mock; import org.mockito.MockitoAnnotations; + import java.time.Duration; import java.time.Instant; @@ -38,7 +37,7 @@ public final class ObjectSleeperTest { @Mock - private ObjectReceiver sleepTimer; + private ObjectReceiver receiver; @Before public void setup() { @@ -47,18 +46,18 @@ public void setup() { @Test public void shouldTestIfClockedTimeExceedsDuration() { - long sleepTime = 10; + long sleepTime = 100; - ObjectSleeper objectSleeper = new ObjectSleeper(); + ObjectSleeper objectSleeper = new ObjectSleeper<>(); + objectSleeper.setReceiver(receiver); objectSleeper.setSleepTime(sleepTime); Instant start = Instant.now(); - sleepTimer.objectSleeper; + objectSleeper.process(null); Instant end = Instant.now(); - Duration timeElapsed = Duration.between(start, end); - assertTrue(timeElapsed.toSeconds() >= sleepTime); + assertTrue(timeElapsed.toMillis() >= sleepTime); } From 74bc15d57b971fcafade30345d07ef13355e8ef9 Mon Sep 17 00:00:00 2001 From: Jens Wille Date: Thu, 28 Nov 2024 15:04:10 +0100 Subject: [PATCH 8/8] Extend and clean up `ObjectSleeper` test. (#559) --- .../flowcontrol/ObjectSleeperTest.java | 57 +++++++++++++------ 1 file changed, 39 insertions(+), 18 deletions(-) diff --git a/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java index ba3ae0982..ab87ee68e 100644 --- a/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java +++ b/metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectSleeperTest.java @@ -16,26 +16,28 @@ package org.metafacture.flowcontrol; -import static org.junit.Assert.assertTrue; - import org.metafacture.framework.ObjectReceiver; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.mockito.Mock; import org.mockito.MockitoAnnotations; -import java.time.Duration; -import java.time.Instant; +import java.util.function.Consumer; /** * Tests for class {@link ObjectSleeper}. -* -* @author Tobias Bülte -* -*/ + * + * @author Tobias Bülte + */ public final class ObjectSleeperTest { + private static final int PROCESS_OVERHEAD_MILLISECONDS = 100; + + private static final int MILLISECONDS_PER_SECOND = 1_000; + private static final int NANOSECONDS_PER_MILLISECOND = 1_000_000; + @Mock private ObjectReceiver receiver; @@ -46,20 +48,39 @@ public void setup() { @Test public void shouldTestIfClockedTimeExceedsDuration() { - long sleepTime = 100; + final int sleepTime = 1234; + assertSleep(sleepTime, s -> s.setSleepTime(sleepTime)); + } - ObjectSleeper objectSleeper = new ObjectSleeper<>(); - objectSleeper.setReceiver(receiver); - objectSleeper.setSleepTime(sleepTime); - Instant start = Instant.now(); - objectSleeper.process(null); - Instant end = Instant.now(); + @Test + public void shouldTestIfClockedTimeExceedsDurationInMilliseconds() { + final int sleepTime = 567; + assertSleep(sleepTime, s -> { + s.setSleepTime(sleepTime); + s.setTimeUnit("MILLISECONDS"); + }); + } + + @Test + public void shouldTestIfClockedTimeExceedsDurationInSeconds() { + final int sleepTime = 1; + assertSleep(sleepTime * MILLISECONDS_PER_SECOND, s -> { + s.setSleepTime(sleepTime); + s.setTimeUnit("SECOND"); + }); + } - Duration timeElapsed = Duration.between(start, end); + private void assertSleep(final long expectedMillis, final Consumer consumer) { + final ObjectSleeper objectSleeper = new ObjectSleeper<>(); + objectSleeper.setReceiver(receiver); + consumer.accept(objectSleeper); - assertTrue(timeElapsed.toMillis() >= sleepTime); + final long startTime = System.nanoTime(); + objectSleeper.process(null); + final long actualMillis = (System.nanoTime() - startTime) / NANOSECONDS_PER_MILLISECOND; + Assert.assertTrue("sleep time too short: " + actualMillis, actualMillis >= expectedMillis); + Assert.assertTrue("sleep time too long: " + actualMillis, actualMillis < expectedMillis + PROCESS_OVERHEAD_MILLISECONDS); } - }