Support of @ExecuteOn(Scheduler) (#865)

services-api/src/main/java/io/scalecube/services/Reflect.java

@@ -5,6 +5,7 @@
 import static io.scalecube.services.CommunicationMode.REQUEST_RESPONSE;
 import static io.scalecube.services.CommunicationMode.REQUEST_STREAM;
 
+import io.scalecube.services.annotations.ExecuteOn;
 import io.scalecube.services.annotations.RequestType;
 import io.scalecube.services.annotations.ResponseType;
 import io.scalecube.services.annotations.Service;
@@ -28,6 +29,8 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 
 public class Reflect {
 
@@ -172,7 +175,8 @@ public static Map<Method, MethodInfo> methodsInfo(Class<?> serviceInterface) {
  method.getParameterCount(),
  requestType(method),
  isRequestTypeServiceMessage(method),
- isSecured(method)))));
+ isSecured(method),
+ null))));
  }
 
  /**
@@ -379,4 +383,54 @@ public static boolean isSecured(Method method) {
  return method.isAnnotationPresent(Secured.class)
  || method.getDeclaringClass().isAnnotationPresent(Secured.class);
  }
+
+ public static Scheduler executeOnScheduler(Method method, Map<String, Scheduler> schedulers) {
+ final Class<?> declaringClass = method.getDeclaringClass();
+
+ if (method.isAnnotationPresent(ExecuteOn.class)) {
+ final var executeOn = method.getAnnotation(ExecuteOn.class);
+ final var name = executeOn.value();
+ final var scheduler = schedulers.get(name);
+ if (scheduler == null) {
+ throw new IllegalArgumentException(
+ "Wrong @ExecuteOn definition on "
+ + declaringClass.getName()
+ + "."
+ + method.getName()
+ + ": scheduler (name="
+ + name
+ + ") cannot be found");
+ }
+ return scheduler;
+ }
+
+ // If @ExecuteOn annotation is not present on service method, then find it on service class
+
+ ExecuteOn executeOn = null;
+ for (var clazz = declaringClass; clazz != null; clazz = clazz.getSuperclass()) {
+ executeOn = clazz.getAnnotation(ExecuteOn.class);
+ if (executeOn != null) {
+ break;
+ }
+ }
+
+ if (executeOn == null) {
+ return Schedulers.immediate();
+ }
+
+ final var name = executeOn.value();
+ final var scheduler = schedulers.get(name);
+ if (scheduler == null) {
+ throw new IllegalArgumentException(
+ "Wrong @ExecuteOn definition on "
+ + declaringClass.getName()
+ + "."
+ + method.getName()
+ + ": scheduler (name="
+ + name
+ + ") cannot be found");
+ }
+
+ return scheduler;
+ }
 }

services-api/src/main/java/io/scalecube/services/annotations/ExecuteOn.java

@@ -0,0 +1,26 @@
+package io.scalecube.services.annotations;
+
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.TYPE;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+/**
+ * This annotation is used to mark that particular service method or all service methods will be
+ * executed in the specified scheduler.
+ */
+@Documented
+@Target({METHOD, TYPE})
+@Retention(RUNTIME)
+public @interface ExecuteOn {
+
+ /**
+ * Returns scheduler name.
+ *
+ * @return scheduler name
+ */
+ String value();
+}

services-api/src/main/java/io/scalecube/services/annotations/Service.java

@@ -5,7 +5,7 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-/** Indicates that an annotated class is an Service Fabric service object. */
+/** Indicates that annotated class is a ScaleCube service object. */
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.TYPE)
 public @interface Service {

services-api/src/main/java/io/scalecube/services/annotations/ServiceMethod.java

@@ -6,7 +6,7 @@
 import java.lang.annotation.Target;
 
 /**
- * Indicates that an annotated method is a service method available via Service Fabric framework.
+ * Indicates that annotated method is a ScaleCube service method.
  */
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})

services-api/src/main/java/io/scalecube/services/methods/MethodInfo.java

@@ -4,6 +4,7 @@
 import io.scalecube.services.api.Qualifier;
 import java.lang.reflect.Type;
 import java.util.StringJoiner;
