Schedule Job API

Author: charan2628

sdk/src/main/java/io/dapr/client/DaprClient.java

@@ -25,6 +25,7 @@
 import io.dapr.client.domain.HttpExtension;
 import io.dapr.client.domain.InvokeBindingRequest;
 import io.dapr.client.domain.InvokeMethodRequest;
+import io.dapr.client.domain.Job;
 import io.dapr.client.domain.PublishEventRequest;
 import io.dapr.client.domain.SaveStateRequest;
 import io.dapr.client.domain.State;
@@ -44,6 +45,8 @@
 import java.util.Map;
 import java.util.function.Function;
 
+import com.google.protobuf.Message;
+
 /**
  * Generic Client Adapter to be used regardless of the GRPC or the HTTP Client implementation required.
  *
@@ -662,6 +665,15 @@ Flux<SubscribeConfigurationResponse> subscribeConfiguration(String storeName, Li
    * @return Mono of {@link UnsubscribeConfigurationResponse} instance.
    */
   Mono<UnsubscribeConfigurationResponse> unsubscribeConfiguration(UnsubscribeConfigurationRequest request);
+  
+  /**
+   * ScheduleJobAlpha1 creates and schedules a job.
+   * 
+   * @param <T> The type of the data for the job.
+   * @param job job to be scheduled
+   * @return a Mono plan of type Void.
+   */
+  <T> Mono<Void> scheduleJobAlpha1(Job<T> job);
 
   /**
    * Returns a newly created gRPC stub with proper interceptors and channel for gRPC proxy invocation.

sdk/src/main/java/io/dapr/client/DaprClientImpl.java

@@ -14,8 +14,10 @@
 package io.dapr.client;
 
 import com.google.common.base.Strings;
+import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Empty;
+import com.google.protobuf.Message;
 import io.dapr.client.domain.ActorMetadata;
 import io.dapr.client.domain.AppConnectionPropertiesHealthMetadata;
 import io.dapr.client.domain.AppConnectionPropertiesMetadata;
@@ -37,6 +39,7 @@
 import io.dapr.client.domain.HttpExtension;
 import io.dapr.client.domain.InvokeBindingRequest;
 import io.dapr.client.domain.InvokeMethodRequest;
+import io.dapr.client.domain.Job;
 import io.dapr.client.domain.LockRequest;
 import io.dapr.client.domain.PublishEventRequest;
 import io.dapr.client.domain.QueryStateItem;
@@ -1175,6 +1178,51 @@ public Mono<UnsubscribeConfigurationResponse> unsubscribeConfiguration(Unsubscri
     }
   }
 
+  @Override
+  public <T> Mono<Void> scheduleJobAlpha1(Job<T> job) {
+    try {
+  	  final String name = job.getName();
+  	  final T data = job.getData();
+  
+  	  if (name == null || name.trim().isEmpty()) {
+  		  throw new IllegalArgumentException("Job name cannot be null or empty");
+  	  }
+  	  if (data == null) {
+  		  throw new IllegalArgumentException("Job data cannot be empty");
+  	  }
+  
+  	  DaprProtos.Job.Builder jobBuilder = DaprProtos.Job.newBuilder()
+  			  .setName(name);
+  	  if (data instanceof Message) {
+  		  jobBuilder.setData(Any.pack((Message)job.getData()));
+  	  } else {
+  		  jobBuilder.setData(Any.newBuilder().setValue(ByteString.copyFrom(this.objectSerializer.serialize(data))));
+  	  }
+  	  if (job.getSchedule() != null && !job.getSchedule().trim().isEmpty()) {
+  		  jobBuilder.setSchedule(job.getSchedule());
+  	  }
+  	  if (job.getRepeats() != null) {
+  		  jobBuilder.setRepeats(job.getRepeats());
+  	  }
+  	  if (job.getDueTime() != null && !job.getDueTime().trim().isEmpty()) {
+  		  jobBuilder.setDueTime(job.getDueTime());
+  	  }
+  	  if (job.getTtl() != null && !job.getTtl().trim().isEmpty()) {
+  		  jobBuilder.setTtl(job.getTtl());
+  	  }
+  
+  	  DaprProtos.ScheduleJobRequest.Builder builder =
+  			  DaprProtos.ScheduleJobRequest.newBuilder()
+  			  .setJob(jobBuilder.build());
+  
+  	  return this.<DaprProtos.ScheduleJobResponse>createMono(
+  			  it -> intercept(null, asyncStub).scheduleJobAlpha1(builder.build(), it))
+  			  .then();
+    } catch (Exception ex) {
+  	  return DaprException.wrapMono(ex);
+    }
+  }
+
   /**
    * Build a new Configuration Item from provided parameter.
    *

sdk/src/main/java/io/dapr/client/domain/Job.java

@@ -0,0 +1,74 @@
+package io.dapr.client.domain;
+
+import com.google.protobuf.Message;
+
+/**
+ * A Job to schedule
+ *
+ * @param <T> The class type of Job data.
+ */
+public final class Job<T> {
+
+	private final String name;
+	
+	private String schedule;
+	
+	private Integer repeats;
+	
+	private String dueTime;
+	
+	private String ttl;
+	
+	private final T data;
+
+	/**
+	 * Constructor for Job
+	 * 
+	 * @param name name of the job to create
+	 */
+	public Job(String name, T data) {
+		super();
+		this.name = name;
+		this.data = data;
+	}
+
+	public String getSchedule() {
+		return schedule;
+	}
+
+	public void setSchedule(String schedule) {
+		this.schedule = schedule;
+	}
+
+	public Integer getRepeats() {
+		return repeats;
+	}
+
+	public void setRepeats(Integer repeats) {
+		this.repeats = repeats;
+	}
+
+	public String getDueTime() {
+		return dueTime;
+	}
+
+	public void setDueTime(String dueTime) {
+		this.dueTime = dueTime;
+	}
+
+	public String getTtl() {
+		return ttl;
+	}
+
+	public void setTtl(String ttl) {
+		this.ttl = ttl;
+	}
+
+	public T getData() {
+		return data;
+	}
+
+	public String getName() {
+		return name;
+	}
+}