Skip to content

Commit 564bc2d

Browse files
author
jodzga
committed
Added Action functional interface.
Added runSideEffect to Context. Added Task.shareable(). Fixed signature of delayedFailure in BaseEngineTest.
1 parent 51f3110 commit 564bc2d

15 files changed

Lines changed: 137 additions & 46 deletions

File tree

‎example/com/linkedin/parseq/example/composite/TimeBoundSearchExample.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import com.linkedin.parseq.example.common.AbstractExample;
3131
import com.linkedin.parseq.example.common.MockService;
3232
import com.linkedin.parseq.example.common.SimpleMockRequest;
33+
import com.linkedin.parseq.function.Action;
3334
import com.linkedin.parseq.promise.Promise;
3435
import com.linkedin.parseq.promise.Promises;
3536
import com.linkedin.parseq.promise.SettablePromise;
@@ -135,7 +136,7 @@ public Promise<List<Integer>> run(final Context ctx)
135136

136137
private Task<?> checkDone()
137138
{
138-
return Task.action("checkDone", new Runnable()
139+
return Task.action("checkDone", new Action()
139140
{
140141
@Override
141142
public void run()
@@ -151,7 +152,7 @@ public void run()
151152

152153
private Task<?> addResponse(final Promise<Integer> response)
153154
{
154-
return Task.action("addResponse", new Runnable()
155+
return Task.action("addResponse", new Action()
155156
{
156157
@Override
157158
public void run()

‎example/com/linkedin/parseq/example/composite/classifier/ClassifierPlanFactory.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import com.linkedin.parseq.example.composite.classifier.client.Request;
3131
import com.linkedin.parseq.example.composite.classifier.client.impl.GetNetworkRequest;
3232
import com.linkedin.parseq.example.composite.classifier.client.impl.TruthMapRequest;
33+
import com.linkedin.parseq.function.Action;
3334
import com.linkedin.parseq.promise.Promise;
3435
import com.linkedin.parseq.promise.Promises;
3536
import com.linkedin.parseq.promise.SettablePromise;
@@ -111,7 +112,7 @@ public Promise<Map<Long, Classification>> run(final Context ctx)
111112

112113
private Task<?> classifyTask(final Classifier classifier)
113114
{
114-
return Task.action(classifier.getClass().getSimpleName(), new Runnable()
115+
return Task.action(classifier.getClass().getSimpleName(), new Action()
115116
{
116117
@Override
117118
public void run()
@@ -137,7 +138,7 @@ private Task<?> truthMapClassifyTask(final String name,
137138
final Classification classification,
138139
final Promise<Map<Long, Boolean>> result)
139140
{
140-
return Task.action(name + "Classifier", new Runnable()
141+
return Task.action(name + "Classifier", new Action()
141142
{
142143
@Override
143144
public void run()
@@ -161,7 +162,7 @@ protected Promise<? extends T> run(final Context context) throws Exception
161162

162163
private Task<?> connectedClassifyTask(final Task<Network> network)
163164
{
164-
return Task.action("ConnectedClassifier", new Runnable()
165+
return Task.action("ConnectedClassifier", new Action()
165166
{
166167
@Override
167168
public void run()
@@ -173,7 +174,7 @@ public void run()
173174

174175
private Task<?> networkClassifyTask(final Task<Network> network)
175176
{
176-
return Task.action("NetworkClassifier", new Runnable()
177+
return Task.action("NetworkClassifier", new Action()
177178
{
178179
@Override
179180
public void run()

‎src-test/com/linkedin/parseq/BaseEngineTest.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ public class BaseEngineTest
5151
private Engine _engine;
5252
private ListLoggerFactory _loggerFactory;
5353

54+
@SuppressWarnings("deprecation")
5455
@BeforeMethod
5556
public void setUp() throws Exception
5657
{
@@ -169,7 +170,7 @@ protected void resetLoggers()
169170
*/
170171
protected <T> Task<T> delayedValue(T value, long time, TimeUnit timeUnit)
171172
{
172-
return Task.async("delayed " + time + " " + TimeUnitHelper.toString(timeUnit), () -> {
173+
return Task.async(value.toString() + " delayed " + time + " " + TimeUnitHelper.toString(timeUnit), () -> {
173174
final SettablePromise<T> promise = Promises.settable();
174175
_scheduler.schedule(() -> promise.done(value), time, timeUnit);
175176
return promise;
@@ -180,9 +181,9 @@ protected <T> Task<T> delayedValue(T value, long time, TimeUnit timeUnit)
180181
* Returns task which fails with given error after specified period
181182
* of time. Timer starts counting the moment this method is invoked.
182183
*/
183-
protected <T> Task<T> delayedFailure(T value, Throwable error, long time, TimeUnit timeUnit)
184+
protected <T> Task<T> delayedFailure(Throwable error, long time, TimeUnit timeUnit)
184185
{
185-
return Task.async("delayedFailure", () -> {
186+
return Task.async(error.toString() + " delayed " + time + " " + TimeUnitHelper.toString(timeUnit), () -> {
186187
final SettablePromise<T> promise = Promises.settable();
187188
_scheduler.schedule(() -> promise.fail(error), time, timeUnit);
188189
return promise;

‎src/com/linkedin/parseq/BaseTask.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -617,6 +617,15 @@ public Long getTaskId() {
617617
public TaskLogger getTaskLogger() {
618618
return _context.getTaskLogger();
619619
}
620+
621+
@Override
622+
public void runSideEffect(Task<?>... tasks) {
623+
_context.runSideEffect(tasks);
624+
for(Task<?> task : tasks)
625+
{
626+
getTraceBuilder().addRelationship(Relationship.POTENTIAL_PARENT_OF, getShallowTraceBuilder(), task.getShallowTraceBuilder());
627+
}
628+
}
620629
}
621630

622631
@Override

‎src/com/linkedin/parseq/Context.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,8 @@ public interface Context
5353
* @param tasks the tasks to run
5454
*/
5555
void run(Task<?>... tasks);
56+
57+
void runSideEffect(Task<?>... tasks);
5658

5759
/**
5860
* TODO

‎src/com/linkedin/parseq/Task.java‎

Lines changed: 24 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.slf4j.Logger;
2929
import org.slf4j.LoggerFactory;
3030

31+
import com.linkedin.parseq.function.Action;
3132
import com.linkedin.parseq.function.Consumer1;
3233
import com.linkedin.parseq.function.Failure;
3334
import com.linkedin.parseq.function.Function1;
@@ -234,8 +235,8 @@ default <R> Task<R> flatMap(final Function1<? super T, Task<R>> func) {
234235
* Creates a new task that will run another task as a side effect once the primary task
235236
* completes successfully. The properties of side effect task are:
236237
* <ul>
237-
* <li>The side effect task will not be run if the primary task fails or
238-
* is canceled.</li>
238+
* <li>The side effect task will not be run if the primary task has not run e.g. due to
239+
* failure or cancellation.</li>
239240
* <li>The side effect does not affect returned task. It means that
240241
* failure of side effect task is not propagated to returned task.</li>
241242
* <li>The returned task is marked done once this task completes, even if
@@ -252,7 +253,8 @@ default <R> Task<R> flatMap(final Function1<? super T, Task<R>> func) {
252253
* Task{@code <String>} userName = id.flatMap("fetch", u -> fetch(u))
253254
* .withSideEffect("update memcache", u -> updateMemcache(u));
254255
* </code></pre>
255-
* @param desc description of a function, it will show up in a trace
256+
*
257+
* @param desc description of a side effect, it will show up in a trace
256258
* @param func function to be applied on result of successful completion of this task
257259
* to get side effect task
258260
* @return a new task that will run side effect task specified by given function upon succesful
@@ -261,7 +263,7 @@ default <R> Task<R> flatMap(final Function1<? super T, Task<R>> func) {
261263
default Task<T> withSideEffect(final String desc, final Function1<? super T, Task<?>> func) {
262264
ArgumentUtil.requireNotNull(func, "function");
263265
final Task<T> that = this;
264-
return async(desc, context -> {
266+
return async("withSideEffect", context -> {
265267
final Task<?> sideEffectWrapper = async(desc, ctx -> {
266268
Task<?> sideEffect = func.apply(that.get());
267269
ctx.run(sideEffect);
@@ -274,13 +276,23 @@ default Task<T> withSideEffect(final String desc, final Function1<? super T, Tas
274276
}
275277

276278
/**
277-
* Equivalent to {@code withSideEffect("withSideEffect", func)}.
279+
* Equivalent to {@code withSideEffect("sideEffect", func)}.
278280
* @see #withSideEffect(String, Function)
279281
*/
280282
default Task<T> withSideEffect(final Function1<? super T, Task<?>> func) {
281-
return withSideEffect("withSideEffect", func);
283+
return withSideEffect("sideEffect", func);
282284
}
283285

286+
default Task<T> shareable() {
287+
final Task<T> that = this;
288+
return async("shareable", context -> {
289+
final SettablePromise<T> result = Promises.settable();
290+
context.runSideEffect(that);
291+
Promises.propagateResult(that, result);
292+
return result;
293+
}, true);
294+
}
295+
284296
/**
285297
* Creates a new task which applies a consumer to the result of this task
286298
* and completes with a result of this task. It is used
@@ -737,7 +749,7 @@ public static <R> Task<R> flatten(final Task<Task<R>> task) {
737749
* Task{@code <Void>} task = Task.action("greeting", () -> System.out.println("Hello"));
738750
* </code></pre>
739751
*
740-
* Returned task will fail if {@code Runnable} passed in as a parameter throws
752+
* Returned task will fail if {@code Action} passed in as a parameter throws
741753
* an exception.
742754
* <pre><code>
743755
* // this task will fail with java.lang.ArithmeticException
@@ -748,7 +760,7 @@ public static <R> Task<R> flatten(final Task<Task<R>> task) {
748760
* @param action the action that will be executed when the task is run
749761
* @return the new task that will execute the action
750762
*/
751-
public static Task<Void> action(final String desc, final Runnable action)
763+
public static Task<Void> action(final String desc, final Action action)
752764
{
753765
ArgumentUtil.requireNotNull(action, "action");
754766
return async(desc, () -> {
@@ -758,12 +770,12 @@ public static Task<Void> action(final String desc, final Runnable action)
758770
}
759771

760772
/**
761-
* Equivalent to {@code action("action", runnable)}.
762-
* @see #action(String, Runnable)
773+
* Equivalent to {@code action("action", action)}.
774+
* @see #action(String, Action)
763775
*/
764-
public static Task<Void> action(final Runnable runnable)
776+
public static Task<Void> action(final Action action)
765777
{
766-
return action("action", runnable);
778+
return action("action", action);
767779
}
768780

769781
/**

‎src/com/linkedin/parseq/Tasks.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ private Tasks() {}
4848
*/
4949
@Deprecated public static Task<Void> action(final String name, final Runnable runnable)
5050
{
51-
return Task.action(name, runnable);
51+
return Task.action(name, runnable::run);
5252
}
5353

5454
/**

‎src/com/linkedin/parseq/TimeoutWithErrorTask.java‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,15 +16,16 @@
1616

1717
package com.linkedin.parseq;
1818

19+
import java.util.concurrent.TimeUnit;
20+
import java.util.concurrent.TimeoutException;
21+
import java.util.concurrent.atomic.AtomicBoolean;
22+
23+
import com.linkedin.parseq.function.Action;
1924
import com.linkedin.parseq.promise.Promise;
2025
import com.linkedin.parseq.promise.PromiseListener;
2126
import com.linkedin.parseq.promise.Promises;
2227
import com.linkedin.parseq.promise.SettablePromise;
2328

24-
import java.util.concurrent.TimeUnit;
25-
import java.util.concurrent.TimeoutException;
26-
import java.util.concurrent.atomic.AtomicBoolean;
27-
2829
/**
2930
* A {@link Task} that will attempt to run the given task within the given
3031
* timeout. If the timeout expires then this task will fail with a
@@ -58,7 +59,7 @@ protected Promise<? extends T> run(final Context context) throws Exception
5859
final SettablePromise<T> result = Promises.settable();
5960
final AtomicBoolean committed = new AtomicBoolean();
6061

61-
final Task<?> timeoutTask = Task.action("timeoutTimer", new Runnable()
62+
final Task<?> timeoutTask = Task.action("timeoutTimer", new Action()
6263
{
6364
@Override
6465
public void run()

‎src/com/linkedin/parseq/function/Action.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
package com.linkedin.parseq.function;
2+
3+
@FunctionalInterface
4+
public interface Action {
5+
public void run() throws Exception;
6+
}

‎src/com/linkedin/parseq/internal/ContextImpl.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,6 @@ public void runTask()
8686
@Override
8787
public void onResolved(Promise<Object> resolvedPromise)
8888
{
89-
//TODO is this iteration safe? can _cancellables be modified at the same time?
9089
for (Iterator<Cancellable> it = _cancellables.iterator(); it.hasNext(); )
9190
{
9291
final Cancellable cancellable = it.next();
@@ -147,6 +146,16 @@ public void run(final Task<?>... tasks)
147146
}
148147
}
149148

149+
@Override
150+
public void runSideEffect(final Task<?>... tasks)
151+
{
152+
checkInTask();
153+
for (final Task<?> task : tasks)
154+
{
155+
runSideEffectSubTask(task, NO_PREDECESSORS);
156+
}
157+
}
158+
150159
@Override
151160
public void runSubTask(Task<?> task, Task<?> rootTask) {
152161
// check reference equality to make sure model is consistent i.e.

0 commit comments

Comments
 (0)