#157

pending completion

Build # #157

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Provide SerializationContext for PayloadConverter and PayloadCodec (#1695)

Issue #1694

Run Details

497 of 497 new or added lines in 32 files covered. (100.0%)

16942 of 20806 relevant lines covered (81.43%)

0.81 hits per line

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

96.97

/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.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.internal.activity;

import com.uber.m3.tally.Scope;
import io.temporal.activity.ActivityExecutionContext;
import io.temporal.activity.ActivityInfo;
import io.temporal.activity.ManualActivityCompletionClient;
import io.temporal.client.ActivityCompletionException;
import io.temporal.common.converter.DataConverter;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.payload.context.ActivitySerializationContext;
import io.temporal.serviceclient.WorkflowServiceStubs;
import io.temporal.workflow.Functions;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Base implementation of an {@link ActivityExecutionContext}.
 *
 * @author fateev, suskin
 * @see ActivityExecutionContext
 */
@ThreadSafe
class ActivityExecutionContextImpl implements ActivityExecutionContext {
  private final Lock lock = new ReentrantLock();
  private final ManualActivityCompletionClientFactory manualCompletionClientFactory;
  private final Functions.Proc completionHandle;
  private final HeartbeatContext heartbeatContext;

  private final Scope metricsScope;
  private final ActivityInfo info;
  private boolean useLocalManualCompletion;
  private boolean doNotCompleteOnReturn;

  /** Create an ActivityExecutionContextImpl with the given attributes. */
  ActivityExecutionContextImpl(
      WorkflowServiceStubs service,
      String namespace,
      ActivityInfo info,
      DataConverter dataConverter,
      ScheduledExecutorService heartbeatExecutor,
      ManualActivityCompletionClientFactory manualCompletionClientFactory,
      Functions.Proc completionHandle,
      Scope metricsScope,
      String identity,
      Duration maxHeartbeatThrottleInterval,
      Duration defaultHeartbeatThrottleInterval) {
    this.metricsScope = metricsScope;
    this.info = info;
    this.completionHandle = completionHandle;
    this.manualCompletionClientFactory = manualCompletionClientFactory;
    this.heartbeatContext =
        new HeartbeatContextImpl(
            service,
            namespace,
            info,
            dataConverter,
            heartbeatExecutor,
            metricsScope,
            identity,
            maxHeartbeatThrottleInterval,
            defaultHeartbeatThrottleInterval);
  }

  /**
   * @see ActivityExecutionContext#heartbeat(Object)
   */
  @Override
  public <V> void heartbeat(V details) throws ActivityCompletionException {
    heartbeatContext.heartbeat(details);
  }

  @Override
  public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
    return getHeartbeatDetails(detailsClass, detailsClass);
  }

  @Override
  public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
    return heartbeatContext.getHeartbeatDetails(detailsClass, detailsGenericType);
  }

  @Override
  public byte[] getTaskToken() {
    return info.getTaskToken();
  }

  @Override
  public void doNotCompleteOnReturn() {
    lock.lock();
    try {
      doNotCompleteOnReturn = true;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public boolean isDoNotCompleteOnReturn() {
    lock.lock();
    try {
      return doNotCompleteOnReturn;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public boolean isUseLocalManualCompletion() {
    lock.lock();
    try {
      return useLocalManualCompletion;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public ManualActivityCompletionClient useLocalManualCompletion() {
    lock.lock();
    try {
      doNotCompleteOnReturn();
      useLocalManualCompletion = true;
      ActivitySerializationContext activitySerializationContext =
          new ActivitySerializationContext(info);
      return new CompletionAwareManualCompletionClient(
          manualCompletionClientFactory.getClient(
              info.getTaskToken(), metricsScope, activitySerializationContext),
          completionHandle);
    } finally {
      lock.unlock();
    }
  }

  @Override
  public Scope getMetricsScope() {
    return metricsScope;
  }

  @Override
  public ActivityInfo getInfo() {
    return info;
  }
}

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.internal.activity;
22
23	import com.uber.m3.tally.Scope;
24	import io.temporal.activity.ActivityExecutionContext;
25	import io.temporal.activity.ActivityInfo;
26	import io.temporal.activity.ManualActivityCompletionClient;
27	import io.temporal.client.ActivityCompletionException;
28	import io.temporal.common.converter.DataConverter;
29	import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
30	import io.temporal.payload.context.ActivitySerializationContext;
31	import io.temporal.serviceclient.WorkflowServiceStubs;
32	import io.temporal.workflow.Functions;
33	import java.lang.reflect.Type;
34	import java.time.Duration;
35	import java.util.Optional;
36	import java.util.concurrent.ScheduledExecutorService;
37	import java.util.concurrent.locks.Lock;
38	import java.util.concurrent.locks.ReentrantLock;
39	import javax.annotation.concurrent.ThreadSafe;
40
41	/**
42	* Base implementation of an {@link ActivityExecutionContext}.
43	*
44	* @author fateev, suskin
45	* @see ActivityExecutionContext
46	*/
47	@ThreadSafe
48	class ActivityExecutionContextImpl implements ActivityExecutionContext {
49	private final Lock lock = new ReentrantLock();	1✔
50	private final ManualActivityCompletionClientFactory manualCompletionClientFactory;
51	private final Functions.Proc completionHandle;
52	private final HeartbeatContext heartbeatContext;
53
54	private final Scope metricsScope;
55	private final ActivityInfo info;
56	private boolean useLocalManualCompletion;
57	private boolean doNotCompleteOnReturn;
58
59	/** Create an ActivityExecutionContextImpl with the given attributes. */
60	ActivityExecutionContextImpl(
61	WorkflowServiceStubs service,
62	String namespace,
63	ActivityInfo info,
64	DataConverter dataConverter,
65	ScheduledExecutorService heartbeatExecutor,
66	ManualActivityCompletionClientFactory manualCompletionClientFactory,
67	Functions.Proc completionHandle,
68	Scope metricsScope,
69	String identity,
70	Duration maxHeartbeatThrottleInterval,
71	Duration defaultHeartbeatThrottleInterval) {	1✔
72	this.metricsScope = metricsScope;	1✔
73	this.info = info;	1✔
74	this.completionHandle = completionHandle;	1✔
75	this.manualCompletionClientFactory = manualCompletionClientFactory;	1✔
76	this.heartbeatContext =	1✔
77	new HeartbeatContextImpl(
78	service,
79	namespace,
80	info,
81	dataConverter,
82	heartbeatExecutor,
83	metricsScope,
84	identity,
85	maxHeartbeatThrottleInterval,
86	defaultHeartbeatThrottleInterval);
87	}	1✔
88
89	/**
90	* @see ActivityExecutionContext#heartbeat(Object)
91	*/
92	@Override
93	public <V> void heartbeat(V details) throws ActivityCompletionException {
94	heartbeatContext.heartbeat(details);	1✔
95	}	1✔
96
97	@Override
98	public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
99	return getHeartbeatDetails(detailsClass, detailsClass);	1✔
100	}
101
102	@Override
103	public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsGenericType) {
104	return heartbeatContext.getHeartbeatDetails(detailsClass, detailsGenericType);	1✔
105	}
106
107	@Override
108	public byte[] getTaskToken() {
109	return info.getTaskToken();	×
110	}
111
112	@Override
113	public void doNotCompleteOnReturn() {
114	lock.lock();	1✔
115	try {
116	doNotCompleteOnReturn = true;	1✔
117	} finally {
118	lock.unlock();	1✔
119	}
120	}	1✔
121
122	@Override
123	public boolean isDoNotCompleteOnReturn() {
124	lock.lock();	1✔
125	try {
126	return doNotCompleteOnReturn;	1✔
127	} finally {
128	lock.unlock();	1✔
129	}
130	}
131
132	@Override
133	public boolean isUseLocalManualCompletion() {
134	lock.lock();	1✔
135	try {
136	return useLocalManualCompletion;	1✔
137	} finally {
138	lock.unlock();	1✔
139	}
140	}
141
142	@Override
143	public ManualActivityCompletionClient useLocalManualCompletion() {
144	lock.lock();	1✔
145	try {
146	doNotCompleteOnReturn();	1✔
147	useLocalManualCompletion = true;	1✔
148	ActivitySerializationContext activitySerializationContext =	1✔
149	new ActivitySerializationContext(info);
150	return new CompletionAwareManualCompletionClient(	1✔
151	manualCompletionClientFactory.getClient(	1✔
152	info.getTaskToken(), metricsScope, activitySerializationContext),	1✔
153	completionHandle);
154	} finally {
155	lock.unlock();	1✔
156	}
157	}
158
159	@Override
160	public Scope getMetricsScope() {
161	return metricsScope;	1✔
162	}
163
164	@Override
165	public ActivityInfo getInfo() {
166	return info;	1✔
167	}
168	}

temporalio / sdk-java / #157

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