#292

Committed 31 Jul 2024 04:22PM UTC coverage: 77.571% (-0.005%) from 77.576%

Build # #292

Build Type

push

github

Committed by

web-flow

Commit Message

Add support for query in listSchedules (#2163)

Run Details

0 of 16 new or added lines in 4 files covered. (0.0%)

1 existing line in 1 file now uncovered.

19724 of 25427 relevant lines covered (77.57%)

0.78 hits per line

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

3.77

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

import com.google.common.collect.Iterators;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.temporal.api.common.v1.Memo;
import io.temporal.api.schedule.v1.*;
import io.temporal.api.workflowservice.v1.*;
import io.temporal.client.ListScheduleListDescriptionIterator;
import io.temporal.client.schedules.*;
import io.temporal.common.interceptors.ScheduleClientCallsInterceptor;
import io.temporal.internal.client.external.GenericWorkflowClient;
import io.temporal.internal.common.ProtobufTimeUtils;
import io.temporal.internal.common.SearchAttributesUtil;
import java.util.*;
import java.util.stream.StreamSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RootScheduleClientInvoker implements ScheduleClientCallsInterceptor {
  private static final Logger log = LoggerFactory.getLogger(RootScheduleClientInvoker.class);

  private final GenericWorkflowClient genericClient;

  private final ScheduleClientOptions clientOptions;

  private final ScheduleProtoUtil scheduleRequestHeader;

  public RootScheduleClientInvoker(
      GenericWorkflowClient genericClient, ScheduleClientOptions clientOptions) {
    this.genericClient = genericClient;
    this.clientOptions = clientOptions;
    this.scheduleRequestHeader = new ScheduleProtoUtil(genericClient, clientOptions);
  }

  @Override
  @SuppressWarnings("deprecation")
  public void createSchedule(CreateScheduleInput input) {

    CreateScheduleRequest.Builder request =
        CreateScheduleRequest.newBuilder()
            .setIdentity(clientOptions.getIdentity())
            .setNamespace(clientOptions.getNamespace())
            .setRequestId(UUID.randomUUID().toString())
            .setScheduleId(input.getId())
            .setSchedule(scheduleRequestHeader.scheduleToProto(input.getSchedule()));

    if (input.getOptions().getMemo() != null) {
      // TODO we don't have a workflow context here, maybe we need a schedule context?
      request.setMemo(
          Memo.newBuilder()
              .putAllFields(
                  intoPayloadMap(clientOptions.getDataConverter(), input.getOptions().getMemo())));
    }

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

    if (input.getOptions().isTriggerImmediately()
        || (input.getOptions().getBackfills() != null
            && input.getOptions().getBackfills().size() > 0)) {
      SchedulePatch.Builder patchBuilder = SchedulePatch.newBuilder();

      if (input.getOptions().getBackfills() != null) {
        input.getOptions().getBackfills().stream()
            .forEach(b -> patchBuilder.addBackfillRequest(backfillToProto(b)));
      }

      if (input.getOptions().isTriggerImmediately()) {
        TriggerImmediatelyRequest.Builder triggerRequest = TriggerImmediatelyRequest.newBuilder();
        if (input.getSchedule().getPolicy() != null) {
          triggerRequest.setOverlapPolicy(input.getSchedule().getPolicy().getOverlap());
        }
        patchBuilder.setTriggerImmediately(triggerRequest.build());
      }

      request.setInitialPatch(patchBuilder.build());
    }

    try {
      genericClient.createSchedule(request.build());
    } catch (StatusRuntimeException e) {
      if (Status.Code.ALREADY_EXISTS.equals(e.getStatus().getCode())) {
        throw new ScheduleAlreadyRunningException(e);
      } else {
        throw new ScheduleException(e);
      }
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }

  @Override
  public ListScheduleOutput listSchedules(ListSchedulesInput input) {
    ListScheduleListDescriptionIterator iterator =
        new ListScheduleListDescriptionIterator(
            clientOptions.getNamespace(), input.getQuery(), input.getPageSize(), genericClient);
    iterator.init();
    Iterator<ScheduleListDescription> wrappedIterator =
        Iterators.transform(
            iterator, entry -> scheduleRequestHeader.protoToScheduleListDescription(entry));

    final int CHARACTERISTICS =
        Spliterator.ORDERED | Spliterator.DISTINCT | Spliterator.NONNULL | Spliterator.IMMUTABLE;
    return new ListScheduleOutput(
        StreamSupport.stream(
            Spliterators.spliteratorUnknownSize(wrappedIterator, CHARACTERISTICS), false));
  }

  public BackfillRequest backfillToProto(ScheduleBackfill backfill) {
    return BackfillRequest.newBuilder()
        .setStartTime(ProtobufTimeUtils.toProtoTimestamp(backfill.getStartAt()))
        .setEndTime(ProtobufTimeUtils.toProtoTimestamp(backfill.getEndAt()))
        .setOverlapPolicy(backfill.getOverlapPolicy())
        .build();
  }

  @Override
  public void backfillSchedule(BackfillScheduleInput input) {
    ArrayList<BackfillRequest> backfillRequests =
        new ArrayList<BackfillRequest>(input.getBackfills().size());
    for (ScheduleBackfill backfill : input.getBackfills()) {
      backfillRequests.add(backfillToProto(backfill));
    }

    SchedulePatch patch =
        SchedulePatch.newBuilder().addAllBackfillRequest(backfillRequests).build();

    PatchScheduleRequest request =
        PatchScheduleRequest.newBuilder()
            .setIdentity(clientOptions.getIdentity())
            .setNamespace(clientOptions.getNamespace())
            .setScheduleId(input.getScheduleId())
            .setPatch(patch)
            .build();
    try {
      genericClient.patchSchedule(request);
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }

  @Override
  public void deleteSchedule(DeleteScheduleInput input) {
    DeleteScheduleRequest request =
        DeleteScheduleRequest.newBuilder()
            .setIdentity(clientOptions.getIdentity())
            .setNamespace(clientOptions.getNamespace())
            .setScheduleId(input.getScheduleId())
            .build();
    try {
      genericClient.deleteSchedule(request);
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }

  @Override
  public DescribeScheduleOutput describeSchedule(DescribeScheduleInput input) {
    DescribeScheduleRequest request =
        DescribeScheduleRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setScheduleId(input.getScheduleId())
            .build();

    try {
      DescribeScheduleResponse response = genericClient.describeSchedule(request);
      return new DescribeScheduleOutput(
          new ScheduleDescription(
              input.getScheduleId(),
              scheduleRequestHeader.protoToScheduleInfo(response.getInfo()),
              scheduleRequestHeader.protoToSchedule(response.getSchedule()),
              Collections.unmodifiableMap(
                  SearchAttributesUtil.decode(response.getSearchAttributes())),
              SearchAttributesUtil.decodeTyped(response.getSearchAttributes()),
              response.getMemo().getFieldsMap(),
              clientOptions.getDataConverter()));
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }

  @Override
  public void pauseSchedule(PauseScheduleInput input) {
    SchedulePatch patch = SchedulePatch.newBuilder().setPause(input.getNote()).build();

    PatchScheduleRequest request =
        PatchScheduleRequest.newBuilder()
            .setIdentity(clientOptions.getIdentity())
            .setNamespace(clientOptions.getNamespace())
            .setScheduleId(input.getScheduleId())
            .setPatch(patch)
            .build();
    try {
      genericClient.patchSchedule(request);
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }

  @Override
  public void triggerSchedule(TriggerScheduleInput input) {
    TriggerImmediatelyRequest trigger =
        TriggerImmediatelyRequest.newBuilder().setOverlapPolicy(input.getOverlapPolicy()).build();

    SchedulePatch patch = SchedulePatch.newBuilder().setTriggerImmediately(trigger).build();

    PatchScheduleRequest request =
        PatchScheduleRequest.newBuilder()
            .setIdentity(clientOptions.getIdentity())
            .setNamespace(clientOptions.getNamespace())
            .setScheduleId(input.getScheduleId())
            .setPatch(patch)
            .build();
    try {
      genericClient.patchSchedule(request);
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }

  @Override
  public void unpauseSchedule(UnpauseScheduleInput input) {
    SchedulePatch patch = SchedulePatch.newBuilder().setUnpause(input.getNote()).build();

    PatchScheduleRequest request =
        PatchScheduleRequest.newBuilder()
            .setIdentity(clientOptions.getIdentity())
            .setNamespace(clientOptions.getNamespace())
            .setScheduleId(input.getScheduleId())
            .setPatch(patch)
            .build();
    try {
      genericClient.patchSchedule(request);
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }

  @Override
  public void updateSchedule(UpdateScheduleInput input) {
    ScheduleUpdate schedule =
        input.getUpdater().apply(new ScheduleUpdateInput(input.getDescription()));
    if (schedule == null) {
      return;
    }

    UpdateScheduleRequest request =
        UpdateScheduleRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setIdentity(clientOptions.getIdentity())
            .setScheduleId(input.getDescription().getId())
            .setRequestId(UUID.randomUUID().toString())
            .setSchedule(scheduleRequestHeader.scheduleToProto(schedule.getSchedule()))
            .build();
    try {
      genericClient.updateSchedule(request);
    } catch (Exception e) {
      throw new ScheduleException(e);
    }
  }
}

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.intoPayloadMap;
24
25	import com.google.common.collect.Iterators;
26	import io.grpc.Status;
27	import io.grpc.StatusRuntimeException;
28	import io.temporal.api.common.v1.Memo;
29	import io.temporal.api.schedule.v1.*;
30	import io.temporal.api.workflowservice.v1.*;
31	import io.temporal.client.ListScheduleListDescriptionIterator;
32	import io.temporal.client.schedules.*;
33	import io.temporal.common.interceptors.ScheduleClientCallsInterceptor;
34	import io.temporal.internal.client.external.GenericWorkflowClient;
35	import io.temporal.internal.common.ProtobufTimeUtils;
36	import io.temporal.internal.common.SearchAttributesUtil;
37	import java.util.*;
38	import java.util.stream.StreamSupport;
39	import org.slf4j.Logger;
40	import org.slf4j.LoggerFactory;
41
42	public class RootScheduleClientInvoker implements ScheduleClientCallsInterceptor {
43	private static final Logger log = LoggerFactory.getLogger(RootScheduleClientInvoker.class);	1✔
44
45	private final GenericWorkflowClient genericClient;
46
47	private final ScheduleClientOptions clientOptions;
48
49	private final ScheduleProtoUtil scheduleRequestHeader;
50
51	public RootScheduleClientInvoker(
52	GenericWorkflowClient genericClient, ScheduleClientOptions clientOptions) {	1✔
53	this.genericClient = genericClient;	1✔
54	this.clientOptions = clientOptions;	1✔
55	this.scheduleRequestHeader = new ScheduleProtoUtil(genericClient, clientOptions);	1✔
56	}	1✔
57
58	@Override
59	@SuppressWarnings("deprecation")
60	public void createSchedule(CreateScheduleInput input) {
61
62	CreateScheduleRequest.Builder request =
63	CreateScheduleRequest.newBuilder()	×
64	.setIdentity(clientOptions.getIdentity())	×
65	.setNamespace(clientOptions.getNamespace())	×
66	.setRequestId(UUID.randomUUID().toString())	×
67	.setScheduleId(input.getId())	×
68	.setSchedule(scheduleRequestHeader.scheduleToProto(input.getSchedule()));	×
69
70	if (input.getOptions().getMemo() != null) {	×
71	// TODO we don't have a workflow context here, maybe we need a schedule context?
72	request.setMemo(	×
73	Memo.newBuilder()	×
74	.putAllFields(	×
75	intoPayloadMap(clientOptions.getDataConverter(), input.getOptions().getMemo())));	×
76	}
77
78	if (input.getOptions().getSearchAttributes() != null	×
79	&& !input.getOptions().getSearchAttributes().isEmpty()) {	×
80	if (input.getOptions().getTypedSearchAttributes() != null) {	×
81	throw new IllegalArgumentException(	×
82	"Cannot have search attributes and typed search attributes");
83	}
84	request.setSearchAttributes(	×
85	SearchAttributesUtil.encode(input.getOptions().getSearchAttributes()));	×
86	} else if (input.getOptions().getTypedSearchAttributes() != null) {	×
87	request.setSearchAttributes(	×
88	SearchAttributesUtil.encodeTyped(input.getOptions().getTypedSearchAttributes()));	×
89	}
90
91	if (input.getOptions().isTriggerImmediately()	×
92	\|\| (input.getOptions().getBackfills() != null	×
93	&& input.getOptions().getBackfills().size() > 0)) {	×
94	SchedulePatch.Builder patchBuilder = SchedulePatch.newBuilder();	×
95
96	if (input.getOptions().getBackfills() != null) {	×
97	input.getOptions().getBackfills().stream()	×
98	.forEach(b -> patchBuilder.addBackfillRequest(backfillToProto(b)));	×
99	}
100
101	if (input.getOptions().isTriggerImmediately()) {	×
102	TriggerImmediatelyRequest.Builder triggerRequest = TriggerImmediatelyRequest.newBuilder();	×
103	if (input.getSchedule().getPolicy() != null) {	×
104	triggerRequest.setOverlapPolicy(input.getSchedule().getPolicy().getOverlap());	×
105	}
106	patchBuilder.setTriggerImmediately(triggerRequest.build());	×
107	}
108
109	request.setInitialPatch(patchBuilder.build());	×
110	}
111
112	try {
113	genericClient.createSchedule(request.build());	×
114	} catch (StatusRuntimeException e) {	×
115	if (Status.Code.ALREADY_EXISTS.equals(e.getStatus().getCode())) {	×
116	throw new ScheduleAlreadyRunningException(e);	×
117	} else {
118	throw new ScheduleException(e);	×
119	}
120	} catch (Exception e) {	×
121	throw new ScheduleException(e);	×
122	}	×
123	}	×
124
125	@Override
126	public ListScheduleOutput listSchedules(ListSchedulesInput input) {
127	ListScheduleListDescriptionIterator iterator =	×
128	new ListScheduleListDescriptionIterator(
NEW 129	clientOptions.getNamespace(), input.getQuery(), input.getPageSize(), genericClient);	×
130	iterator.init();	×
131	Iterator<ScheduleListDescription> wrappedIterator =	×
132	Iterators.transform(	×
133	iterator, entry -> scheduleRequestHeader.protoToScheduleListDescription(entry));	×
134
135	final int CHARACTERISTICS =	×
136	Spliterator.ORDERED \| Spliterator.DISTINCT \| Spliterator.NONNULL \| Spliterator.IMMUTABLE;
137	return new ListScheduleOutput(	×
138	StreamSupport.stream(	×
139	Spliterators.spliteratorUnknownSize(wrappedIterator, CHARACTERISTICS), false));	×
140	}
141
142	public BackfillRequest backfillToProto(ScheduleBackfill backfill) {
143	return BackfillRequest.newBuilder()	×
144	.setStartTime(ProtobufTimeUtils.toProtoTimestamp(backfill.getStartAt()))	×
145	.setEndTime(ProtobufTimeUtils.toProtoTimestamp(backfill.getEndAt()))	×
146	.setOverlapPolicy(backfill.getOverlapPolicy())	×
147	.build();	×
148	}
149
150	@Override
151	public void backfillSchedule(BackfillScheduleInput input) {
152	ArrayList<BackfillRequest> backfillRequests =	×
153	new ArrayList<BackfillRequest>(input.getBackfills().size());	×
154	for (ScheduleBackfill backfill : input.getBackfills()) {	×
155	backfillRequests.add(backfillToProto(backfill));	×
156	}	×
157
158	SchedulePatch patch =
159	SchedulePatch.newBuilder().addAllBackfillRequest(backfillRequests).build();	×
160
161	PatchScheduleRequest request =
162	PatchScheduleRequest.newBuilder()	×
163	.setIdentity(clientOptions.getIdentity())	×
164	.setNamespace(clientOptions.getNamespace())	×
165	.setScheduleId(input.getScheduleId())	×
166	.setPatch(patch)	×
167	.build();	×
168	try {
169	genericClient.patchSchedule(request);	×
170	} catch (Exception e) {	×
171	throw new ScheduleException(e);	×
172	}	×
173	}	×
174
175	@Override
176	public void deleteSchedule(DeleteScheduleInput input) {
177	DeleteScheduleRequest request =
178	DeleteScheduleRequest.newBuilder()	×
179	.setIdentity(clientOptions.getIdentity())	×
180	.setNamespace(clientOptions.getNamespace())	×
181	.setScheduleId(input.getScheduleId())	×
182	.build();	×
183	try {
184	genericClient.deleteSchedule(request);	×
185	} catch (Exception e) {	×
186	throw new ScheduleException(e);	×
187	}	×
188	}	×
189
190	@Override
191	public DescribeScheduleOutput describeSchedule(DescribeScheduleInput input) {
192	DescribeScheduleRequest request =
193	DescribeScheduleRequest.newBuilder()	×
194	.setNamespace(clientOptions.getNamespace())	×
195	.setScheduleId(input.getScheduleId())	×
196	.build();	×
197
198	try {
199	DescribeScheduleResponse response = genericClient.describeSchedule(request);	×
200	return new DescribeScheduleOutput(	×
201	new ScheduleDescription(
202	input.getScheduleId(),	×
203	scheduleRequestHeader.protoToScheduleInfo(response.getInfo()),	×
204	scheduleRequestHeader.protoToSchedule(response.getSchedule()),	×
205	Collections.unmodifiableMap(	×
206	SearchAttributesUtil.decode(response.getSearchAttributes())),	×
207	SearchAttributesUtil.decodeTyped(response.getSearchAttributes()),	×
208	response.getMemo().getFieldsMap(),	×
209	clientOptions.getDataConverter()));	×
210	} catch (Exception e) {	×
211	throw new ScheduleException(e);	×
212	}
213	}
214
215	@Override
216	public void pauseSchedule(PauseScheduleInput input) {
217	SchedulePatch patch = SchedulePatch.newBuilder().setPause(input.getNote()).build();	×
218
219	PatchScheduleRequest request =
220	PatchScheduleRequest.newBuilder()	×
221	.setIdentity(clientOptions.getIdentity())	×
222	.setNamespace(clientOptions.getNamespace())	×
223	.setScheduleId(input.getScheduleId())	×
224	.setPatch(patch)	×
225	.build();	×
226	try {
227	genericClient.patchSchedule(request);	×
228	} catch (Exception e) {	×
229	throw new ScheduleException(e);	×
230	}	×
231	}	×
232
233	@Override
234	public void triggerSchedule(TriggerScheduleInput input) {
235	TriggerImmediatelyRequest trigger =
236	TriggerImmediatelyRequest.newBuilder().setOverlapPolicy(input.getOverlapPolicy()).build();	×
237
238	SchedulePatch patch = SchedulePatch.newBuilder().setTriggerImmediately(trigger).build();	×
239
240	PatchScheduleRequest request =
241	PatchScheduleRequest.newBuilder()	×
242	.setIdentity(clientOptions.getIdentity())	×
243	.setNamespace(clientOptions.getNamespace())	×
244	.setScheduleId(input.getScheduleId())	×
245	.setPatch(patch)	×
246	.build();	×
247	try {
248	genericClient.patchSchedule(request);	×
249	} catch (Exception e) {	×
250	throw new ScheduleException(e);	×
251	}	×
252	}	×
253
254	@Override
255	public void unpauseSchedule(UnpauseScheduleInput input) {
256	SchedulePatch patch = SchedulePatch.newBuilder().setUnpause(input.getNote()).build();	×
257
258	PatchScheduleRequest request =
259	PatchScheduleRequest.newBuilder()	×
260	.setIdentity(clientOptions.getIdentity())	×
261	.setNamespace(clientOptions.getNamespace())	×
262	.setScheduleId(input.getScheduleId())	×
263	.setPatch(patch)	×
264	.build();	×
265	try {
266	genericClient.patchSchedule(request);	×
267	} catch (Exception e) {	×
268	throw new ScheduleException(e);	×
269	}	×
270	}	×
271
272	@Override
273	public void updateSchedule(UpdateScheduleInput input) {
274	ScheduleUpdate schedule =	×
275	input.getUpdater().apply(new ScheduleUpdateInput(input.getDescription()));	×
276	if (schedule == null) {	×
277	return;	×
278	}
279
280	UpdateScheduleRequest request =
281	UpdateScheduleRequest.newBuilder()	×
282	.setNamespace(clientOptions.getNamespace())	×
283	.setIdentity(clientOptions.getIdentity())	×
284	.setScheduleId(input.getDescription().getId())	×
285	.setRequestId(UUID.randomUUID().toString())	×
286	.setSchedule(scheduleRequestHeader.scheduleToProto(schedule.getSchedule()))	×
287	.build();	×
288	try {
289	genericClient.updateSchedule(request);	×
290	} catch (Exception e) {	×
291	throw new ScheduleException(e);	×
292	}	×
293	}	×
294	}

temporalio / sdk-java / #292

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