#169

pending completion

Build # #169

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Remove use of deprecated API (#1758)

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

17345 of 21558 relevant lines covered (80.46%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.65

/temporal-testing/src/main/java/io/temporal/testing/internal/TracingWorkerInterceptor.java

/*
 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
 *
 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this material 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 io.temporal.testing.internal;

import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import io.temporal.activity.ActivityExecutionContext;
import io.temporal.client.ActivityCompletionException;
import io.temporal.common.interceptors.*;
import io.temporal.internal.sync.WorkflowMethodThreadNameStrategy;
import io.temporal.workflow.Functions;
import io.temporal.workflow.Promise;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInfo;
import io.temporal.workflow.unsafe.WorkflowUnsafe;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.*;
import java.util.function.BiPredicate;
import java.util.function.Supplier;
import javax.annotation.Nonnull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TracingWorkerInterceptor implements WorkerInterceptor {

  private static final Logger log = LoggerFactory.getLogger(TracingWorkerInterceptor.class);

  private final FilteredTrace trace;
  private List<String> expected;

  public TracingWorkerInterceptor(FilteredTrace trace) {
    this.trace = trace;
  }

  public String getTrace() {
    return String.join("\n", trace.getImpl());
  }

  public void setExpected(String... expected) {
    this.expected = Arrays.asList(expected);
  }

  public void assertExpected() {
    // As it stands, when the trace is empty but the expected list isn't this still passes.
    if (expected != null) {
      List<String> traceElements = trace.getImpl();
      for (int i = 0; i < traceElements.size(); i++) {
        String t = traceElements.get(i);
        String expectedRegExp;
        if (expected.size() <= i) {
          expectedRegExp = "";
        } else {
          expectedRegExp = expected.get(i);
        }
        assertTrue(
            t
                + " doesn't match "
                + expectedRegExp
                + ": \n expected=\n"
                + String.join("\n", expected)
                + "\n actual=\n"
                + String.join("\n", traceElements)
                + "\n",
            t.matches(expectedRegExp));
      }
    }
  }

  @Override
  public WorkflowInboundCallsInterceptor interceptWorkflow(WorkflowInboundCallsInterceptor next) {
    if (!WorkflowUnsafe.isReplaying()) {
      trace.add("interceptExecuteWorkflow " + Workflow.getInfo().getWorkflowId());
    }
    return new WorkflowInboundCallsInterceptorBase(next) {
      @Override
      public void init(WorkflowOutboundCallsInterceptor outboundCalls) {
        next.init(new TracingWorkflowOutboundCallsInterceptor(trace, outboundCalls));
      }

      @Override
      public void handleSignal(SignalInput input) {
        trace.add("handleSignal " + input.getSignalName());
        super.handleSignal(input);
      }

      @Override
      public QueryOutput handleQuery(QueryInput input) {
        trace.add("handleQuery " + input.getQueryName());
        return super.handleQuery(input);
      }

      @Nonnull
      @Override
      public Object newWorkflowMethodThread(Runnable runnable, String name) {
        if (!WorkflowUnsafe.isReplaying()) {
          if (name.startsWith(WorkflowMethodThreadNameStrategy.WORKFLOW_MAIN_THREAD_PREFIX)) {
            // strip the IDs we add to identify WF thread method
            trace.add("newThread " + WorkflowMethodThreadNameStrategy.WORKFLOW_MAIN_THREAD_PREFIX);
          } else {
            trace.add("newThread " + name);
          }
        }
        return next.newWorkflowMethodThread(runnable, name);
      }
    };
  }

  @Override
  public ActivityInboundCallsInterceptor interceptActivity(ActivityInboundCallsInterceptor next) {
    return new TracingActivityInboundCallsInterceptor(trace, next);
  }

  public static class FilteredTrace {

    private final List<String> impl = Collections.synchronizedList(new ArrayList<>());

    public boolean add(String s) {
      log.trace("FilteredTrace isReplaying=" + WorkflowUnsafe.isReplaying());
      if (!WorkflowUnsafe.isReplaying()) {
        return impl.add(s);
      }
      return true;
    }

    List<String> getImpl() {
      return impl;
    }
  }

  private static class TracingWorkflowOutboundCallsInterceptor
      implements WorkflowOutboundCallsInterceptor {

    private final FilteredTrace trace;
    private final WorkflowOutboundCallsInterceptor next;

    private TracingWorkflowOutboundCallsInterceptor(
        FilteredTrace trace, WorkflowOutboundCallsInterceptor next) {
      WorkflowInfo workflowInfo =
          Workflow.getInfo(); // checks that info is available in the constructor
      assertNotNull(workflowInfo);
      this.trace = trace;
      this.next = Objects.requireNonNull(next);
    }

    @Override
    public <R> ActivityOutput<R> executeActivity(ActivityInput<R> input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("executeActivity " + input.getActivityName());
      }
      return next.executeActivity(input);
    }

    @Override
    public <R> LocalActivityOutput<R> executeLocalActivity(LocalActivityInput<R> input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("executeLocalActivity " + input.getActivityName());
      }
      return next.executeLocalActivity(input);
    }

    @Override
    public <R> ChildWorkflowOutput<R> executeChildWorkflow(ChildWorkflowInput<R> input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("executeChildWorkflow " + input.getWorkflowType());
      }
      return next.executeChildWorkflow(input);
    }

    @Override
    public Random newRandom() {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("newRandom");
      }
      return next.newRandom();
    }

    @Override
    public SignalExternalOutput signalExternalWorkflow(SignalExternalInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add(
            "signalExternalWorkflow "
                + input.getExecution().getWorkflowId()
                + " "
                + input.getSignalName());
      }
      return next.signalExternalWorkflow(input);
    }

    @Override
    public CancelWorkflowOutput cancelWorkflow(CancelWorkflowInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("cancelWorkflow " + input.getExecution().getWorkflowId());
      }
      return next.cancelWorkflow(input);
    }

    @Override
    public void sleep(Duration duration) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("sleep " + duration);
      }
      next.sleep(duration);
    }

    @Override
    public boolean await(Duration timeout, String reason, Supplier<Boolean> unblockCondition) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("await " + timeout + " " + reason);
      }
      return next.await(timeout, reason, unblockCondition);
    }

    @Override
    public void await(String reason, Supplier<Boolean> unblockCondition) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("await " + reason);
      }
      next.await(reason, unblockCondition);
    }

    @Override
    public Promise<Void> newTimer(Duration duration) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("newTimer " + duration);
      }
      return next.newTimer(duration);
    }

    @Override
    public <R> R sideEffect(Class<R> resultClass, Type resultType, Functions.Func<R> func) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("sideEffect");
      }
      return next.sideEffect(resultClass, resultType, func);
    }

    @Override
    public <R> R mutableSideEffect(
        String id,
        Class<R> resultClass,
        Type resultType,
        BiPredicate<R, R> updated,
        Functions.Func<R> func) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("mutableSideEffect");
      }
      return next.mutableSideEffect(id, resultClass, resultType, updated, func);
    }

    @Override
    public int getVersion(String changeId, int minSupported, int maxSupported) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("getVersion");
      }
      return next.getVersion(changeId, minSupported, maxSupported);
    }

    @Override
    public void continueAsNew(ContinueAsNewInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("continueAsNew");
      }
      next.continueAsNew(input);
    }

    @Override
    public void registerQuery(RegisterQueryInput input) {
      String queryType = input.getQueryType();
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("registerQuery " + queryType);
      }
      next.registerQuery(
          new RegisterQueryInput(
              queryType,
              input.getArgTypes(),
              input.getGenericArgTypes(),
              (args) -> {
                Object result = input.getCallback().apply(args);
                if (!WorkflowUnsafe.isReplaying()) {
                  if (queryType.equals("query")) {
                    log.trace("query", new Throwable());
                  }
                  trace.add("query " + queryType);
                }
                return result;
              }));
    }

    @Override
    public void registerSignalHandlers(RegisterSignalHandlersInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        StringBuilder signals = new StringBuilder();
        for (SignalRegistrationRequest request : input.getRequests()) {
          if (signals.length() > 0) {
            signals.append(", ");
          }
          signals.append(request.getSignalType());
        }
        trace.add("registerSignalHandlers " + signals);
      }
      next.registerSignalHandlers(input);
    }

    @Override
    public void registerUpdateHandlers(RegisterUpdateHandlersInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        StringBuilder updates = new StringBuilder();
        for (UpdateRegistrationRequest request : input.getRequests()) {
          if (updates.length() > 0) {
            updates.append(", ");
          }
          updates.append(request.getUpdateName());
        }
        trace.add("registerUpdateHandlers " + updates);
      }
      next.registerUpdateHandlers(input);
    }

    @Override
    public void registerDynamicUpdateHandler(RegisterDynamicUpdateHandlerInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("registerDynamicUpdateHandler");
      }
      this.registerDynamicUpdateHandler(input);
    }

    @Override
    public void registerDynamicSignalHandler(RegisterDynamicSignalHandlerInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("registerDynamicSignalHandler");
      }
      next.registerDynamicSignalHandler(input);
    }

    @Override
    public void registerDynamicQueryHandler(RegisterDynamicQueryHandlerInput input) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("registerDynamicQueryHandler");
      }
      next.registerDynamicQueryHandler(input);
    }

    @Override
    public UUID randomUUID() {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("randomUUID");
      }
      return next.randomUUID();
    }

    @Override
    public void upsertSearchAttributes(Map<String, ?> searchAttributes) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("upsertSearchAttributes");
      }
      next.upsertSearchAttributes(searchAttributes);
    }

    @Override
    public Object newChildThread(Runnable runnable, boolean detached, String name) {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("newThread " + name);
      }
      return next.newChildThread(runnable, detached, name);
    }

    @Override
    public long currentTimeMillis() {
      if (!WorkflowUnsafe.isReplaying()) {
        trace.add("currentTimeMillis");
      }
      return next.currentTimeMillis();
    }
  }

  private static class TracingActivityInboundCallsInterceptor
      implements ActivityInboundCallsInterceptor {

    private final FilteredTrace trace;
    private final ActivityInboundCallsInterceptor next;
    private String type;
    private boolean local;

    public TracingActivityInboundCallsInterceptor(
        FilteredTrace trace, ActivityInboundCallsInterceptor next) {
      this.trace = trace;
      this.next = next;
    }

    @Override
    public void init(ActivityExecutionContext context) {
      this.type = context.getInfo().getActivityType();
      this.local = context.getInfo().isLocal();
      next.init(
          new ActivityExecutionContextBase(context) {
            @Override
            public <V> void heartbeat(V details) throws ActivityCompletionException {
              trace.add("heartbeat " + details);
              super.heartbeat(details);
            }
          });
    }

    @Override
    public ActivityOutput execute(ActivityInput input) {
      trace.add((local ? "local " : "") + "activity " + type);
      return next.execute(input);
    }
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.testing.internal;
22
23	import static org.junit.Assert.assertNotNull;
24	import static org.junit.Assert.assertTrue;
25
26	import io.temporal.activity.ActivityExecutionContext;
27	import io.temporal.client.ActivityCompletionException;
28	import io.temporal.common.interceptors.*;
29	import io.temporal.internal.sync.WorkflowMethodThreadNameStrategy;
30	import io.temporal.workflow.Functions;
31	import io.temporal.workflow.Promise;
32	import io.temporal.workflow.Workflow;
33	import io.temporal.workflow.WorkflowInfo;
34	import io.temporal.workflow.unsafe.WorkflowUnsafe;
35	import java.lang.reflect.Type;
36	import java.time.Duration;
37	import java.util.*;
38	import java.util.function.BiPredicate;
39	import java.util.function.Supplier;
40	import javax.annotation.Nonnull;
41	import org.slf4j.Logger;
42	import org.slf4j.LoggerFactory;
43
44	public class TracingWorkerInterceptor implements WorkerInterceptor {
45
46	private static final Logger log = LoggerFactory.getLogger(TracingWorkerInterceptor.class);	1✔
47
48	private final FilteredTrace trace;
49	private List<String> expected;
50
51	public TracingWorkerInterceptor(FilteredTrace trace) {	1✔
52	this.trace = trace;	1✔
53	}	1✔
54
55	public String getTrace() {
56	return String.join("\n", trace.getImpl());	1✔
57	}
58
59	public void setExpected(String... expected) {
60	this.expected = Arrays.asList(expected);	1✔
61	}	1✔
62
63	public void assertExpected() {
64	// As it stands, when the trace is empty but the expected list isn't this still passes.
65	if (expected != null) {	×
66	List<String> traceElements = trace.getImpl();	×
67	for (int i = 0; i < traceElements.size(); i++) {	×
68	String t = traceElements.get(i);	×
69	String expectedRegExp;
70	if (expected.size() <= i) {	×
71	expectedRegExp = "";	×
72	} else {
73	expectedRegExp = expected.get(i);	×
74	}
75	assertTrue(	×
76	t
77	+ " doesn't match "
78	+ expectedRegExp
79	+ ": \n expected=\n"
80	+ String.join("\n", expected)	×
81	+ "\n actual=\n"
82	+ String.join("\n", traceElements)	×
83	+ "\n",
84	t.matches(expectedRegExp));	×
85	}
86	}
87	}	×
88
89	@Override
90	public WorkflowInboundCallsInterceptor interceptWorkflow(WorkflowInboundCallsInterceptor next) {
91	if (!WorkflowUnsafe.isReplaying()) {	1✔
92	trace.add("interceptExecuteWorkflow " + Workflow.getInfo().getWorkflowId());	1✔
93	}
94	return new WorkflowInboundCallsInterceptorBase(next) {	1✔
95	@Override
96	public void init(WorkflowOutboundCallsInterceptor outboundCalls) {
97	next.init(new TracingWorkflowOutboundCallsInterceptor(trace, outboundCalls));	1✔
98	}	1✔
99
100	@Override
101	public void handleSignal(SignalInput input) {
102	trace.add("handleSignal " + input.getSignalName());	1✔
103	super.handleSignal(input);	1✔
104	}	1✔
105
106	@Override
107	public QueryOutput handleQuery(QueryInput input) {
108	trace.add("handleQuery " + input.getQueryName());	1✔
109	return super.handleQuery(input);	1✔
110	}
111
112	@Nonnull
113	@Override
114	public Object newWorkflowMethodThread(Runnable runnable, String name) {
115	if (!WorkflowUnsafe.isReplaying()) {	1✔
116	if (name.startsWith(WorkflowMethodThreadNameStrategy.WORKFLOW_MAIN_THREAD_PREFIX)) {	1✔
117	// strip the IDs we add to identify WF thread method
118	trace.add("newThread " + WorkflowMethodThreadNameStrategy.WORKFLOW_MAIN_THREAD_PREFIX);	1✔
119	} else {
120	trace.add("newThread " + name);	×
121	}
122	}
123	return next.newWorkflowMethodThread(runnable, name);	1✔
124	}
125	};
126	}
127
128	@Override
129	public ActivityInboundCallsInterceptor interceptActivity(ActivityInboundCallsInterceptor next) {
130	return new TracingActivityInboundCallsInterceptor(trace, next);	1✔
131	}
132
133	public static class FilteredTrace {	1✔
134
135	private final List<String> impl = Collections.synchronizedList(new ArrayList<>());	1✔
136
137	public boolean add(String s) {
138	log.trace("FilteredTrace isReplaying=" + WorkflowUnsafe.isReplaying());	1✔
139	if (!WorkflowUnsafe.isReplaying()) {	1✔
140	return impl.add(s);	1✔
141	}
142	return true;	1✔
143	}
144
145	List<String> getImpl() {
146	return impl;	1✔
147	}
148	}
149
150	private static class TracingWorkflowOutboundCallsInterceptor
151	implements WorkflowOutboundCallsInterceptor {
152
153	private final FilteredTrace trace;
154	private final WorkflowOutboundCallsInterceptor next;
155
156	private TracingWorkflowOutboundCallsInterceptor(
157	FilteredTrace trace, WorkflowOutboundCallsInterceptor next) {	1✔
158	WorkflowInfo workflowInfo =
159	Workflow.getInfo(); // checks that info is available in the constructor	1✔
160	assertNotNull(workflowInfo);	1✔
161	this.trace = trace;	1✔
162	this.next = Objects.requireNonNull(next);	1✔
163	}	1✔
164
165	@Override
166	public <R> ActivityOutput<R> executeActivity(ActivityInput<R> input) {
167	if (!WorkflowUnsafe.isReplaying()) {	1✔
168	trace.add("executeActivity " + input.getActivityName());	1✔
169	}
170	return next.executeActivity(input);	1✔
171	}
172
173	@Override
174	public <R> LocalActivityOutput<R> executeLocalActivity(LocalActivityInput<R> input) {
175	if (!WorkflowUnsafe.isReplaying()) {	1✔
176	trace.add("executeLocalActivity " + input.getActivityName());	1✔
177	}
178	return next.executeLocalActivity(input);	1✔
179	}
180
181	@Override
182	public <R> ChildWorkflowOutput<R> executeChildWorkflow(ChildWorkflowInput<R> input) {
183	if (!WorkflowUnsafe.isReplaying()) {	1✔
184	trace.add("executeChildWorkflow " + input.getWorkflowType());	1✔
185	}
186	return next.executeChildWorkflow(input);	1✔
187	}
188
189	@Override
190	public Random newRandom() {
191	if (!WorkflowUnsafe.isReplaying()) {	×
192	trace.add("newRandom");	×
193	}
194	return next.newRandom();	×
195	}
196
197	@Override
198	public SignalExternalOutput signalExternalWorkflow(SignalExternalInput input) {
199	if (!WorkflowUnsafe.isReplaying()) {	1✔
200	trace.add(	1✔
201	"signalExternalWorkflow "
202	+ input.getExecution().getWorkflowId()	1✔
203	+ " "
204	+ input.getSignalName());	1✔
205	}
206	return next.signalExternalWorkflow(input);	1✔
207	}
208
209	@Override
210	public CancelWorkflowOutput cancelWorkflow(CancelWorkflowInput input) {
211	if (!WorkflowUnsafe.isReplaying()) {	×
212	trace.add("cancelWorkflow " + input.getExecution().getWorkflowId());	×
213	}
214	return next.cancelWorkflow(input);	×
215	}
216
217	@Override
218	public void sleep(Duration duration) {
219	if (!WorkflowUnsafe.isReplaying()) {	1✔
220	trace.add("sleep " + duration);	1✔
221	}
222	next.sleep(duration);	1✔
223	}	1✔
224
225	@Override
226	public boolean await(Duration timeout, String reason, Supplier<Boolean> unblockCondition) {
227	if (!WorkflowUnsafe.isReplaying()) {	1✔
228	trace.add("await " + timeout + " " + reason);	1✔
229	}
230	return next.await(timeout, reason, unblockCondition);	1✔
231	}
232
233	@Override
234	public void await(String reason, Supplier<Boolean> unblockCondition) {
235	if (!WorkflowUnsafe.isReplaying()) {	1✔
236	trace.add("await " + reason);	1✔
237	}
238	next.await(reason, unblockCondition);	1✔
239	}	1✔
240
241	@Override
242	public Promise<Void> newTimer(Duration duration) {
243	if (!WorkflowUnsafe.isReplaying()) {	1✔
244	trace.add("newTimer " + duration);	1✔
245	}
246	return next.newTimer(duration);	1✔
247	}
248
249	@Override
250	public <R> R sideEffect(Class<R> resultClass, Type resultType, Functions.Func<R> func) {
251	if (!WorkflowUnsafe.isReplaying()) {	1✔
252	trace.add("sideEffect");	1✔
253	}
254	return next.sideEffect(resultClass, resultType, func);	1✔
255	}
256
257	@Override
258	public <R> R mutableSideEffect(
259	String id,
260	Class<R> resultClass,
261	Type resultType,
262	BiPredicate<R, R> updated,
263	Functions.Func<R> func) {
264	if (!WorkflowUnsafe.isReplaying()) {	1✔
265	trace.add("mutableSideEffect");	1✔
266	}
267	return next.mutableSideEffect(id, resultClass, resultType, updated, func);	1✔
268	}
269
270	@Override
271	public int getVersion(String changeId, int minSupported, int maxSupported) {
272	if (!WorkflowUnsafe.isReplaying()) {	1✔
273	trace.add("getVersion");	1✔
274	}
275	return next.getVersion(changeId, minSupported, maxSupported);	1✔
276	}
277
278	@Override
279	public void continueAsNew(ContinueAsNewInput input) {
280	if (!WorkflowUnsafe.isReplaying()) {	1✔
281	trace.add("continueAsNew");	1✔
282	}
283	next.continueAsNew(input);	×
284	}	×
285
286	@Override
287	public void registerQuery(RegisterQueryInput input) {
288	String queryType = input.getQueryType();	1✔
289	if (!WorkflowUnsafe.isReplaying()) {	1✔
290	trace.add("registerQuery " + queryType);	1✔
291	}
292	next.registerQuery(	1✔
293	new RegisterQueryInput(
294	queryType,
295	input.getArgTypes(),	1✔
296	input.getGenericArgTypes(),	1✔
297	(args) -> {
298	Object result = input.getCallback().apply(args);	1✔
299	if (!WorkflowUnsafe.isReplaying()) {	1✔
300	if (queryType.equals("query")) {	1✔
301	log.trace("query", new Throwable());	1✔
302	}
303	trace.add("query " + queryType);	1✔
304	}
305	return result;	1✔
306	}));
307	}	1✔
308
309	@Override
310	public void registerSignalHandlers(RegisterSignalHandlersInput input) {
311	if (!WorkflowUnsafe.isReplaying()) {	1✔
312	StringBuilder signals = new StringBuilder();	1✔
313	for (SignalRegistrationRequest request : input.getRequests()) {	1✔
314	if (signals.length() > 0) {	1✔
315	signals.append(", ");	×
316	}
317	signals.append(request.getSignalType());	1✔
318	}	1✔
319	trace.add("registerSignalHandlers " + signals);	1✔
320	}
321	next.registerSignalHandlers(input);	1✔
322	}	1✔
323
324	@Override
325	public void registerUpdateHandlers(RegisterUpdateHandlersInput input) {
326	if (!WorkflowUnsafe.isReplaying()) {	×
327	StringBuilder updates = new StringBuilder();	×
328	for (UpdateRegistrationRequest request : input.getRequests()) {	×
329	if (updates.length() > 0) {	×
330	updates.append(", ");	×
331	}
332	updates.append(request.getUpdateName());	×
333	}	×
334	trace.add("registerUpdateHandlers " + updates);	×
335	}
336	next.registerUpdateHandlers(input);	×
337	}	×
338
339	@Override
340	public void registerDynamicUpdateHandler(RegisterDynamicUpdateHandlerInput input) {
341	if (!WorkflowUnsafe.isReplaying()) {	×
342	trace.add("registerDynamicUpdateHandler");	×
343	}
344	this.registerDynamicUpdateHandler(input);	×
345	}	×
346
347	@Override
348	public void registerDynamicSignalHandler(RegisterDynamicSignalHandlerInput input) {
349	if (!WorkflowUnsafe.isReplaying()) {	1✔
350	trace.add("registerDynamicSignalHandler");	1✔
351	}
352	next.registerDynamicSignalHandler(input);	1✔
353	}	1✔
354
355	@Override
356	public void registerDynamicQueryHandler(RegisterDynamicQueryHandlerInput input) {
357	if (!WorkflowUnsafe.isReplaying()) {	1✔
358	trace.add("registerDynamicQueryHandler");	1✔
359	}
360	next.registerDynamicQueryHandler(input);	1✔
361	}	1✔
362
363	@Override
364	public UUID randomUUID() {
365	if (!WorkflowUnsafe.isReplaying()) {	×
366	trace.add("randomUUID");	×
367	}
368	return next.randomUUID();	×
369	}
370
371	@Override
372	public void upsertSearchAttributes(Map<String, ?> searchAttributes) {
373	if (!WorkflowUnsafe.isReplaying()) {	1✔
374	trace.add("upsertSearchAttributes");	1✔
375	}
376	next.upsertSearchAttributes(searchAttributes);	1✔
377	}	1✔
378
379	@Override
380	public Object newChildThread(Runnable runnable, boolean detached, String name) {
381	if (!WorkflowUnsafe.isReplaying()) {	1✔
382	trace.add("newThread " + name);	1✔
383	}
384	return next.newChildThread(runnable, detached, name);	1✔
385	}
386
387	@Override
388	public long currentTimeMillis() {
389	if (!WorkflowUnsafe.isReplaying()) {	1✔
390	trace.add("currentTimeMillis");	1✔
391	}
392	return next.currentTimeMillis();	1✔
393	}
394	}
395
396	private static class TracingActivityInboundCallsInterceptor
397	implements ActivityInboundCallsInterceptor {
398
399	private final FilteredTrace trace;
400	private final ActivityInboundCallsInterceptor next;
401	private String type;
402	private boolean local;
403
404	public TracingActivityInboundCallsInterceptor(
405	FilteredTrace trace, ActivityInboundCallsInterceptor next) {	1✔
406	this.trace = trace;	1✔
407	this.next = next;	1✔
408	}	1✔
409
410	@Override
411	public void init(ActivityExecutionContext context) {
412	this.type = context.getInfo().getActivityType();	1✔
413	this.local = context.getInfo().isLocal();	1✔
414	next.init(	1✔
415	new ActivityExecutionContextBase(context) {	1✔
416	@Override
417	public <V> void heartbeat(V details) throws ActivityCompletionException {
418	trace.add("heartbeat " + details);	1✔
419	super.heartbeat(details);	1✔
420	}	1✔
421	});
422	}	1✔
423
424	@Override
425	public ActivityOutput execute(ActivityInput input) {
426	trace.add((local ? "local " : "") + "activity " + type);	1✔
427	return next.execute(input);	1✔
428	}
429	}
430	}

temporalio / sdk-java / #169

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous