#174

pending completion

Build # #174

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Add schedules API (#1776)

Add schedules API

Run Details

1143 of 1143 new or added lines in 35 files covered. (100.0%)

18101 of 23284 relevant lines covered (77.74%)

0.78 hits per line

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

0.0

/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.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 com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import io.temporal.api.common.v1.*;
import io.temporal.api.schedule.v1.*;
import io.temporal.api.schedule.v1.Schedule;
import io.temporal.api.schedule.v1.ScheduleAction;
import io.temporal.api.schedule.v1.ScheduleActionResult;
import io.temporal.api.schedule.v1.ScheduleInfo;
import io.temporal.api.schedule.v1.ScheduleSpec;
import io.temporal.api.schedule.v1.ScheduleState;
import io.temporal.api.taskqueue.v1.TaskQueue;
import io.temporal.api.workflow.v1.NewWorkflowExecutionInfo;
import io.temporal.client.WorkflowOptions;
import io.temporal.client.schedules.*;
import io.temporal.common.context.ContextPropagator;
import io.temporal.common.converter.EncodedValues;
import io.temporal.internal.client.external.GenericWorkflowClient;
import io.temporal.internal.common.ProtobufTimeUtils;
import io.temporal.internal.common.RetryOptionsUtils;
import io.temporal.internal.common.SearchAttributesUtil;
import java.time.Instant;
import java.util.*;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

public class ScheduleProtoUtil {

  private final GenericWorkflowClient genericClient;
  private final ScheduleClientOptions clientOptions;

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

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

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

  public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction action) {
    if (action instanceof ScheduleActionStartWorkflow) {
      ScheduleActionStartWorkflow startWorkflowAction = (ScheduleActionStartWorkflow) action;

      WorkflowOptions wfOptions = startWorkflowAction.getOptions();
      // Disallow some options
      if (wfOptions.getWorkflowIdReusePolicy() != null) {
        throw new IllegalArgumentException(
            "ID reuse policy cannot change from default for scheduled workflow");
      }
      if (wfOptions.getCronSchedule() != null) {
        throw new IllegalArgumentException("Cron schedule cannot be set on scheduled workflow");
      }
      // Validate required options
      if (wfOptions.getWorkflowId() == null || wfOptions.getWorkflowId().isEmpty()) {
        throw new IllegalArgumentException("ID required on workflow action");
      }
      if (wfOptions.getTaskQueue() == null || wfOptions.getTaskQueue().isEmpty()) {
        throw new IllegalArgumentException("Task queue required on workflow action");
      }

      NewWorkflowExecutionInfo.Builder workflowRequest =
          NewWorkflowExecutionInfo.newBuilder()
              .setWorkflowId(wfOptions.getWorkflowId())
              .setWorkflowType(
                  WorkflowType.newBuilder().setName(startWorkflowAction.getWorkflowType()).build())
              .setWorkflowRunTimeout(
                  ProtobufTimeUtils.toProtoDuration(wfOptions.getWorkflowRunTimeout()))
              .setWorkflowExecutionTimeout(
                  ProtobufTimeUtils.toProtoDuration(wfOptions.getWorkflowExecutionTimeout()))
              .setWorkflowTaskTimeout(
                  ProtobufTimeUtils.toProtoDuration(wfOptions.getWorkflowTaskTimeout()))
              .setTaskQueue(TaskQueue.newBuilder().setName(wfOptions.getTaskQueue()).build());

      startWorkflowAction.getArguments().setDataConverter(clientOptions.getDataConverter());
      Optional<Payloads> inputArgs = startWorkflowAction.getArguments().toPayloads();
      if (inputArgs.isPresent()) {
        workflowRequest.setInput(inputArgs.get());
      }

      if (startWorkflowAction.getOptions().getMemo() != null) {
        Map<String, Payload> memo = new HashMap<>();
        for (Map.Entry<String, Object> item :
            startWorkflowAction.getOptions().getMemo().entrySet()) {
          if (item.getValue() instanceof EncodedValues) {
            memo.put(
                item.getKey(), ((EncodedValues) item.getValue()).toPayloads().get().getPayloads(0));
          } else {
            memo.put(
                item.getKey(), clientOptions.getDataConverter().toPayload(item.getValue()).get());
          }
        }
        workflowRequest.setMemo(Memo.newBuilder().putAllFields(memo).build());
      }

      if (wfOptions.getSearchAttributes() != null && !wfOptions.getSearchAttributes().isEmpty()) {
        workflowRequest.setSearchAttributes(
            SearchAttributesUtil.encode(wfOptions.getSearchAttributes()));
      }

      Header grpcHeader =
          toHeaderGrpc(
              startWorkflowAction.getHeader(),
              extractContextsAndConvertToBytes(wfOptions.getContextPropagators()));
      workflowRequest.setHeader(grpcHeader);

      return ScheduleAction.newBuilder().setStartWorkflow(workflowRequest.build()).build();
    }
    throw new IllegalArgumentException("Unsupported action " + action.getClass());
  }

  public SchedulePolicies policyToProto(SchedulePolicy policy) {
    return SchedulePolicies.newBuilder()
        .setCatchupWindow(ProtobufTimeUtils.toProtoDuration(policy.getCatchupWindow()))
        .setPauseOnFailure(policy.isPauseOnFailure())
        .setOverlapPolicy(policy.getOverlap())
        .build();
  }

  public List<Range> scheduleRangeToProto(List<ScheduleRange> scheduleRanges) {
    ArrayList<Range> ranges = new ArrayList<Range>(scheduleRanges.size());
    for (ScheduleRange scheduleRange : scheduleRanges) {
      ranges.add(
          Range.newBuilder()
              .setStart(scheduleRange.getStart())
              .setEnd(scheduleRange.getEnd())
              .setStep(scheduleRange.getStep())
              .build());
    }
    return ranges;
  }

  public ScheduleSpec specToProto(io.temporal.client.schedules.ScheduleSpec spec) {
    ScheduleSpec.Builder builder = ScheduleSpec.newBuilder();

    if (spec.getTimeZoneName() != null && !spec.getTimeZoneName().isEmpty()) {
      builder.setTimezoneName(spec.getTimeZoneName());
    }

    if (spec.getJitter() != null) {
      builder.setJitter(ProtobufTimeUtils.toProtoDuration(spec.getJitter()));
    }

    if (spec.getStartAt() != null) {
      builder.setStartTime(ProtobufTimeUtils.toProtoTimestamp(spec.getStartAt()));
    }

    if (spec.getEndAt() != null) {
      builder.setEndTime(ProtobufTimeUtils.toProtoTimestamp(spec.getEndAt()));
    }

    if (spec.getCalendars() != null && !spec.getCalendars().isEmpty()) {
      for (ScheduleCalendarSpec calendarSpec : spec.getCalendars()) {
        builder.addStructuredCalendar(
            StructuredCalendarSpec.newBuilder()
                .addAllSecond(this.scheduleRangeToProto(calendarSpec.getSeconds()))
                .addAllMinute(this.scheduleRangeToProto(calendarSpec.getMinutes()))
                .addAllHour(this.scheduleRangeToProto(calendarSpec.getHour()))
                .addAllDayOfMonth(this.scheduleRangeToProto(calendarSpec.getDayOfMonth()))
                .addAllMonth(this.scheduleRangeToProto(calendarSpec.getMonth()))
                .addAllYear(this.scheduleRangeToProto(calendarSpec.getYear()))
                .addAllDayOfWeek(this.scheduleRangeToProto(calendarSpec.getDayOfWeek()))
                .setComment(calendarSpec.getComment())
                .build());
      }
    }

    if (spec.getIntervals() != null && !spec.getIntervals().isEmpty()) {
      for (ScheduleIntervalSpec intervalSpec : spec.getIntervals()) {
        builder.addInterval(
            IntervalSpec.newBuilder()
                .setInterval(ProtobufTimeUtils.toProtoDuration(intervalSpec.getEvery()))
                .setPhase(ProtobufTimeUtils.toProtoDuration(intervalSpec.getOffset()))
                .build());
      }
    }

    if (spec.getCronExpressions() != null && !spec.getCronExpressions().isEmpty()) {
      builder.addAllCronString(spec.getCronExpressions());
    }

    if (spec.getSkip() != null && !spec.getSkip().isEmpty()) {
      for (ScheduleCalendarSpec calendarSpec : spec.getSkip()) {
        builder.addExcludeStructuredCalendar(
            StructuredCalendarSpec.newBuilder()
                .addAllSecond(this.scheduleRangeToProto(calendarSpec.getSeconds()))
                .addAllMinute(this.scheduleRangeToProto(calendarSpec.getMinutes()))
                .addAllHour(this.scheduleRangeToProto(calendarSpec.getHour()))
                .addAllDayOfMonth(this.scheduleRangeToProto(calendarSpec.getDayOfMonth()))
                .addAllMonth(this.scheduleRangeToProto(calendarSpec.getMonth()))
                .addAllYear(this.scheduleRangeToProto(calendarSpec.getYear()))
                .addAllDayOfWeek(this.scheduleRangeToProto(calendarSpec.getDayOfWeek()))
                .setComment(calendarSpec.getComment())
                .build());
      }
    }

    return builder.build();
  }

  public @Nonnull Schedule scheduleToProto(
      @Nonnull io.temporal.client.schedules.Schedule schedule) {
    Preconditions.checkNotNull(schedule);

    Schedule.Builder scheduleBuilder =
        Schedule.newBuilder()
            .setAction(this.actionToProto(schedule.getAction()))
            .setSpec(this.specToProto(schedule.getSpec()));

    if (schedule.getPolicy() != null) {
      scheduleBuilder.setPolicies(this.policyToProto(schedule.getPolicy()));
    }

    if (schedule.getState() != null) {
      scheduleBuilder.setState(this.stateToProto(schedule.getState()));
    }

    return scheduleBuilder.build();
  }

  public ScheduleState stateToProto(io.temporal.client.schedules.ScheduleState state) {
    ScheduleState.Builder stateBuilder =
        ScheduleState.newBuilder()
            .setLimitedActions(state.isLimitedAction())
            .setRemainingActions(state.getRemainingActions())
            .setPaused(state.isPaused());
    if (state.getNote() != null) {
      stateBuilder.setNotes(state.getNote());
    }
    return stateBuilder.build();
  }

  public ScheduleRange protoToScheduleRange(Range protoRange) {
    return new ScheduleRange(protoRange.getStart(), protoRange.getEnd(), protoRange.getStep());
  }

  public ScheduleIntervalSpec protoToScheduleInterval(IntervalSpec protoInterval) {
    return new ScheduleIntervalSpec(
        ProtobufTimeUtils.toJavaDuration(protoInterval.getInterval()),
        ProtobufTimeUtils.toJavaDuration(protoInterval.getPhase()));
  }

  public List<ScheduleRange> protoToScheduleRanges(List<Range> protoRanges) {
    return protoRanges.stream().map(t -> this.protoToScheduleRange(t)).collect(Collectors.toList());
  }

  public ScheduleCalendarSpec protoToScheduleCalendar(StructuredCalendarSpec protoSpec) {
    ScheduleCalendarSpec.Builder calendarBuilder =
        ScheduleCalendarSpec.newBuilder()
            .setComment(protoSpec.getComment())
            .setSeconds(protoToScheduleRanges(protoSpec.getSecondList()))
            .setMinutes(protoToScheduleRanges(protoSpec.getMinuteList()))
            .setHour(protoToScheduleRanges(protoSpec.getHourList()))
            .setDayOfMonth(protoToScheduleRanges(protoSpec.getDayOfMonthList()))
            .setMonth(protoToScheduleRanges(protoSpec.getMonthList()))
            .setYear(protoToScheduleRanges(protoSpec.getYearList()))
            .setDayOfWeek(protoToScheduleRanges(protoSpec.getDayOfWeekList()));

    return calendarBuilder.build();
  }

  @Nonnull
  public io.temporal.client.schedules.ScheduleSpec protoToScheduleSpec(
      @Nonnull ScheduleSpec scheduleSpec) {
    Objects.requireNonNull(scheduleSpec);
    io.temporal.client.schedules.ScheduleSpec.Builder specBuilder =
        io.temporal.client.schedules.ScheduleSpec.newBuilder()
            .setTimeZoneName(
                scheduleSpec.getTimezoneName() == null ? "" : scheduleSpec.getTimezoneName());

    if (scheduleSpec.hasJitter()) {
      specBuilder.setJitter(ProtobufTimeUtils.toJavaDuration(scheduleSpec.getJitter()));
    }

    if (scheduleSpec.hasStartTime()) {
      specBuilder.setStartAt(ProtobufTimeUtils.toJavaInstant(scheduleSpec.getStartTime()));
    }

    if (scheduleSpec.hasEndTime()) {
      specBuilder.setEndAt(ProtobufTimeUtils.toJavaInstant(scheduleSpec.getEndTime()));
    }

    specBuilder.setCalendars(
        scheduleSpec.getStructuredCalendarList().stream()
            .map(c -> this.protoToScheduleCalendar(c))
            .collect(Collectors.toList()));

    specBuilder.setCronExpressions(scheduleSpec.getCronStringList());

    specBuilder.setIntervals(
        scheduleSpec.getIntervalList().stream()
            .map(i -> this.protoToScheduleInterval(i))
            .collect(Collectors.toList()));

    specBuilder.setSkip(
        scheduleSpec.getExcludeStructuredCalendarList().stream()
            .map(c -> this.protoToScheduleCalendar(c))
            .collect(Collectors.toList()));

    return specBuilder.build();
  }

  public List<io.temporal.client.schedules.ScheduleActionResult> protoToActionResults(
      List<ScheduleActionResult> results) {
    return results.stream()
        .map(
            a ->
                new io.temporal.client.schedules.ScheduleActionResult(
                    ProtobufTimeUtils.toJavaInstant(a.getScheduleTime()),
                    ProtobufTimeUtils.toJavaInstant(a.getActualTime()),
                    new ScheduleActionExecutionStartWorkflow(
                        a.getStartWorkflowResult().getWorkflowId(),
                        a.getStartWorkflowResult().getRunId())))
        .collect(Collectors.toList());
  }

  public ScheduleListDescription protoToScheduleListDescription(ScheduleListEntry entry) {
    List<Instant> nextActionTimes =
        entry.getInfo().getFutureActionTimesList().stream()
            .map(ProtobufTimeUtils::toJavaInstant)
            .collect(Collectors.toList());

    io.temporal.client.schedules.ScheduleListInfo info =
        new io.temporal.client.schedules.ScheduleListInfo(
            this.protoToActionResults(entry.getInfo().getRecentActionsList()), nextActionTimes);

    ScheduleListActionStartWorkflow action =
        new ScheduleListActionStartWorkflow(entry.getInfo().getWorkflowType().getName());

    ScheduleListState state =
        new ScheduleListState(entry.getInfo().getNotes(), entry.getInfo().getPaused());

    io.temporal.client.schedules.ScheduleSpec spec = protoToScheduleSpec(entry.getInfo().getSpec());

    return new ScheduleListDescription(
        entry.getScheduleId(),
        new ScheduleListSchedule(action, spec, state),
        info,
        entry.getMemo().getFieldsMap(),
        this.clientOptions.getDataConverter(),
        Collections.unmodifiableMap(SearchAttributesUtil.decode(entry.getSearchAttributes())));
  }

  @Nonnull
  public io.temporal.client.schedules.ScheduleAction protoToAction(@Nonnull ScheduleAction action) {
    Objects.requireNonNull(action);
    if (action.hasStartWorkflow()) {
      NewWorkflowExecutionInfo startWfAction = action.getStartWorkflow();
      ScheduleActionStartWorkflow.Builder builder = ScheduleActionStartWorkflow.newBuilder();
      builder.setWorkflowType(startWfAction.getWorkflowType().getName());

      builder.setRawArguments(
          new EncodedValues(
              Optional.of(startWfAction.getInput()), clientOptions.getDataConverter()));

      WorkflowOptions.Builder wfOptionsBuilder = WorkflowOptions.newBuilder();
      // set required options
      wfOptionsBuilder.setWorkflowId(startWfAction.getWorkflowId());
      wfOptionsBuilder.setTaskQueue(startWfAction.getTaskQueue().getName());
      // set timeouts
      wfOptionsBuilder.setWorkflowExecutionTimeout(
          ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowExecutionTimeout()));

      wfOptionsBuilder.setWorkflowRunTimeout(
          ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowRunTimeout()));

      wfOptionsBuilder.setWorkflowTaskTimeout(
          ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowTaskTimeout()));

      if (startWfAction.getRetryPolicy() != null) {
        wfOptionsBuilder.setRetryOptions(
            RetryOptionsUtils.toRetryOptions(startWfAction.getRetryPolicy()));
      }

      if (startWfAction.hasMemo()) {
        Map<String, Object> memos = new HashMap<>();
        for (Map.Entry<String, Payload> memo : startWfAction.getMemo().getFieldsMap().entrySet()) {
          EncodedValues encodedMemo =
              new EncodedValues(
                  Optional.of(Payloads.newBuilder().addPayloads(memo.getValue()).build()),
                  clientOptions.getDataConverter());
          memos.put(memo.getKey(), encodedMemo);
        }
        wfOptionsBuilder.setMemo(memos);
      }

      if (startWfAction.hasSearchAttributes()) {
        wfOptionsBuilder.setSearchAttributes(
            Collections.unmodifiableMap(
                SearchAttributesUtil.decode(startWfAction.getSearchAttributes())));
      }

      builder.setOptions(wfOptionsBuilder.build());
      return builder.build();
    }
    throw new IllegalArgumentException("Unsupported action " + action.getActionCase());
  }

  @Nullable
  public SchedulePolicy protoToPolicy(@Nullable SchedulePolicies policy) {
    if (policy == null) {
      return null;
    }
    SchedulePolicy.Builder policyBuilder =
        SchedulePolicy.newBuilder()
            .setPauseOnFailure(policy.getPauseOnFailure())
            .setOverlap(policy.getOverlapPolicy());
    if (policy.hasCatchupWindow()) {
      policyBuilder.setCatchupWindow(ProtobufTimeUtils.toJavaDuration(policy.getCatchupWindow()));
    }
    return policyBuilder.build();
  }

  @Nullable
  public io.temporal.client.schedules.ScheduleState protoToScheduleState(
      @Nullable ScheduleState state) {
    if (state == null) {
      return null;
    }
    return io.temporal.client.schedules.ScheduleState.newBuilder()
        .setNote(state.getNotes())
        .setPaused(state.getPaused())
        .setRemainingActions(state.getRemainingActions())
        .setLimitedAction(state.getLimitedActions())
        .build();
  }

  public io.temporal.client.schedules.Schedule protoToSchedule(Schedule schedule) {
    return io.temporal.client.schedules.Schedule.newBuilder()
        .setAction(protoToAction(schedule.getAction()))
        .setSpec(protoToScheduleSpec(schedule.getSpec()))
        .setPolicy(protoToPolicy(schedule.getPolicies()))
        .setState(protoToScheduleState(schedule.getState()))
        .build();
  }

  public io.temporal.client.schedules.ScheduleInfo protoToScheduleInfo(ScheduleInfo info) {
    return new io.temporal.client.schedules.ScheduleInfo(
        info.getActionCount(),
        info.getMissedCatchupWindow(),
        info.getOverlapSkipped(),
        info.getRunningWorkflowsList().stream()
            .map(wf -> new ScheduleActionExecutionStartWorkflow(wf.getWorkflowId(), wf.getRunId()))
            .collect(Collectors.toList()),
        this.protoToActionResults(info.getRecentActionsList()),
        info.getFutureActionTimesList().stream()
            .map(t -> ProtobufTimeUtils.toJavaInstant(t))
            .collect(Collectors.toList()),
        info.hasCreateTime() ? ProtobufTimeUtils.toJavaInstant(info.getCreateTime()) : null,
        info.hasUpdateTime() ? ProtobufTimeUtils.toJavaInstant(info.getUpdateTime()) : null);
  }
}

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
25	import com.google.common.base.MoreObjects;
26	import com.google.common.base.Preconditions;
27	import io.temporal.api.common.v1.*;
28	import io.temporal.api.schedule.v1.*;
29	import io.temporal.api.schedule.v1.Schedule;
30	import io.temporal.api.schedule.v1.ScheduleAction;
31	import io.temporal.api.schedule.v1.ScheduleActionResult;
32	import io.temporal.api.schedule.v1.ScheduleInfo;
33	import io.temporal.api.schedule.v1.ScheduleSpec;
34	import io.temporal.api.schedule.v1.ScheduleState;
35	import io.temporal.api.taskqueue.v1.TaskQueue;
36	import io.temporal.api.workflow.v1.NewWorkflowExecutionInfo;
37	import io.temporal.client.WorkflowOptions;
38	import io.temporal.client.schedules.*;
39	import io.temporal.common.context.ContextPropagator;
40	import io.temporal.common.converter.EncodedValues;
41	import io.temporal.internal.client.external.GenericWorkflowClient;
42	import io.temporal.internal.common.ProtobufTimeUtils;
43	import io.temporal.internal.common.RetryOptionsUtils;
44	import io.temporal.internal.common.SearchAttributesUtil;
45	import java.time.Instant;
46	import java.util.*;
47	import java.util.stream.Collectors;
48	import javax.annotation.Nonnull;
49	import javax.annotation.Nullable;
50
51	public class ScheduleProtoUtil {
52
53	private final GenericWorkflowClient genericClient;
54	private final ScheduleClientOptions clientOptions;
55
56	public ScheduleProtoUtil(
57	GenericWorkflowClient genericClient, ScheduleClientOptions clientOptions) {	×
58	this.genericClient = genericClient;	×
59	this.clientOptions = clientOptions;	×
60	}	×
61
62	private io.temporal.common.interceptors.Header extractContextsAndConvertToBytes(
63	List<ContextPropagator> scheduleOptionsContextPropagators) {
64	List<ContextPropagator> scheduleClientContextPropagators =	×
65	clientOptions.getContextPropagators();	×
66	if ((scheduleClientContextPropagators.isEmpty() && scheduleOptionsContextPropagators == null)	×
67	\|\| (scheduleOptionsContextPropagators != null
68	&& scheduleOptionsContextPropagators.isEmpty())) {	×
69	return null;	×
70	}
71
72	List<ContextPropagator> listToUse =	×
73	MoreObjects.firstNonNull(	×
74	scheduleOptionsContextPropagators, scheduleClientContextPropagators);
75	Map<String, Payload> result = new HashMap<>();	×
76	for (ContextPropagator propagator : listToUse) {	×
77	result.putAll(propagator.serializeContext(propagator.getCurrentContext()));	×
78	}	×
79	return new io.temporal.common.interceptors.Header(result);	×
80	}
81
82	public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction action) {
83	if (action instanceof ScheduleActionStartWorkflow) {	×
84	ScheduleActionStartWorkflow startWorkflowAction = (ScheduleActionStartWorkflow) action;	×
85
86	WorkflowOptions wfOptions = startWorkflowAction.getOptions();	×
87	// Disallow some options
88	if (wfOptions.getWorkflowIdReusePolicy() != null) {	×
89	throw new IllegalArgumentException(	×
90	"ID reuse policy cannot change from default for scheduled workflow");
91	}
92	if (wfOptions.getCronSchedule() != null) {	×
93	throw new IllegalArgumentException("Cron schedule cannot be set on scheduled workflow");	×
94	}
95	// Validate required options
96	if (wfOptions.getWorkflowId() == null \|\| wfOptions.getWorkflowId().isEmpty()) {	×
97	throw new IllegalArgumentException("ID required on workflow action");	×
98	}
99	if (wfOptions.getTaskQueue() == null \|\| wfOptions.getTaskQueue().isEmpty()) {	×
100	throw new IllegalArgumentException("Task queue required on workflow action");	×
101	}
102
103	NewWorkflowExecutionInfo.Builder workflowRequest =
104	NewWorkflowExecutionInfo.newBuilder()	×
105	.setWorkflowId(wfOptions.getWorkflowId())	×
106	.setWorkflowType(	×
107	WorkflowType.newBuilder().setName(startWorkflowAction.getWorkflowType()).build())	×
108	.setWorkflowRunTimeout(	×
109	ProtobufTimeUtils.toProtoDuration(wfOptions.getWorkflowRunTimeout()))	×
110	.setWorkflowExecutionTimeout(	×
111	ProtobufTimeUtils.toProtoDuration(wfOptions.getWorkflowExecutionTimeout()))	×
112	.setWorkflowTaskTimeout(	×
113	ProtobufTimeUtils.toProtoDuration(wfOptions.getWorkflowTaskTimeout()))	×
114	.setTaskQueue(TaskQueue.newBuilder().setName(wfOptions.getTaskQueue()).build());	×
115
116	startWorkflowAction.getArguments().setDataConverter(clientOptions.getDataConverter());	×
117	Optional<Payloads> inputArgs = startWorkflowAction.getArguments().toPayloads();	×
118	if (inputArgs.isPresent()) {	×
119	workflowRequest.setInput(inputArgs.get());	×
120	}
121
122	if (startWorkflowAction.getOptions().getMemo() != null) {	×
123	Map<String, Payload> memo = new HashMap<>();	×
124	for (Map.Entry<String, Object> item :
125	startWorkflowAction.getOptions().getMemo().entrySet()) {	×
126	if (item.getValue() instanceof EncodedValues) {	×
127	memo.put(	×
128	item.getKey(), ((EncodedValues) item.getValue()).toPayloads().get().getPayloads(0));	×
129	} else {
130	memo.put(	×
131	item.getKey(), clientOptions.getDataConverter().toPayload(item.getValue()).get());	×
132	}
133	}	×
134	workflowRequest.setMemo(Memo.newBuilder().putAllFields(memo).build());	×
135	}
136
137	if (wfOptions.getSearchAttributes() != null && !wfOptions.getSearchAttributes().isEmpty()) {	×
138	workflowRequest.setSearchAttributes(	×
139	SearchAttributesUtil.encode(wfOptions.getSearchAttributes()));	×
140	}
141
142	Header grpcHeader =	×
143	toHeaderGrpc(	×
144	startWorkflowAction.getHeader(),	×
145	extractContextsAndConvertToBytes(wfOptions.getContextPropagators()));	×
146	workflowRequest.setHeader(grpcHeader);	×
147
148	return ScheduleAction.newBuilder().setStartWorkflow(workflowRequest.build()).build();	×
149	}
150	throw new IllegalArgumentException("Unsupported action " + action.getClass());	×
151	}
152
153	public SchedulePolicies policyToProto(SchedulePolicy policy) {
154	return SchedulePolicies.newBuilder()	×
155	.setCatchupWindow(ProtobufTimeUtils.toProtoDuration(policy.getCatchupWindow()))	×
156	.setPauseOnFailure(policy.isPauseOnFailure())	×
157	.setOverlapPolicy(policy.getOverlap())	×
158	.build();	×
159	}
160
161	public List<Range> scheduleRangeToProto(List<ScheduleRange> scheduleRanges) {
162	ArrayList<Range> ranges = new ArrayList<Range>(scheduleRanges.size());	×
163	for (ScheduleRange scheduleRange : scheduleRanges) {	×
164	ranges.add(	×
165	Range.newBuilder()	×
166	.setStart(scheduleRange.getStart())	×
167	.setEnd(scheduleRange.getEnd())	×
168	.setStep(scheduleRange.getStep())	×
169	.build());	×
170	}	×
171	return ranges;	×
172	}
173
174	public ScheduleSpec specToProto(io.temporal.client.schedules.ScheduleSpec spec) {
175	ScheduleSpec.Builder builder = ScheduleSpec.newBuilder();	×
176
177	if (spec.getTimeZoneName() != null && !spec.getTimeZoneName().isEmpty()) {	×
178	builder.setTimezoneName(spec.getTimeZoneName());	×
179	}
180
181	if (spec.getJitter() != null) {	×
182	builder.setJitter(ProtobufTimeUtils.toProtoDuration(spec.getJitter()));	×
183	}
184
185	if (spec.getStartAt() != null) {	×
186	builder.setStartTime(ProtobufTimeUtils.toProtoTimestamp(spec.getStartAt()));	×
187	}
188
189	if (spec.getEndAt() != null) {	×
190	builder.setEndTime(ProtobufTimeUtils.toProtoTimestamp(spec.getEndAt()));	×
191	}
192
193	if (spec.getCalendars() != null && !spec.getCalendars().isEmpty()) {	×
194	for (ScheduleCalendarSpec calendarSpec : spec.getCalendars()) {	×
195	builder.addStructuredCalendar(	×
196	StructuredCalendarSpec.newBuilder()	×
197	.addAllSecond(this.scheduleRangeToProto(calendarSpec.getSeconds()))	×
198	.addAllMinute(this.scheduleRangeToProto(calendarSpec.getMinutes()))	×
199	.addAllHour(this.scheduleRangeToProto(calendarSpec.getHour()))	×
200	.addAllDayOfMonth(this.scheduleRangeToProto(calendarSpec.getDayOfMonth()))	×
201	.addAllMonth(this.scheduleRangeToProto(calendarSpec.getMonth()))	×
202	.addAllYear(this.scheduleRangeToProto(calendarSpec.getYear()))	×
203	.addAllDayOfWeek(this.scheduleRangeToProto(calendarSpec.getDayOfWeek()))	×
204	.setComment(calendarSpec.getComment())	×
205	.build());	×
206	}	×
207	}
208
209	if (spec.getIntervals() != null && !spec.getIntervals().isEmpty()) {	×
210	for (ScheduleIntervalSpec intervalSpec : spec.getIntervals()) {	×
211	builder.addInterval(	×
212	IntervalSpec.newBuilder()	×
213	.setInterval(ProtobufTimeUtils.toProtoDuration(intervalSpec.getEvery()))	×
214	.setPhase(ProtobufTimeUtils.toProtoDuration(intervalSpec.getOffset()))	×
215	.build());	×
216	}	×
217	}
218
219	if (spec.getCronExpressions() != null && !spec.getCronExpressions().isEmpty()) {	×
220	builder.addAllCronString(spec.getCronExpressions());	×
221	}
222
223	if (spec.getSkip() != null && !spec.getSkip().isEmpty()) {	×
224	for (ScheduleCalendarSpec calendarSpec : spec.getSkip()) {	×
225	builder.addExcludeStructuredCalendar(	×
226	StructuredCalendarSpec.newBuilder()	×
227	.addAllSecond(this.scheduleRangeToProto(calendarSpec.getSeconds()))	×
228	.addAllMinute(this.scheduleRangeToProto(calendarSpec.getMinutes()))	×
229	.addAllHour(this.scheduleRangeToProto(calendarSpec.getHour()))	×
230	.addAllDayOfMonth(this.scheduleRangeToProto(calendarSpec.getDayOfMonth()))	×
231	.addAllMonth(this.scheduleRangeToProto(calendarSpec.getMonth()))	×
232	.addAllYear(this.scheduleRangeToProto(calendarSpec.getYear()))	×
233	.addAllDayOfWeek(this.scheduleRangeToProto(calendarSpec.getDayOfWeek()))	×
234	.setComment(calendarSpec.getComment())	×
235	.build());	×
236	}	×
237	}
238
239	return builder.build();	×
240	}
241
242	public @Nonnull Schedule scheduleToProto(
243	@Nonnull io.temporal.client.schedules.Schedule schedule) {
244	Preconditions.checkNotNull(schedule);	×
245
246	Schedule.Builder scheduleBuilder =
247	Schedule.newBuilder()	×
248	.setAction(this.actionToProto(schedule.getAction()))	×
249	.setSpec(this.specToProto(schedule.getSpec()));	×
250
251	if (schedule.getPolicy() != null) {	×
252	scheduleBuilder.setPolicies(this.policyToProto(schedule.getPolicy()));	×
253	}
254
255	if (schedule.getState() != null) {	×
256	scheduleBuilder.setState(this.stateToProto(schedule.getState()));	×
257	}
258
259	return scheduleBuilder.build();	×
260	}
261
262	public ScheduleState stateToProto(io.temporal.client.schedules.ScheduleState state) {
263	ScheduleState.Builder stateBuilder =
264	ScheduleState.newBuilder()	×
265	.setLimitedActions(state.isLimitedAction())	×
266	.setRemainingActions(state.getRemainingActions())	×
267	.setPaused(state.isPaused());	×
268	if (state.getNote() != null) {	×
269	stateBuilder.setNotes(state.getNote());	×
270	}
271	return stateBuilder.build();	×
272	}
273
274	public ScheduleRange protoToScheduleRange(Range protoRange) {
275	return new ScheduleRange(protoRange.getStart(), protoRange.getEnd(), protoRange.getStep());	×
276	}
277
278	public ScheduleIntervalSpec protoToScheduleInterval(IntervalSpec protoInterval) {
279	return new ScheduleIntervalSpec(	×
280	ProtobufTimeUtils.toJavaDuration(protoInterval.getInterval()),	×
281	ProtobufTimeUtils.toJavaDuration(protoInterval.getPhase()));	×
282	}
283
284	public List<ScheduleRange> protoToScheduleRanges(List<Range> protoRanges) {
285	return protoRanges.stream().map(t -> this.protoToScheduleRange(t)).collect(Collectors.toList());	×
286	}
287
288	public ScheduleCalendarSpec protoToScheduleCalendar(StructuredCalendarSpec protoSpec) {
289	ScheduleCalendarSpec.Builder calendarBuilder =
290	ScheduleCalendarSpec.newBuilder()	×
291	.setComment(protoSpec.getComment())	×
292	.setSeconds(protoToScheduleRanges(protoSpec.getSecondList()))	×
293	.setMinutes(protoToScheduleRanges(protoSpec.getMinuteList()))	×
294	.setHour(protoToScheduleRanges(protoSpec.getHourList()))	×
295	.setDayOfMonth(protoToScheduleRanges(protoSpec.getDayOfMonthList()))	×
296	.setMonth(protoToScheduleRanges(protoSpec.getMonthList()))	×
297	.setYear(protoToScheduleRanges(protoSpec.getYearList()))	×
298	.setDayOfWeek(protoToScheduleRanges(protoSpec.getDayOfWeekList()));	×
299
300	return calendarBuilder.build();	×
301	}
302
303	@Nonnull
304	public io.temporal.client.schedules.ScheduleSpec protoToScheduleSpec(
305	@Nonnull ScheduleSpec scheduleSpec) {
306	Objects.requireNonNull(scheduleSpec);	×
307	io.temporal.client.schedules.ScheduleSpec.Builder specBuilder =
308	io.temporal.client.schedules.ScheduleSpec.newBuilder()	×
309	.setTimeZoneName(	×
310	scheduleSpec.getTimezoneName() == null ? "" : scheduleSpec.getTimezoneName());	×
311
312	if (scheduleSpec.hasJitter()) {	×
313	specBuilder.setJitter(ProtobufTimeUtils.toJavaDuration(scheduleSpec.getJitter()));	×
314	}
315
316	if (scheduleSpec.hasStartTime()) {	×
317	specBuilder.setStartAt(ProtobufTimeUtils.toJavaInstant(scheduleSpec.getStartTime()));	×
318	}
319
320	if (scheduleSpec.hasEndTime()) {	×
321	specBuilder.setEndAt(ProtobufTimeUtils.toJavaInstant(scheduleSpec.getEndTime()));	×
322	}
323
324	specBuilder.setCalendars(	×
325	scheduleSpec.getStructuredCalendarList().stream()	×
326	.map(c -> this.protoToScheduleCalendar(c))	×
327	.collect(Collectors.toList()));	×
328
329	specBuilder.setCronExpressions(scheduleSpec.getCronStringList());	×
330
331	specBuilder.setIntervals(	×
332	scheduleSpec.getIntervalList().stream()	×
333	.map(i -> this.protoToScheduleInterval(i))	×
334	.collect(Collectors.toList()));	×
335
336	specBuilder.setSkip(	×
337	scheduleSpec.getExcludeStructuredCalendarList().stream()	×
338	.map(c -> this.protoToScheduleCalendar(c))	×
339	.collect(Collectors.toList()));	×
340
341	return specBuilder.build();	×
342	}
343
344	public List<io.temporal.client.schedules.ScheduleActionResult> protoToActionResults(
345	List<ScheduleActionResult> results) {
346	return results.stream()	×
347	.map(	×
348	a ->
349	new io.temporal.client.schedules.ScheduleActionResult(	×
350	ProtobufTimeUtils.toJavaInstant(a.getScheduleTime()),	×
351	ProtobufTimeUtils.toJavaInstant(a.getActualTime()),	×
352	new ScheduleActionExecutionStartWorkflow(
353	a.getStartWorkflowResult().getWorkflowId(),	×
354	a.getStartWorkflowResult().getRunId())))	×
355	.collect(Collectors.toList());	×
356	}
357
358	public ScheduleListDescription protoToScheduleListDescription(ScheduleListEntry entry) {
359	List<Instant> nextActionTimes =	×
360	entry.getInfo().getFutureActionTimesList().stream()	×
361	.map(ProtobufTimeUtils::toJavaInstant)	×
362	.collect(Collectors.toList());	×
363
364	io.temporal.client.schedules.ScheduleListInfo info =	×
365	new io.temporal.client.schedules.ScheduleListInfo(
366	this.protoToActionResults(entry.getInfo().getRecentActionsList()), nextActionTimes);	×
367
368	ScheduleListActionStartWorkflow action =	×
369	new ScheduleListActionStartWorkflow(entry.getInfo().getWorkflowType().getName());	×
370
371	ScheduleListState state =	×
372	new ScheduleListState(entry.getInfo().getNotes(), entry.getInfo().getPaused());	×
373
374	io.temporal.client.schedules.ScheduleSpec spec = protoToScheduleSpec(entry.getInfo().getSpec());	×
375
376	return new ScheduleListDescription(	×
377	entry.getScheduleId(),	×
378	new ScheduleListSchedule(action, spec, state),
379	info,
380	entry.getMemo().getFieldsMap(),	×
381	this.clientOptions.getDataConverter(),	×
382	Collections.unmodifiableMap(SearchAttributesUtil.decode(entry.getSearchAttributes())));	×
383	}
384
385	@Nonnull
386	public io.temporal.client.schedules.ScheduleAction protoToAction(@Nonnull ScheduleAction action) {
387	Objects.requireNonNull(action);	×
388	if (action.hasStartWorkflow()) {	×
389	NewWorkflowExecutionInfo startWfAction = action.getStartWorkflow();	×
390	ScheduleActionStartWorkflow.Builder builder = ScheduleActionStartWorkflow.newBuilder();	×
391	builder.setWorkflowType(startWfAction.getWorkflowType().getName());	×
392
393	builder.setRawArguments(	×
394	new EncodedValues(
395	Optional.of(startWfAction.getInput()), clientOptions.getDataConverter()));	×
396
397	WorkflowOptions.Builder wfOptionsBuilder = WorkflowOptions.newBuilder();	×
398	// set required options
399	wfOptionsBuilder.setWorkflowId(startWfAction.getWorkflowId());	×
400	wfOptionsBuilder.setTaskQueue(startWfAction.getTaskQueue().getName());	×
401	// set timeouts
402	wfOptionsBuilder.setWorkflowExecutionTimeout(	×
403	ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowExecutionTimeout()));	×
404
405	wfOptionsBuilder.setWorkflowRunTimeout(	×
406	ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowRunTimeout()));	×
407
408	wfOptionsBuilder.setWorkflowTaskTimeout(	×
409	ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowTaskTimeout()));	×
410
411	if (startWfAction.getRetryPolicy() != null) {	×
412	wfOptionsBuilder.setRetryOptions(	×
413	RetryOptionsUtils.toRetryOptions(startWfAction.getRetryPolicy()));	×
414	}
415
416	if (startWfAction.hasMemo()) {	×
417	Map<String, Object> memos = new HashMap<>();	×
418	for (Map.Entry<String, Payload> memo : startWfAction.getMemo().getFieldsMap().entrySet()) {	×
419	EncodedValues encodedMemo =	×
420	new EncodedValues(
421	Optional.of(Payloads.newBuilder().addPayloads(memo.getValue()).build()),	×
422	clientOptions.getDataConverter());	×
423	memos.put(memo.getKey(), encodedMemo);	×
424	}	×
425	wfOptionsBuilder.setMemo(memos);	×
426	}
427
428	if (startWfAction.hasSearchAttributes()) {	×
429	wfOptionsBuilder.setSearchAttributes(	×
430	Collections.unmodifiableMap(	×
431	SearchAttributesUtil.decode(startWfAction.getSearchAttributes())));	×
432	}
433
434	builder.setOptions(wfOptionsBuilder.build());	×
435	return builder.build();	×
436	}
437	throw new IllegalArgumentException("Unsupported action " + action.getActionCase());	×
438	}
439
440	@Nullable
441	public SchedulePolicy protoToPolicy(@Nullable SchedulePolicies policy) {
442	if (policy == null) {	×
443	return null;	×
444	}
445	SchedulePolicy.Builder policyBuilder =
446	SchedulePolicy.newBuilder()	×
447	.setPauseOnFailure(policy.getPauseOnFailure())	×
448	.setOverlap(policy.getOverlapPolicy());	×
449	if (policy.hasCatchupWindow()) {	×
450	policyBuilder.setCatchupWindow(ProtobufTimeUtils.toJavaDuration(policy.getCatchupWindow()));	×
451	}
452	return policyBuilder.build();	×
453	}
454
455	@Nullable
456	public io.temporal.client.schedules.ScheduleState protoToScheduleState(
457	@Nullable ScheduleState state) {
458	if (state == null) {	×
459	return null;	×
460	}
461	return io.temporal.client.schedules.ScheduleState.newBuilder()	×
462	.setNote(state.getNotes())	×
463	.setPaused(state.getPaused())	×
464	.setRemainingActions(state.getRemainingActions())	×
465	.setLimitedAction(state.getLimitedActions())	×
466	.build();	×
467	}
468
469	public io.temporal.client.schedules.Schedule protoToSchedule(Schedule schedule) {
470	return io.temporal.client.schedules.Schedule.newBuilder()	×
471	.setAction(protoToAction(schedule.getAction()))	×
472	.setSpec(protoToScheduleSpec(schedule.getSpec()))	×
473	.setPolicy(protoToPolicy(schedule.getPolicies()))	×
474	.setState(protoToScheduleState(schedule.getState()))	×
475	.build();	×
476	}
477
478	public io.temporal.client.schedules.ScheduleInfo protoToScheduleInfo(ScheduleInfo info) {
479	return new io.temporal.client.schedules.ScheduleInfo(	×
480	info.getActionCount(),	×
481	info.getMissedCatchupWindow(),	×
482	info.getOverlapSkipped(),	×
483	info.getRunningWorkflowsList().stream()	×
484	.map(wf -> new ScheduleActionExecutionStartWorkflow(wf.getWorkflowId(), wf.getRunId()))	×
485	.collect(Collectors.toList()),	×
486	this.protoToActionResults(info.getRecentActionsList()),	×
487	info.getFutureActionTimesList().stream()	×
488	.map(t -> ProtobufTimeUtils.toJavaInstant(t))	×
489	.collect(Collectors.toList()),	×
490	info.hasCreateTime() ? ProtobufTimeUtils.toJavaInstant(info.getCreateTime()) : null,	×
491	info.hasUpdateTime() ? ProtobufTimeUtils.toJavaInstant(info.getUpdateTime()) : null);	×
492	}
493	}

temporalio / sdk-java / #174

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