#201

Committed 16 Oct 2023 03:47PM UTC coverage: 77.389% (+0.02%) from 77.368%

Build # #201

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Apply data converter context in more places (#1896)

Add data converter context to memo, lastFailure and schedules

Run Details

23 of 23 new or added lines in 5 files covered. (100.0%)

18718 of 24187 relevant lines covered (77.39%)

0.77 hits per line

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

95.74

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

import static io.temporal.internal.common.HeaderUtils.toHeaderGrpc;
import static io.temporal.internal.common.SerializerUtils.toRetryPolicy;

import com.google.common.base.MoreObjects;
import com.google.common.base.Strings;
import com.google.protobuf.ByteString;
import io.temporal.api.common.v1.*;
import io.temporal.api.enums.v1.HistoryEventFilterType;
import io.temporal.api.taskqueue.v1.TaskQueue;
import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest;
import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest;
import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest;
import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequestOrBuilder;
import io.temporal.client.WorkflowClientOptions;
import io.temporal.client.WorkflowOptions;
import io.temporal.common.RetryOptions;
import io.temporal.common.context.ContextPropagator;
import io.temporal.internal.common.ProtobufTimeUtils;
import io.temporal.internal.common.SearchAttributesUtil;
import java.util.*;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

final class WorkflowClientRequestFactory {
  private final WorkflowClientOptions clientOptions;

  public WorkflowClientRequestFactory(WorkflowClientOptions clientOptions) {
    this.clientOptions = clientOptions;
  }

  // If you add anything new here, keep newSignalWithStartWorkflowExecutionRequest in sync
  @SuppressWarnings("deprecation")
  @Nonnull
  StartWorkflowExecutionRequest.Builder newStartWorkflowExecutionRequest(
      @Nonnull String workflowId,
      @Nonnull String workflowTypeName,
      @Nonnull io.temporal.common.interceptors.Header header,
      @Nonnull WorkflowOptions options,
      @Nullable Payloads inputArgs,
      @Nullable Memo memo) {
    StartWorkflowExecutionRequest.Builder request =
        StartWorkflowExecutionRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setRequestId(generateUniqueId())
            .setIdentity(clientOptions.getIdentity())
            .setWorkflowId(workflowId)
            .setWorkflowType(WorkflowType.newBuilder().setName(workflowTypeName))
            .setWorkflowRunTimeout(
                ProtobufTimeUtils.toProtoDuration(options.getWorkflowRunTimeout()))
            .setWorkflowExecutionTimeout(
                ProtobufTimeUtils.toProtoDuration(options.getWorkflowExecutionTimeout()))
            .setWorkflowTaskTimeout(
                ProtobufTimeUtils.toProtoDuration(options.getWorkflowTaskTimeout()));

    if (inputArgs != null) {
      request.setInput(inputArgs);
    }

    if (options.getWorkflowIdReusePolicy() != null) {
      request.setWorkflowIdReusePolicy(options.getWorkflowIdReusePolicy());
    }

    String taskQueue = options.getTaskQueue();
    if (taskQueue != null && !taskQueue.isEmpty()) {
      request.setTaskQueue(TaskQueue.newBuilder().setName(taskQueue).build());
    }

    RetryOptions retryOptions = options.getRetryOptions();
    if (retryOptions != null) {
      request.setRetryPolicy(toRetryPolicy(retryOptions));
    }

    if (!Strings.isNullOrEmpty(options.getCronSchedule())) {
      request.setCronSchedule(options.getCronSchedule());
    }

    if (memo != null) {
      request.setMemo(memo);
    }

    if (options.getStartDelay() != null) {
      request.setWorkflowStartDelay(ProtobufTimeUtils.toProtoDuration(options.getStartDelay()));
    }

    if (options.getSearchAttributes() != null && !options.getSearchAttributes().isEmpty()) {
      if (options.getTypedSearchAttributes() != null) {
        throw new IllegalArgumentException(
            "Cannot have search attributes and typed search attributes");
      }
      request.setSearchAttributes(SearchAttributesUtil.encode(options.getSearchAttributes()));
    } else if (options.getTypedSearchAttributes() != null) {
      request.setSearchAttributes(
          SearchAttributesUtil.encodeTyped(options.getTypedSearchAttributes()));
    }

    Header grpcHeader =
        toHeaderGrpc(header, extractContextsAndConvertToBytes(options.getContextPropagators()));
    request.setHeader(grpcHeader);

    return request;
  }

  @Nonnull
  SignalWithStartWorkflowExecutionRequest.Builder newSignalWithStartWorkflowExecutionRequest(
      @Nonnull StartWorkflowExecutionRequestOrBuilder startParameters,
      @Nonnull String signalName,
      @Nullable Payloads signalInput) {
    SignalWithStartWorkflowExecutionRequest.Builder request =
        SignalWithStartWorkflowExecutionRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setRequestId(generateUniqueId())
            .setIdentity(clientOptions.getIdentity())
            .setSignalName(signalName)
            .setWorkflowRunTimeout(startParameters.getWorkflowRunTimeout())
            .setWorkflowExecutionTimeout(startParameters.getWorkflowExecutionTimeout())
            .setWorkflowTaskTimeout(startParameters.getWorkflowTaskTimeout())
            .setWorkflowType(startParameters.getWorkflowType())
            .setWorkflowIdReusePolicy(startParameters.getWorkflowIdReusePolicy())
            .setCronSchedule(startParameters.getCronSchedule());

    String workflowId = startParameters.getWorkflowId();
    if (workflowId.isEmpty()) {
      workflowId = generateUniqueId();
    }
    request.setWorkflowId(workflowId);

    if (signalInput != null) {
      request.setSignalInput(signalInput);
    }

    if (startParameters.hasInput()) {
      request.setInput(startParameters.getInput());
    }

    if (startParameters.hasTaskQueue()) {
      request.setTaskQueue(startParameters.getTaskQueue());
    }

    if (startParameters.hasRetryPolicy()) {
      request.setRetryPolicy(startParameters.getRetryPolicy());
    }

    if (startParameters.hasMemo()) {
      request.setMemo(startParameters.getMemo());
    }

    if (startParameters.hasSearchAttributes()) {
      request.setSearchAttributes(startParameters.getSearchAttributes());
    }

    if (startParameters.hasHeader()) {
      request.setHeader(startParameters.getHeader());
    }

    if (startParameters.hasWorkflowStartDelay()) {
      request.setWorkflowStartDelay(startParameters.getWorkflowStartDelay());
    }

    return request;
  }

  @Nonnull
  GetWorkflowExecutionHistoryRequest newHistoryLongPollRequest(
      WorkflowExecution workflowExecution, ByteString pageToken) {
    return GetWorkflowExecutionHistoryRequest.newBuilder()
        .setNamespace(clientOptions.getNamespace())
        .setExecution(workflowExecution)
        .setHistoryEventFilterType(HistoryEventFilterType.HISTORY_EVENT_FILTER_TYPE_CLOSE_EVENT)
        .setWaitNewEvent(true)
        .setNextPageToken(pageToken)
        .build();
  }

  private io.temporal.common.interceptors.Header extractContextsAndConvertToBytes(
      List<ContextPropagator> workflowOptionsContextPropagators) {
    List<ContextPropagator> workflowClientContextPropagators =
        clientOptions.getContextPropagators();
    if ((workflowClientContextPropagators.isEmpty() && workflowOptionsContextPropagators == null)
        || (workflowOptionsContextPropagators != null
            && workflowOptionsContextPropagators.isEmpty())) {
      return null;
    }

    List<ContextPropagator> listToUse =
        MoreObjects.firstNonNull(
            workflowOptionsContextPropagators, workflowClientContextPropagators);
    Map<String, Payload> result = new HashMap<>();
    for (ContextPropagator propagator : listToUse) {
      result.putAll(propagator.serializeContext(propagator.getCurrentContext()));
    }
    return new io.temporal.common.interceptors.Header(result);
  }

  private static String generateUniqueId() {
    return UUID.randomUUID().toString();
  }
}

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.client;
22
23	import static io.temporal.internal.common.HeaderUtils.toHeaderGrpc;
24	import static io.temporal.internal.common.SerializerUtils.toRetryPolicy;
25
26	import com.google.common.base.MoreObjects;
27	import com.google.common.base.Strings;
28	import com.google.protobuf.ByteString;
29	import io.temporal.api.common.v1.*;
30	import io.temporal.api.enums.v1.HistoryEventFilterType;
31	import io.temporal.api.taskqueue.v1.TaskQueue;
32	import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest;
33	import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest;
34	import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest;
35	import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequestOrBuilder;
36	import io.temporal.client.WorkflowClientOptions;
37	import io.temporal.client.WorkflowOptions;
38	import io.temporal.common.RetryOptions;
39	import io.temporal.common.context.ContextPropagator;
40	import io.temporal.internal.common.ProtobufTimeUtils;
41	import io.temporal.internal.common.SearchAttributesUtil;
42	import java.util.*;
43	import javax.annotation.Nonnull;
44	import javax.annotation.Nullable;
45
46	final class WorkflowClientRequestFactory {
47	private final WorkflowClientOptions clientOptions;
48
49	public WorkflowClientRequestFactory(WorkflowClientOptions clientOptions) {	1✔
50	this.clientOptions = clientOptions;	1✔
51	}	1✔
52
53	// If you add anything new here, keep newSignalWithStartWorkflowExecutionRequest in sync
54	@SuppressWarnings("deprecation")
55	@Nonnull
56	StartWorkflowExecutionRequest.Builder newStartWorkflowExecutionRequest(
57	@Nonnull String workflowId,
58	@Nonnull String workflowTypeName,
59	@Nonnull io.temporal.common.interceptors.Header header,
60	@Nonnull WorkflowOptions options,
61	@Nullable Payloads inputArgs,
62	@Nullable Memo memo) {
63	StartWorkflowExecutionRequest.Builder request =
64	StartWorkflowExecutionRequest.newBuilder()	1✔
65	.setNamespace(clientOptions.getNamespace())	1✔
66	.setRequestId(generateUniqueId())	1✔
67	.setIdentity(clientOptions.getIdentity())	1✔
68	.setWorkflowId(workflowId)	1✔
69	.setWorkflowType(WorkflowType.newBuilder().setName(workflowTypeName))	1✔
70	.setWorkflowRunTimeout(	1✔
71	ProtobufTimeUtils.toProtoDuration(options.getWorkflowRunTimeout()))	1✔
72	.setWorkflowExecutionTimeout(	1✔
73	ProtobufTimeUtils.toProtoDuration(options.getWorkflowExecutionTimeout()))	1✔
74	.setWorkflowTaskTimeout(	1✔
75	ProtobufTimeUtils.toProtoDuration(options.getWorkflowTaskTimeout()));	1✔
76
77	if (inputArgs != null) {	1✔
78	request.setInput(inputArgs);	1✔
79	}
80
81	if (options.getWorkflowIdReusePolicy() != null) {	1✔
82	request.setWorkflowIdReusePolicy(options.getWorkflowIdReusePolicy());	1✔
83	}
84
85	String taskQueue = options.getTaskQueue();	1✔
86	if (taskQueue != null && !taskQueue.isEmpty()) {	1✔
87	request.setTaskQueue(TaskQueue.newBuilder().setName(taskQueue).build());	1✔
88	}
89
90	RetryOptions retryOptions = options.getRetryOptions();	1✔
91	if (retryOptions != null) {	1✔
92	request.setRetryPolicy(toRetryPolicy(retryOptions));	1✔
93	}
94
95	if (!Strings.isNullOrEmpty(options.getCronSchedule())) {	1✔
96	request.setCronSchedule(options.getCronSchedule());	1✔
97	}
98
99	if (memo != null) {	1✔
100	request.setMemo(memo);	1✔
101	}
102
103	if (options.getStartDelay() != null) {	1✔
104	request.setWorkflowStartDelay(ProtobufTimeUtils.toProtoDuration(options.getStartDelay()));	1✔
105	}
106
107	if (options.getSearchAttributes() != null && !options.getSearchAttributes().isEmpty()) {	1✔
108	if (options.getTypedSearchAttributes() != null) {	1✔
109	throw new IllegalArgumentException(	×
110	"Cannot have search attributes and typed search attributes");
111	}
112	request.setSearchAttributes(SearchAttributesUtil.encode(options.getSearchAttributes()));	1✔
113	} else if (options.getTypedSearchAttributes() != null) {	1✔
114	request.setSearchAttributes(	1✔
115	SearchAttributesUtil.encodeTyped(options.getTypedSearchAttributes()));	1✔
116	}
117
118	Header grpcHeader =	1✔
119	toHeaderGrpc(header, extractContextsAndConvertToBytes(options.getContextPropagators()));	1✔
120	request.setHeader(grpcHeader);	1✔
121
122	return request;	1✔
123	}
124
125	@Nonnull
126	SignalWithStartWorkflowExecutionRequest.Builder newSignalWithStartWorkflowExecutionRequest(
127	@Nonnull StartWorkflowExecutionRequestOrBuilder startParameters,
128	@Nonnull String signalName,
129	@Nullable Payloads signalInput) {
130	SignalWithStartWorkflowExecutionRequest.Builder request =
131	SignalWithStartWorkflowExecutionRequest.newBuilder()	1✔
132	.setNamespace(clientOptions.getNamespace())	1✔
133	.setRequestId(generateUniqueId())	1✔
134	.setIdentity(clientOptions.getIdentity())	1✔
135	.setSignalName(signalName)	1✔
136	.setWorkflowRunTimeout(startParameters.getWorkflowRunTimeout())	1✔
137	.setWorkflowExecutionTimeout(startParameters.getWorkflowExecutionTimeout())	1✔
138	.setWorkflowTaskTimeout(startParameters.getWorkflowTaskTimeout())	1✔
139	.setWorkflowType(startParameters.getWorkflowType())	1✔
140	.setWorkflowIdReusePolicy(startParameters.getWorkflowIdReusePolicy())	1✔
141	.setCronSchedule(startParameters.getCronSchedule());	1✔
142
143	String workflowId = startParameters.getWorkflowId();	1✔
144	if (workflowId.isEmpty()) {	1✔
145	workflowId = generateUniqueId();	×
146	}
147	request.setWorkflowId(workflowId);	1✔
148
149	if (signalInput != null) {	1✔
150	request.setSignalInput(signalInput);	1✔
151	}
152
153	if (startParameters.hasInput()) {	1✔
154	request.setInput(startParameters.getInput());	1✔
155	}
156
157	if (startParameters.hasTaskQueue()) {	1✔
158	request.setTaskQueue(startParameters.getTaskQueue());	1✔
159	}
160
161	if (startParameters.hasRetryPolicy()) {	1✔
162	request.setRetryPolicy(startParameters.getRetryPolicy());	×
163	}
164
165	if (startParameters.hasMemo()) {	1✔
166	request.setMemo(startParameters.getMemo());	×
167	}
168
169	if (startParameters.hasSearchAttributes()) {	1✔
170	request.setSearchAttributes(startParameters.getSearchAttributes());	1✔
171	}
172
173	if (startParameters.hasHeader()) {	1✔
174	request.setHeader(startParameters.getHeader());	1✔
175	}
176
177	if (startParameters.hasWorkflowStartDelay()) {	1✔
178	request.setWorkflowStartDelay(startParameters.getWorkflowStartDelay());	1✔
179	}
180
181	return request;	1✔
182	}
183
184	@Nonnull
185	GetWorkflowExecutionHistoryRequest newHistoryLongPollRequest(
186	WorkflowExecution workflowExecution, ByteString pageToken) {
187	return GetWorkflowExecutionHistoryRequest.newBuilder()	1✔
188	.setNamespace(clientOptions.getNamespace())	1✔
189	.setExecution(workflowExecution)	1✔
190	.setHistoryEventFilterType(HistoryEventFilterType.HISTORY_EVENT_FILTER_TYPE_CLOSE_EVENT)	1✔
191	.setWaitNewEvent(true)	1✔
192	.setNextPageToken(pageToken)	1✔
193	.build();	1✔
194	}
195
196	private io.temporal.common.interceptors.Header extractContextsAndConvertToBytes(
197	List<ContextPropagator> workflowOptionsContextPropagators) {
198	List<ContextPropagator> workflowClientContextPropagators =	1✔
199	clientOptions.getContextPropagators();	1✔
200	if ((workflowClientContextPropagators.isEmpty() && workflowOptionsContextPropagators == null)	1✔
201	\|\| (workflowOptionsContextPropagators != null
202	&& workflowOptionsContextPropagators.isEmpty())) {	1✔
203	return null;	1✔
204	}
205
206	List<ContextPropagator> listToUse =	1✔
207	MoreObjects.firstNonNull(	1✔
208	workflowOptionsContextPropagators, workflowClientContextPropagators);
209	Map<String, Payload> result = new HashMap<>();	1✔
210	for (ContextPropagator propagator : listToUse) {	1✔
211	result.putAll(propagator.serializeContext(propagator.getCurrentContext()));	1✔
212	}	1✔
213	return new io.temporal.common.interceptors.Header(result);	1✔
214	}
215
216	private static String generateUniqueId() {
217	return UUID.randomUUID().toString();	1✔
218	}
219	}

temporalio / sdk-java / #201

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