+import reactor.core.scheduler.Scheduler;
 
 public final class MethodInfo {
 
@@ -17,6 +18,7 @@ public final class MethodInfo {
  private final Class<?> requestType;
  private final boolean isRequestTypeServiceMessage;
  private final boolean isSecured;
+ private final Scheduler scheduler;
 
  /**
  * Create a new service info.
@@ -30,6 +32,7 @@ public final class MethodInfo {
  * @param requestType the type of the request
  * @param isRequestTypeServiceMessage is request service message
  * @param isSecured is method protected by authentication
+ * @param scheduler scheduler
  */
  public MethodInfo(
  String serviceName,
@@ -40,7 +43,8 @@ public MethodInfo(
  int parameterCount,
  Class<?> requestType,
  boolean isRequestTypeServiceMessage,
- boolean isSecured) {
+ boolean isSecured,
+ Scheduler scheduler) {
  this.parameterizedReturnType = parameterizedReturnType;
  this.isReturnTypeServiceMessage = isReturnTypeServiceMessage;
  this.communicationMode = communicationMode;
@@ -51,6 +55,7 @@ public MethodInfo(
  this.requestType = requestType;
  this.isRequestTypeServiceMessage = isRequestTypeServiceMessage;
  this.isSecured = isSecured;
+ this.scheduler = scheduler;
  }
 
  public String serviceName() {
@@ -101,19 +106,24 @@ public boolean isSecured() {
  return isSecured;
  }
 
+ public Scheduler scheduler() {
+ return scheduler;
+ }
+
  @Override
  public String toString() {
  return new StringJoiner(", ", MethodInfo.class.getSimpleName() + "[", "]")
- .add("serviceName=" + serviceName)
- .add("methodName=" + methodName)
- .add("qualifier=" + qualifier)
+ .add("serviceName='" + serviceName + "'")
+ .add("methodName='" + methodName + "'")
+ .add("qualifier='" + qualifier + "'")
  .add("parameterizedReturnType=" + parameterizedReturnType)
  .add("isReturnTypeServiceMessage=" + isReturnTypeServiceMessage)
  .add("communicationMode=" + communicationMode)
  .add("parameterCount=" + parameterCount)
  .add("requestType=" + requestType)
  .add("isRequestTypeServiceMessage=" + isRequestTypeServiceMessage)
  .add("isSecured=" + isSecured)
+ .add("scheduler=" + scheduler)
  .toString();
  }
 }

services-api/src/main/java/io/scalecube/services/methods/ServiceMethodInvoker.java

@@ -70,7 +70,8 @@ public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
  .flatMap(authData -> deferWithContextOne(message, authData))
  .map(response -> toResponse(response, message.qualifier(), message.dataFormat()))
  .onErrorResume(
- throwable -> Mono.just(errorMapper.toMessage(message.qualifier(), throwable)));
+ throwable -> Mono.just(errorMapper.toMessage(message.qualifier(), throwable)))
+ .subscribeOn(methodInfo.scheduler());
  }
 
  /**
@@ -84,7 +85,8 @@ public Flux<ServiceMessage> invokeMany(ServiceMessage message) {
  .flatMapMany(authData -> deferWithContextMany(message, authData))
  .map(response -> toResponse(response, message.qualifier(), message.dataFormat()))
  .onErrorResume(
- throwable -> Flux.just(errorMapper.toMessage(message.qualifier(), throwable)));
+ throwable -> Flux.just(errorMapper.toMessage(message.qualifier(), throwable)))
+ .subscribeOn(methodInfo.scheduler());
  }
 
  /**
@@ -104,7 +106,8 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
  toResponse(response, first.get().qualifier(), first.get().dataFormat()))
  .onErrorResume(
  throwable ->
- Flux.just(errorMapper.toMessage(first.get().qualifier(), throwable))));
+ Flux.just(errorMapper.toMessage(first.get().qualifier(), throwable)))
+ .subscribeOn(methodInfo.scheduler()));
  }
 
  private Mono<?> deferWithContextOne(ServiceMessage message, Object authData) {

services-api/src/test/java/io/scalecube/services/methods/ServiceMethodInvokerTest.java

@@ -18,6 +18,7 @@
 import org.mockito.Mockito;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 
 class ServiceMethodInvokerTest {
@@ -62,7 +63,8 @@ void testInvokeOneWhenReturnNull() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -100,7 +102,8 @@ void testInvokeManyWhenReturnNull() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -138,7 +141,8 @@ void testInvokeBidirectionalWhenReturnNull() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -177,7 +181,8 @@ void testInvokeOneWhenThrowException() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -219,7 +224,8 @@ void testInvokeManyWhenThrowException() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -260,7 +266,8 @@ void testInvokeBidirectionalWhenThrowException() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -305,7 +312,8 @@ void testAuthMethodWhenNoContextAndNoAuthenticator() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -347,7 +355,8 @@ void testAuthMethodWhenThereIsContextAndNoAuthenticator() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  serviceMethodInvoker =
  new ServiceMethodInvoker(
@@ -387,7 +396,8 @@ void testAuthMethodWhenNoContextButThereIsAuthenticator() throws Exception {
  method.getParameterCount(),
  Void.TYPE,
  IS_REQUEST_TYPE_SERVICE_MESSAGE,
- AUTH);
+ AUTH,
+ Schedulers.immediate());
 
  //noinspection unchecked,rawtypes
  Authenticator<Map> mockedAuthenticator = Mockito.mock(Authenticator.class);