#175

pending completion

Build # #175

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Worker / Build Id versioning (#1786)

Implement new worker build id based versioning feature

Run Details

236 of 236 new or added lines in 24 files covered. (100.0%)

18343 of 23697 relevant lines covered (77.41%)

0.81 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.getTypedSearchAttributes() != null
          && wfOptions.getTypedSearchAttributes().size() > 0) {
        workflowRequest.setSearchAttributes(
            SearchAttributesUtil.encodeTyped(wfOptions.getTypedSearchAttributes()));
      }

      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.setTypedSearchAttributes(
            SearchAttributesUtil.decodeTyped(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.getTypedSearchAttributes() != null	×
138	&& wfOptions.getTypedSearchAttributes().size() > 0) {	×
139	workflowRequest.setSearchAttributes(	×
140	SearchAttributesUtil.encodeTyped(wfOptions.getTypedSearchAttributes()));	×
141	}
142
143	Header grpcHeader =	×
144	toHeaderGrpc(	×
145	startWorkflowAction.getHeader(),	×
146	extractContextsAndConvertToBytes(wfOptions.getContextPropagators()));	×
147	workflowRequest.setHeader(grpcHeader);	×
148
149	return ScheduleAction.newBuilder().setStartWorkflow(workflowRequest.build()).build();	×
150	}
151	throw new IllegalArgumentException("Unsupported action " + action.getClass());	×
152	}
153
154	public SchedulePolicies policyToProto(SchedulePolicy policy) {
155	return SchedulePolicies.newBuilder()	×
156	.setCatchupWindow(ProtobufTimeUtils.toProtoDuration(policy.getCatchupWindow()))	×
157	.setPauseOnFailure(policy.isPauseOnFailure())	×
158	.setOverlapPolicy(policy.getOverlap())	×
159	.build();	×
160	}
161
162	public List<Range> scheduleRangeToProto(List<ScheduleRange> scheduleRanges) {
163	ArrayList<Range> ranges = new ArrayList<Range>(scheduleRanges.size());	×
164	for (ScheduleRange scheduleRange : scheduleRanges) {	×
165	ranges.add(	×
166	Range.newBuilder()	×
167	.setStart(scheduleRange.getStart())	×
168	.setEnd(scheduleRange.getEnd())	×
169	.setStep(scheduleRange.getStep())	×
170	.build());	×
171	}	×
172	return ranges;	×
173	}
174
175	public ScheduleSpec specToProto(io.temporal.client.schedules.ScheduleSpec spec) {
176	ScheduleSpec.Builder builder = ScheduleSpec.newBuilder();	×
177
178	if (spec.getTimeZoneName() != null && !spec.getTimeZoneName().isEmpty()) {	×
179	builder.setTimezoneName(spec.getTimeZoneName());	×
180	}
181
182	if (spec.getJitter() != null) {	×
183	builder.setJitter(ProtobufTimeUtils.toProtoDuration(spec.getJitter()));	×
184	}
185
186	if (spec.getStartAt() != null) {	×
187	builder.setStartTime(ProtobufTimeUtils.toProtoTimestamp(spec.getStartAt()));	×
188	}
189
190	if (spec.getEndAt() != null) {	×
191	builder.setEndTime(ProtobufTimeUtils.toProtoTimestamp(spec.getEndAt()));	×
192	}
193
194	if (spec.getCalendars() != null && !spec.getCalendars().isEmpty()) {	×
195	for (ScheduleCalendarSpec calendarSpec : spec.getCalendars()) {	×
196	builder.addStructuredCalendar(	×
197	StructuredCalendarSpec.newBuilder()	×
198	.addAllSecond(this.scheduleRangeToProto(calendarSpec.getSeconds()))	×
199	.addAllMinute(this.scheduleRangeToProto(calendarSpec.getMinutes()))	×
200	.addAllHour(this.scheduleRangeToProto(calendarSpec.getHour()))	×
201	.addAllDayOfMonth(this.scheduleRangeToProto(calendarSpec.getDayOfMonth()))	×
202	.addAllMonth(this.scheduleRangeToProto(calendarSpec.getMonth()))	×
203	.addAllYear(this.scheduleRangeToProto(calendarSpec.getYear()))	×
204	.addAllDayOfWeek(this.scheduleRangeToProto(calendarSpec.getDayOfWeek()))	×
205	.setComment(calendarSpec.getComment())	×
206	.build());	×
207	}	×
208	}
209
210	if (spec.getIntervals() != null && !spec.getIntervals().isEmpty()) {	×
211	for (ScheduleIntervalSpec intervalSpec : spec.getIntervals()) {	×
212	builder.addInterval(	×
213	IntervalSpec.newBuilder()	×
214	.setInterval(ProtobufTimeUtils.toProtoDuration(intervalSpec.getEvery()))	×
215	.setPhase(ProtobufTimeUtils.toProtoDuration(intervalSpec.getOffset()))	×
216	.build());	×
217	}	×
218	}
219
220	if (spec.getCronExpressions() != null && !spec.getCronExpressions().isEmpty()) {	×
221	builder.addAllCronString(spec.getCronExpressions());	×
222	}
223
224	if (spec.getSkip() != null && !spec.getSkip().isEmpty()) {	×
225	for (ScheduleCalendarSpec calendarSpec : spec.getSkip()) {	×
226	builder.addExcludeStructuredCalendar(	×
227	StructuredCalendarSpec.newBuilder()	×
228	.addAllSecond(this.scheduleRangeToProto(calendarSpec.getSeconds()))	×
229	.addAllMinute(this.scheduleRangeToProto(calendarSpec.getMinutes()))	×
230	.addAllHour(this.scheduleRangeToProto(calendarSpec.getHour()))	×
231	.addAllDayOfMonth(this.scheduleRangeToProto(calendarSpec.getDayOfMonth()))	×
232	.addAllMonth(this.scheduleRangeToProto(calendarSpec.getMonth()))	×
233	.addAllYear(this.scheduleRangeToProto(calendarSpec.getYear()))	×
234	.addAllDayOfWeek(this.scheduleRangeToProto(calendarSpec.getDayOfWeek()))	×
235	.setComment(calendarSpec.getComment())	×
236	.build());	×
237	}	×
238	}
239
240	return builder.build();	×
241	}
242
243	public @Nonnull Schedule scheduleToProto(
244	@Nonnull io.temporal.client.schedules.Schedule schedule) {
245	Preconditions.checkNotNull(schedule);	×
246
247	Schedule.Builder scheduleBuilder =
248	Schedule.newBuilder()	×
249	.setAction(this.actionToProto(schedule.getAction()))	×
250	.setSpec(this.specToProto(schedule.getSpec()));	×
251
252	if (schedule.getPolicy() != null) {	×
253	scheduleBuilder.setPolicies(this.policyToProto(schedule.getPolicy()));	×
254	}
255
256	if (schedule.getState() != null) {	×
257	scheduleBuilder.setState(this.stateToProto(schedule.getState()));	×
258	}
259
260	return scheduleBuilder.build();	×
261	}
262
263	public ScheduleState stateToProto(io.temporal.client.schedules.ScheduleState state) {
264	ScheduleState.Builder stateBuilder =
265	ScheduleState.newBuilder()	×
266	.setLimitedActions(state.isLimitedAction())	×
267	.setRemainingActions(state.getRemainingActions())	×
268	.setPaused(state.isPaused());	×
269	if (state.getNote() != null) {	×
270	stateBuilder.setNotes(state.getNote());	×
271	}
272	return stateBuilder.build();	×
273	}
274
275	public ScheduleRange protoToScheduleRange(Range protoRange) {
276	return new ScheduleRange(protoRange.getStart(), protoRange.getEnd(), protoRange.getStep());	×
277	}
278
279	public ScheduleIntervalSpec protoToScheduleInterval(IntervalSpec protoInterval) {
280	return new ScheduleIntervalSpec(	×
281	ProtobufTimeUtils.toJavaDuration(protoInterval.getInterval()),	×
282	ProtobufTimeUtils.toJavaDuration(protoInterval.getPhase()));	×
283	}
284
285	public List<ScheduleRange> protoToScheduleRanges(List<Range> protoRanges) {
286	return protoRanges.stream().map(t -> this.protoToScheduleRange(t)).collect(Collectors.toList());	×
287	}
288
289	public ScheduleCalendarSpec protoToScheduleCalendar(StructuredCalendarSpec protoSpec) {
290	ScheduleCalendarSpec.Builder calendarBuilder =
291	ScheduleCalendarSpec.newBuilder()	×
292	.setComment(protoSpec.getComment())	×
293	.setSeconds(protoToScheduleRanges(protoSpec.getSecondList()))	×
294	.setMinutes(protoToScheduleRanges(protoSpec.getMinuteList()))	×
295	.setHour(protoToScheduleRanges(protoSpec.getHourList()))	×
296	.setDayOfMonth(protoToScheduleRanges(protoSpec.getDayOfMonthList()))	×
297	.setMonth(protoToScheduleRanges(protoSpec.getMonthList()))	×
298	.setYear(protoToScheduleRanges(protoSpec.getYearList()))	×
299	.setDayOfWeek(protoToScheduleRanges(protoSpec.getDayOfWeekList()));	×
300
301	return calendarBuilder.build();	×
302	}
303
304	@Nonnull
305	public io.temporal.client.schedules.ScheduleSpec protoToScheduleSpec(
306	@Nonnull ScheduleSpec scheduleSpec) {
307	Objects.requireNonNull(scheduleSpec);	×
308	io.temporal.client.schedules.ScheduleSpec.Builder specBuilder =
309	io.temporal.client.schedules.ScheduleSpec.newBuilder()	×
310	.setTimeZoneName(	×
311	scheduleSpec.getTimezoneName() == null ? "" : scheduleSpec.getTimezoneName());	×
312
313	if (scheduleSpec.hasJitter()) {	×
314	specBuilder.setJitter(ProtobufTimeUtils.toJavaDuration(scheduleSpec.getJitter()));	×
315	}
316
317	if (scheduleSpec.hasStartTime()) {	×
318	specBuilder.setStartAt(ProtobufTimeUtils.toJavaInstant(scheduleSpec.getStartTime()));	×
319	}
320
321	if (scheduleSpec.hasEndTime()) {	×
322	specBuilder.setEndAt(ProtobufTimeUtils.toJavaInstant(scheduleSpec.getEndTime()));	×
323	}
324
325	specBuilder.setCalendars(	×
326	scheduleSpec.getStructuredCalendarList().stream()	×
327	.map(c -> this.protoToScheduleCalendar(c))	×
328	.collect(Collectors.toList()));	×
329
330	specBuilder.setCronExpressions(scheduleSpec.getCronStringList());	×
331
332	specBuilder.setIntervals(	×
333	scheduleSpec.getIntervalList().stream()	×
334	.map(i -> this.protoToScheduleInterval(i))	×
335	.collect(Collectors.toList()));	×
336
337	specBuilder.setSkip(	×
338	scheduleSpec.getExcludeStructuredCalendarList().stream()	×
339	.map(c -> this.protoToScheduleCalendar(c))	×
340	.collect(Collectors.toList()));	×
341
342	return specBuilder.build();	×
343	}
344
345	public List<io.temporal.client.schedules.ScheduleActionResult> protoToActionResults(
346	List<ScheduleActionResult> results) {
347	return results.stream()	×
348	.map(	×
349	a ->
350	new io.temporal.client.schedules.ScheduleActionResult(	×
351	ProtobufTimeUtils.toJavaInstant(a.getScheduleTime()),	×
352	ProtobufTimeUtils.toJavaInstant(a.getActualTime()),	×
353	new ScheduleActionExecutionStartWorkflow(
354	a.getStartWorkflowResult().getWorkflowId(),	×
355	a.getStartWorkflowResult().getRunId())))	×
356	.collect(Collectors.toList());	×
357	}
358
359	public ScheduleListDescription protoToScheduleListDescription(ScheduleListEntry entry) {
360	List<Instant> nextActionTimes =	×
361	entry.getInfo().getFutureActionTimesList().stream()	×
362	.map(ProtobufTimeUtils::toJavaInstant)	×
363	.collect(Collectors.toList());	×
364
365	io.temporal.client.schedules.ScheduleListInfo info =	×
366	new io.temporal.client.schedules.ScheduleListInfo(
367	this.protoToActionResults(entry.getInfo().getRecentActionsList()), nextActionTimes);	×
368
369	ScheduleListActionStartWorkflow action =	×
370	new ScheduleListActionStartWorkflow(entry.getInfo().getWorkflowType().getName());	×
371
372	ScheduleListState state =	×
373	new ScheduleListState(entry.getInfo().getNotes(), entry.getInfo().getPaused());	×
374
375	io.temporal.client.schedules.ScheduleSpec spec = protoToScheduleSpec(entry.getInfo().getSpec());	×
376
377	return new ScheduleListDescription(	×
378	entry.getScheduleId(),	×
379	new ScheduleListSchedule(action, spec, state),
380	info,
381	entry.getMemo().getFieldsMap(),	×
382	this.clientOptions.getDataConverter(),	×
383	Collections.unmodifiableMap(SearchAttributesUtil.decode(entry.getSearchAttributes())));	×
384	}
385
386	@Nonnull
387	public io.temporal.client.schedules.ScheduleAction protoToAction(@Nonnull ScheduleAction action) {
388	Objects.requireNonNull(action);	×
389	if (action.hasStartWorkflow()) {	×
390	NewWorkflowExecutionInfo startWfAction = action.getStartWorkflow();	×
391	ScheduleActionStartWorkflow.Builder builder = ScheduleActionStartWorkflow.newBuilder();	×
392	builder.setWorkflowType(startWfAction.getWorkflowType().getName());	×
393
394	builder.setRawArguments(	×
395	new EncodedValues(
396	Optional.of(startWfAction.getInput()), clientOptions.getDataConverter()));	×
397
398	WorkflowOptions.Builder wfOptionsBuilder = WorkflowOptions.newBuilder();	×
399	// set required options
400	wfOptionsBuilder.setWorkflowId(startWfAction.getWorkflowId());	×
401	wfOptionsBuilder.setTaskQueue(startWfAction.getTaskQueue().getName());	×
402	// set timeouts
403	wfOptionsBuilder.setWorkflowExecutionTimeout(	×
404	ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowExecutionTimeout()));	×
405
406	wfOptionsBuilder.setWorkflowRunTimeout(	×
407	ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowRunTimeout()));	×
408
409	wfOptionsBuilder.setWorkflowTaskTimeout(	×
410	ProtobufTimeUtils.toJavaDuration(startWfAction.getWorkflowTaskTimeout()));	×
411
412	if (startWfAction.getRetryPolicy() != null) {	×
413	wfOptionsBuilder.setRetryOptions(	×
414	RetryOptionsUtils.toRetryOptions(startWfAction.getRetryPolicy()));	×
415	}
416
417	if (startWfAction.hasMemo()) {	×
418	Map<String, Object> memos = new HashMap<>();	×
419	for (Map.Entry<String, Payload> memo : startWfAction.getMemo().getFieldsMap().entrySet()) {	×
420	EncodedValues encodedMemo =	×
421	new EncodedValues(
422	Optional.of(Payloads.newBuilder().addPayloads(memo.getValue()).build()),	×
423	clientOptions.getDataConverter());	×
424	memos.put(memo.getKey(), encodedMemo);	×
425	}	×
426	wfOptionsBuilder.setMemo(memos);	×
427	}
428
429	if (startWfAction.hasSearchAttributes()) {	×
430	wfOptionsBuilder.setTypedSearchAttributes(	×
431	SearchAttributesUtil.decodeTyped(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 / #175

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