DeleteJobRequest and GetJobRequest API's

Author: charan2628

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

@@ -666,14 +666,6 @@ Flux<SubscribeConfigurationResponse> subscribeConfiguration(String storeName, Li
    */
   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

@@ -13,6 +13,7 @@
 
 package io.dapr.client;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Strings;
 import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
@@ -59,6 +60,7 @@
 import io.dapr.client.domain.UnsubscribeConfigurationRequest;
 import io.dapr.client.domain.UnsubscribeConfigurationResponse;
 import io.dapr.client.resiliency.ResiliencyOptions;
+import io.dapr.config.Properties;
 import io.dapr.exceptions.DaprException;
 import io.dapr.internal.exceptions.DaprHttpException;
 import io.dapr.internal.grpc.DaprClientGrpcInterceptors;
@@ -92,6 +94,7 @@
 import reactor.util.retry.Retry;
 
 import java.io.IOException;
+import java.nio.charset.Charset;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -140,6 +143,10 @@ public class DaprClientImpl extends AbstractDaprClient {
   private final DaprHttp httpClient;
 
   private final DaprClientGrpcInterceptors grpcInterceptors;
+  
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+  
+  private static final Charset CHARSET = Properties.STRING_CHARSET.get();
 
   /**
    * Default access level constructor, in order to create an instance of this class use io.dapr.client.DaprClientBuilder
@@ -1313,10 +1320,15 @@ public <T> Mono<Void> scheduleJobAlpha1(Job<T> job) {
 
   	  DaprProtos.Job.Builder jobBuilder = DaprProtos.Job.newBuilder()
   			  .setName(name);
-  	  if (data instanceof Message) {
-  		  jobBuilder.setData(Any.pack((Message)job.getData()));
+  	  if (data instanceof String) {
+  		  jobBuilder.setData(Any.newBuilder().setValue(ByteString.copyFrom((String) data, CHARSET)));
+  	  } else if (data instanceof byte[]) {
+  		  String base64 = OBJECT_MAPPER.writeValueAsString(data);
+  		  jobBuilder.setData(Any.newBuilder().setValue(ByteString.copyFrom(base64, CHARSET)));
   	  } else {
-  		  jobBuilder.setData(Any.newBuilder().setValue(ByteString.copyFrom(this.objectSerializer.serialize(data))));
+  		return Mono.error(() -> {
+            throw new IllegalArgumentException("Job data value must be String or byte[]");
+          });
   	  }
   	  if (job.getSchedule() != null && !job.getSchedule().trim().isEmpty()) {
   		  jobBuilder.setSchedule(job.getSchedule());
@@ -1342,6 +1354,46 @@ public <T> Mono<Void> scheduleJobAlpha1(Job<T> job) {
   	  return DaprException.wrapMono(ex);
     }
   }
+  
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> Mono<Job<T>> getJobAlpha1(String name, Class<T> clazz) {
+	  try {
+		  if (name == null || name.trim().isEmpty()) {
+	  		  throw new IllegalArgumentException("Job name cannot be null or empty");
+	  	  }
+		  return this.<DaprProtos.GetJobResponse>createMono(
+				  it -> intercept(null, asyncStub).getJobAlpha1(DaprProtos.GetJobRequest.newBuilder().setName(name).build(), it))
+		  .map(it -> {
+			  DaprProtos.Job _job = it.getJob();
+			  T data = null;
+			  if (clazz.isInstance(String.class)) {
+				  data = (T)_job.getData().toByteString().toString(CHARSET);
+			  } else if (clazz.isInstance(byte[].class)) {
+				  data = (T) _job.getData().toByteArray();
+			  } else {
+				  throw new IllegalArgumentException("Job data type must be String or byte[]");
+			  }
+			  return new Job<>(_job.getName(), _job.getSchedule(), _job.getRepeats(), _job.getDueTime(), _job.getTtl(), data);
+		  });
+	  } catch (Exception ex) {
+		  return DaprException.wrapMono(ex);
+	  }
+  }
+  
+  @Override
+  public Mono<Void> deleteJobAlpha1(String name) {
+	  try {
+		  if (name == null || name.trim().isEmpty()) {
+	  		  throw new IllegalArgumentException("Job name cannot be null or empty");
+	  	  }
+		  return this.<DaprProtos.DeleteJobResponse>createMono(
+				  it -> intercept(null, asyncStub).deleteJobAlpha1(DaprProtos.DeleteJobRequest.newBuilder().setName(name).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/DaprPreviewClient.java

@@ -17,6 +17,7 @@
 import io.dapr.client.domain.BulkPublishRequest;
 import io.dapr.client.domain.BulkPublishResponse;
 import io.dapr.client.domain.BulkPublishResponseFailedEntry;
+import io.dapr.client.domain.Job;
 import io.dapr.client.domain.LockRequest;
 import io.dapr.client.domain.QueryStateRequest;
 import io.dapr.client.domain.QueryStateResponse;
@@ -268,4 +269,30 @@ <T> Mono<BulkPublishResponse<T>> publishEvents(String pubsubName, String topicNa
    */
   <T> Subscription subscribeToEvents(
       String pubsubName, String topic, SubscriptionListener<T> listener, TypeRef<T> type);
+  
+  /**
+   * 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);
+  
+  /**
+   * GetJobAlpha1 retrieve Job by name.
+   * 
+   * @param <T> The type of the data for the job.
+   * @param name name of the job
+   * @return a Mono of Job
+   */
+  <T> Mono<Job<T>> getJobAlpha1(String name, Class<T> clazz);
+  
+  /**
+   * Delete a Job.
+   * 
+   * @param name name of the job
+   * @return
+   */
+  Mono<Void> deleteJobAlpha1(String name);
 }

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

@@ -32,6 +32,26 @@ public Job(String name, T data) {
 		this.data = data;
 	}
 
+	/**
+	 * Constructor for Job
+	 * 
+	 * @param name name of the job to create
+	 * @param schedule schedule for the job
+	 * @param repeats jobs with fixed repeat counts (accounting for Actor Reminders).
+	 * @param dueTime sets time at which or time interval before the callback is invoked for the first time.
+	 * @param ttl Time To Live to allow for auto deletes (accounting for Actor Reminders).
+	 * @param data Job data
+	 */
+	public Job(String name, String schedule, Integer repeats, String dueTime, String ttl, T data) {
+		super();
+		this.name = name;
+		this.schedule = schedule;
+		this.repeats = repeats;
+		this.dueTime = dueTime;
+		this.ttl = ttl;
+		this.data = data;
+	}
+
 	public String getSchedule() {
 		return schedule;
 	}