feat: powerjob-remote http support spring-webmvc

powerjob-remote/pom.xml

@@ -14,6 +14,7 @@
         <module>powerjob-remote-benchmark</module>
         <module>powerjob-remote-impl-http</module>
         <module>powerjob-remote-impl-akka</module>
+        <module>powerjob-remote-impl-http-spring</module>
     </modules>
 
     <artifactId>powerjob-remote</artifactId>

powerjob-remote/powerjob-remote-impl-http-spring/pom.xml

@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>powerjob-remote</artifactId>
+        <groupId>tech.powerjob</groupId>
+        <version>4.0.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>powerjob-remote-impl-http-spring</artifactId>
+    <version>4.3.0</version>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+
+        <spring.version>5.3.23</spring.version>
+        <powerjob-remote-framework.version>4.3.0</powerjob-remote-framework.version>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>tech.powerjob</groupId>
+            <artifactId>powerjob-remote-framework</artifactId>
+            <version>${powerjob-remote-framework.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webmvc</artifactId>
+            <version>${spring.version}</version>
+            <scope>compile</scope>
+        </dependency>
+
+    </dependencies>
+
+</project>

powerjob-remote/powerjob-remote-impl-http-spring/src/main/java/tech/powerjob/remote/http/HttpProtocol.java

@@ -0,0 +1,17 @@
+package tech.powerjob.remote.http;
+
+import tech.powerjob.remote.framework.transporter.Protocol;
+
+/**
+ * HttpProtocol
+ *
+ * @author tjq
+ * @since 2022/12/31
+ */
+public class HttpProtocol implements Protocol {
+
+    @Override
+    public String name() {
+        return tech.powerjob.common.enums.Protocol.HTTP.name();
+    }
+}

powerjob-remote/powerjob-remote-impl-http-spring/src/main/java/tech/powerjob/remote/http/HttpSpringCSInitializer.java

@@ -0,0 +1,70 @@
+package tech.powerjob.remote.http;
+
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.servlet.mvc.method.RequestMappingInfo;
+import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping;
+import org.springframework.web.util.pattern.PathPatternParser;
+import tech.powerjob.remote.framework.actor.ActorInfo;
+import tech.powerjob.remote.framework.actor.HandlerInfo;
+import tech.powerjob.remote.framework.cs.CSInitializer;
+import tech.powerjob.remote.framework.cs.CSInitializerConfig;
+import tech.powerjob.remote.framework.transporter.Transporter;
+import tech.powerjob.remote.http.spring.SpringMvcTransporter;
+import tech.powerjob.remote.http.spring.SpringUtils;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.List;
+
+/**
+ * @author songyinyin
+ * @since 2023/2/11 19:55
+ */
+public class HttpSpringCSInitializer implements CSInitializer {
+
+  private RequestMappingHandlerMapping requestMappingHandlerMapping;
+
+  @Override
+  public String type() {
+    return tech.powerjob.common.enums.Protocol.HTTP.name();
+  }
+
+  @Override
+  public void init(CSInitializerConfig config) {
+    this.requestMappingHandlerMapping = (RequestMappingHandlerMapping) SpringUtils.getBean("requestMappingHandlerMapping");
+  }
+
+  @Override
+  public Transporter buildTransporter() {
+    return new SpringMvcTransporter();
+  }
+
+  @Override
+  public void bindHandlers(List<ActorInfo> actorInfos) {
+    for (ActorInfo actorInfo : actorInfos) {
+      for (HandlerInfo handlerInfo : actorInfo.getHandlerInfos()) {
+
+        RequestMappingInfo.BuilderConfiguration options = new RequestMappingInfo.BuilderConfiguration();
+        options.setPatternParser(new PathPatternParser());
+        RequestMappingInfo mapping = RequestMappingInfo.paths(handlerInfo.getLocation().toPath())
+            .methods(RequestMethod.POST)
+            // 处理请求的提交内容类型
+//            .consumes(MediaType.APPLICATION_JSON_VALUE)
+            // 返回的内容类型
+            .produces(MediaType.APPLICATION_JSON_VALUE)
+            .options(options)
+            .build();
+        Method method = handlerInfo.getMethod();
+        requestMappingHandlerMapping.registerMapping(mapping, actorInfo.getActor(), method);
+      }
+    }
+  }
+
+  @Override
+  public void close() throws IOException {
+
+  }
+
+
+}

powerjob-remote/powerjob-remote-impl-http-spring/src/main/java/tech/powerjob/remote/http/spring/PowerjobCSMethodProcessor.java

@@ -0,0 +1,54 @@
+package tech.powerjob.remote.http.spring;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.web.bind.support.WebDataBinderFactory;
+import org.springframework.web.context.request.NativeWebRequest;
+import org.springframework.web.method.support.HandlerMethodArgumentResolver;
+import org.springframework.web.method.support.ModelAndViewContainer;
+import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter;
+import org.springframework.web.servlet.mvc.method.annotation.RequestResponseBodyMethodProcessor;
+import tech.powerjob.common.exception.PowerJobException;
+import tech.powerjob.remote.framework.actor.Handler;
+
+import java.util.List;
+
+/**
+ * 带有 @Handler 注解的，接收的请求参数 使用 json 解析，等同于Spring中，在请求参数前使用注解：@RequestBody
+ *
+ * @author songyinyin
+ * @see RequestResponseBodyMethodProcessor
+ * @since 2023/2/12 18:02
+ */
+public class PowerjobCSMethodProcessor implements HandlerMethodArgumentResolver {
+
+  private HandlerMethodArgumentResolver requestResponseBodyMethodProcessor;
+
+  @Override
+  public boolean supportsParameter(MethodParameter parameter) {
+    return parameter.hasMethodAnnotation(Handler.class);
+  }
+
+  @Override
+  public Object resolveArgument(MethodParameter parameter, ModelAndViewContainer mavContainer, NativeWebRequest webRequest, WebDataBinderFactory binderFactory) throws Exception {
+    if (requestResponseBodyMethodProcessor == null) {
+      RequestMappingHandlerAdapter requestMappingHandlerAdapter = SpringUtils.getBean(RequestMappingHandlerAdapter.class);
+      this.requestResponseBodyMethodProcessor = getRequestResponseBodyMethodProcessor(requestMappingHandlerAdapter.getArgumentResolvers());
+    }
+    if (requestResponseBodyMethodProcessor == null) {
+      throw new PowerJobException("requestResponseBodyMethodProcessor is null");
+    }
+    return requestResponseBodyMethodProcessor.resolveArgument(parameter, mavContainer, webRequest, binderFactory);
+  }
+
+  private HandlerMethodArgumentResolver getRequestResponseBodyMethodProcessor(List<HandlerMethodArgumentResolver> resolvers) {
+    if (resolvers == null) {
+      return null;
+    }
+    for (HandlerMethodArgumentResolver resolver : resolvers) {
+      if (resolver instanceof RequestResponseBodyMethodProcessor) {
+        return resolver;
+      }
+    }
+    return null;
+  }
+}

powerjob-remote/powerjob-remote-impl-http-spring/src/main/java/tech/powerjob/remote/http/spring/SpringMvcTransporter.java

@@ -0,0 +1,57 @@
+package tech.powerjob.remote.http.spring;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
+import tech.powerjob.common.PowerSerializable;
+import tech.powerjob.remote.framework.base.RemotingException;
+import tech.powerjob.remote.framework.base.URL;
+import tech.powerjob.remote.framework.transporter.Protocol;
+import tech.powerjob.remote.framework.transporter.Transporter;
+import tech.powerjob.remote.http.HttpProtocol;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+/**
+ * spring-webmvc 使用 RestTemplate 发送http请求，后续兼容 spring-webflux 后，部分请求可以是非阻塞式的
+ *
+ * @author songyinyin
+ * @since 2023/2/12 11:43
+ */
+public class SpringMvcTransporter implements Transporter {
+
+  private static final Protocol PROTOCOL = new HttpProtocol();
+
+  private final RestTemplate restTemplate = new RestTemplate();
+
+  @Override
+  public Protocol getProtocol() {
+    return PROTOCOL;
+  }
+
+  @Override
+  public void tell(URL url, PowerSerializable request) {
+    String fullUrl = getFullUrl(url);
+    restTemplate.postForEntity(fullUrl, request, String.class);
+  }
+
+  @Override
+  public <T> CompletionStage<T> ask(URL url, PowerSerializable request, Class<T> clz) throws RemotingException {
+    String fullUrl = getFullUrl(url);
+    ResponseEntity<T> responseEntity = restTemplate.postForEntity(fullUrl, request, clz);
+    // throw exception
+    final int statusCode = responseEntity.getStatusCodeValue();
+    if (statusCode != HttpStatus.OK.value()) {
+      // CompletableFuture.get() 时会传递抛出该异常
+      throw new RemotingException(String.format("request [url:%s] failed, status: %d, msg: %s",
+          fullUrl, statusCode, responseEntity.getBody()
+      ));
+    }
+    return CompletableFuture.completedFuture(responseEntity.getBody());
+  }
+
+  private String getFullUrl(URL url) {
+    return "http://" + url.getAddress().toFullAddress() + url.getLocation().toPath();
+  }
+}

powerjob-remote/powerjob-remote-impl-http-spring/src/main/java/tech/powerjob/remote/http/spring/SpringUtils.java

@@ -0,0 +1,30 @@
+package tech.powerjob.remote.http.spring;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+
+/**
+ * Spring ApplicationContext 工具类
+ *
+ * @author tjq
+ * @since 2020/4/7
+ */
+public class SpringUtils implements ApplicationContextAware {
+
+    private static ApplicationContext context;
+
+    public static <T> T getBean(Class<T> clz) {
+        return context.getBean(clz);
+    }
+
+    public static Object getBean(String beanName) {
+        return context.getBean(beanName);
+    }
+
+    @Override
+    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
+        context = ctx;
+    }
+}
+

powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/vertx/VertxTransporter.java

@@ -57,7 +57,11 @@ public class VertxTransporter implements Transporter {
         // 获取远程服务器的HTTP连接
         Future<HttpClientRequest> httpClientRequestFuture = httpClient.request(requestOptions);
         // 转换 -> 发送请求获取响应
-        Future<HttpClientResponse> responseFuture = httpClientRequestFuture.compose(httpClientRequest -> httpClientRequest.send(JsonObject.mapFrom(request).toBuffer()));
+        Future<HttpClientResponse> responseFuture = httpClientRequestFuture.compose(httpClientRequest ->
+            httpClientRequest
+                .putHeader("content-type", "application/json")
+                .send(JsonObject.mapFrom(request).toBuffer())
+            );
         return responseFuture.compose(httpClientResponse -> {
             // throw exception
             final int statusCode = httpClientResponse.statusCode();

powerjob-worker-spring-boot-starter/pom.xml

@@ -24,6 +24,21 @@
             <groupId>tech.powerjob</groupId>
             <artifactId>powerjob-worker</artifactId>
             <version>${powerjob.worker.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>tech.powerjob</groupId>
+                    <artifactId>powerjob-remote-impl-http</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>tech.powerjob</groupId>
+                    <artifactId>powerjob-remote-impl-akka</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>tech.powerjob</groupId>
+            <artifactId>powerjob-remote-impl-http-spring</artifactId>
+            <version>${powerjob.worker.version}</version>
         </dependency>
 
         <dependency>

powerjob-worker-spring-boot-starter/src/main/java/tech/powerjob/worker/autoconfigure/PowerJobRemoteAutoConfiguration.java

@@ -0,0 +1,40 @@
+package tech.powerjob.worker.autoconfigure;
+
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.method.support.HandlerMethodArgumentResolver;
+import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
+import tech.powerjob.remote.http.HttpSpringCSInitializer;
+import tech.powerjob.remote.http.spring.PowerjobCSMethodProcessor;
+import tech.powerjob.remote.http.spring.SpringUtils;
+
+import java.util.List;
+
+/**
+ * @author songyinyin
+ * @since 2023/2/12 22:23
+ */
+@Configuration
+@AutoConfigureBefore(PowerJobAutoConfiguration.class)
+@ConditionalOnClass(HttpSpringCSInitializer.class)
+public class PowerJobRemoteAutoConfiguration implements WebMvcConfigurer {
+
+  @Bean
+  public SpringUtils powerJobSpringUtils() {
+    return new SpringUtils();
+  }
+
+  @Bean
+  @ConditionalOnMissingBean
+  public PowerjobCSMethodProcessor powerjobCSMethodProcessor() {
+    return new PowerjobCSMethodProcessor();
+  }
+
+  @Override
+  public void addArgumentResolvers(List<HandlerMethodArgumentResolver> resolvers) {
+    resolvers.add(powerjobCSMethodProcessor());
+  }
+}

powerjob-worker-spring-boot-starter/src/main/resources/META-INF/spring.factories

@@ -1,2 +1,3 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-  tech.powerjob.worker.autoconfigure.PowerJobAutoConfiguration
+  tech.powerjob.worker.autoconfigure.PowerJobAutoConfiguration,\
+  tech.powerjob.worker.autoconfigure.PowerJobRemoteAutoConfiguration