From ad833d36c7cd9d32723a0cb2e63737e71fb50511 Mon Sep 17 00:00:00 2001
From: linwaiwai <jiansihun@foxmail.com>
Date: Mon, 11 Jan 2021 18:42:10 +0800
Subject: [PATCH] add application context and weak ref to input and step,
 implement dubbo aggregation without unit test.

---
 pom.xml                                       | 15 ++++
 src/main/java/we/fizz/Step.java               |  2 +-
 .../java/we/fizz/exception/FizzException.java |  2 -
 src/main/java/we/fizz/input/DubboInput.java   | 70 +++++++++++++++++++
 .../java/we/fizz/input/DubboInputConfig.java  | 69 ++++++++++++++++++
 .../java/we/fizz/input/DubboRPCResponse.java  |  4 ++
 src/main/java/we/fizz/input/InputConfig.java  |  2 +-
 src/main/java/we/fizz/input/InputType.java    |  3 +-
 .../we/fizz/input/RequestRPCResponse.java     | 14 ++++
 .../input/extension/request/RequestInput.java |  4 --
 .../proxy/dubbo/ApacheDubboGenericProxy.java  | 70 +++++++++++++++++++
 .../dubbo/DubboInterfaceDeclaration.java      | 44 ++++++++++++
 src/main/java/we/proxy/dubbo/DubboUtils.java  | 40 +++++++++++
 13 files changed, 330 insertions(+), 9 deletions(-)
 create mode 100644 src/main/java/we/fizz/input/DubboInput.java
 create mode 100644 src/main/java/we/fizz/input/DubboInputConfig.java
 create mode 100644 src/main/java/we/fizz/input/DubboRPCResponse.java
 create mode 100644 src/main/java/we/fizz/input/RequestRPCResponse.java
 create mode 100644 src/main/java/we/proxy/dubbo/ApacheDubboGenericProxy.java
 create mode 100644 src/main/java/we/proxy/dubbo/DubboInterfaceDeclaration.java
 create mode 100644 src/main/java/we/proxy/dubbo/DubboUtils.java

diff --git a/pom.xml b/pom.xml
index fe1bd3f..9398ef2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -41,6 +41,7 @@
 		<netty.version>4.1.58.Final</netty.version>	
 		<httpcore.version>4.4.14</httpcore.version>
 		<log4j2.version>2.13.3</log4j2.version>
+		<apache.dubbo.version>2.7.5</apache.dubbo.version>
 	</properties>
 	
 	<dependencyManagement>
@@ -218,6 +219,20 @@
 		    <artifactId>java-jwt</artifactId>
 		    <version>3.12.1</version>
 		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-inline</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.dubbo</groupId>
+			<artifactId>dubbo</artifactId>
+			<version>${apache.dubbo.version}</version>
+		</dependency>
 
 		<!--<dependency>
 			<groupId>io.netty</groupId>
diff --git a/src/main/java/we/fizz/Step.java b/src/main/java/we/fizz/Step.java
index f690210..55bd5e1 100644
--- a/src/main/java/we/fizz/Step.java
+++ b/src/main/java/we/fizz/Step.java
@@ -64,7 +64,7 @@ public class Step {
 	public void setWeakPipeline(SoftReference<Pipeline> weakPipeline) {
 		this.weakPipeline = weakPipeline;
 	}
-
+	
 	public ConfigurableApplicationContext getCurrentApplicationContext() {
 		return this.getWeakPipeline() != null  ? this.getWeakPipeline().get().getApplicationContext(): null;
 	}
diff --git a/src/main/java/we/fizz/exception/FizzException.java b/src/main/java/we/fizz/exception/FizzException.java
index 92d8710..e18085a 100644
--- a/src/main/java/we/fizz/exception/FizzException.java
+++ b/src/main/java/we/fizz/exception/FizzException.java
@@ -7,6 +7,4 @@ public class FizzException extends Exception {
     public FizzException(String message) {
         super(message);
     }
-
-
 }
diff --git a/src/main/java/we/fizz/input/DubboInput.java b/src/main/java/we/fizz/input/DubboInput.java
new file mode 100644
index 0000000..922e08f
--- /dev/null
+++ b/src/main/java/we/fizz/input/DubboInput.java
@@ -0,0 +1,70 @@
+package we.fizz.input;
+
+import com.alibaba.fastjson.JSON;
+import org.springframework.context.ConfigurableApplicationContext;
+import reactor.core.publisher.Mono;
+import we.constants.CommonConstants;
+import we.proxy.dubbo.ApacheDubboGenericProxy;
+import we.proxy.dubbo.DubboInterfaceDeclaration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.HashMap;
+import java.util.Map;
+/**
+ *
+ * @author linwaiwai
+ *
+ */
+public class DubboInput extends RPCInput {
+    private static final Logger LOGGER = LoggerFactory.getLogger(DubboInput.class);
+
+    @Override
+    protected Mono<RPCResponse> getClientSpecFromContext(InputConfig aConfig, InputContext inputContext) {
+        DubboInputConfig config = (DubboInputConfig) aConfig;
+
+        int timeout = config.getTimeout() < 1 ? 3000 : config.getTimeout() > 10000 ? 10000 : config.getTimeout();
+        Map<String, String> attachments = (Map<String, String>) request.get("attachments");
+        ConfigurableApplicationContext applicationContext = this.getCurrentApplicationContext();
+        String body = (String)request.get("body");
+
+        ApacheDubboGenericProxy proxy = applicationContext.getBean(ApacheDubboGenericProxy.class);
+        DubboInterfaceDeclaration declaration = new DubboInterfaceDeclaration();
+        declaration.setServiceName(config.getServiceName());
+        declaration.setMethod(config.getMethod());
+        declaration.setParameterTypes(config.getParameterTypes());
+        declaration.setTimeout(timeout);
+
+        HashMap<String, String> contextAttachment = new HashMap<String, String>(attachments);
+        contextAttachment.put(CommonConstants.HEADER_TRACE_ID, inputContext.getStepContext().getTraceId());
+        return proxy.send(body, declaration, contextAttachment).flatMap(cr->{
+            DubboRPCResponse response = new DubboRPCResponse();
+            String responseStr = JSON.toJSONString(cr);
+            response.setBodyMono(Mono.just(responseStr));
+            return Mono.just(response);
+        });
+    }
+
+    protected void doRequestMapping(InputConfig aConfig, InputContext inputContext) {
+
+    }
+
+    protected void doOnResponseSuccess(RPCResponse cr, long elapsedMillis) {
+
+    }
+    protected Mono<String> bodyToMono(RPCResponse cr){
+        return cr.getBodyMono();
+    }
+
+    protected void doOnBodyError(Throwable ex, long elapsedMillis) {
+
+    }
+
+    protected void doOnBodySuccess(String resp, long elapsedMillis) {
+
+    }
+
+    protected void doResponseMapping(InputConfig aConfig, InputContext inputContext, String responseBody) {
+    }
+
+
+}
diff --git a/src/main/java/we/fizz/input/DubboInputConfig.java b/src/main/java/we/fizz/input/DubboInputConfig.java
new file mode 100644
index 0000000..4f489f7
--- /dev/null
+++ b/src/main/java/we/fizz/input/DubboInputConfig.java
@@ -0,0 +1,69 @@
+package we.fizz.input;
+
+import org.springframework.util.StringUtils;
+import we.fizz.exception.FizzException;
+import we.fizz.exception.FizzRuntimeException;
+
+import java.util.Map;
+
+public class DubboInputConfig extends InputConfig {
+    private String serviceName;
+    private String method;
+
+    public String getParameterTypes() {
+        return parameterTypes;
+    }
+
+    public void setParameterTypes(String parameterTypes) {
+        this.parameterTypes = parameterTypes;
+    }
+
+    private String parameterTypes;
+
+    public void setServiceName(String serviceName) {
+        this.serviceName = serviceName;
+    }
+
+    public void setMethod(String method) {
+        this.method = method;
+    }
+
+    public String getServiceName() {
+        return serviceName;
+    }
+
+    public String getMethod() {
+        return method;
+    }
+
+
+
+
+
+    public DubboInputConfig(Map configMap)  {
+        super(configMap);
+    }
+    public void parse(){
+        String serviceName = (String) configMap.get("serviceName");
+        if(StringUtils.isEmpty(serviceName)) {
+            throw new FizzRuntimeException("service name can not be blank");
+        }
+        setServiceName(serviceName);
+        String method = (String) configMap.get("method");
+        if (StringUtils.isEmpty(method)) {
+            throw new FizzRuntimeException("method can not be blank");
+        }
+        setMethod(method);
+    }
+
+    private int timeout;
+    public int getTimeout() {
+        return timeout;
+    }
+
+    public void setTimeout(int timeout) {
+        this.timeout = timeout;
+    }
+
+
+}
diff --git a/src/main/java/we/fizz/input/DubboRPCResponse.java b/src/main/java/we/fizz/input/DubboRPCResponse.java
new file mode 100644
index 0000000..2fa8198
--- /dev/null
+++ b/src/main/java/we/fizz/input/DubboRPCResponse.java
@@ -0,0 +1,4 @@
+package we.fizz.input;
+
+public class DubboRPCResponse extends RPCResponse{
+}
diff --git a/src/main/java/we/fizz/input/InputConfig.java b/src/main/java/we/fizz/input/InputConfig.java
index 5df57fb..60eee68 100644
--- a/src/main/java/we/fizz/input/InputConfig.java
+++ b/src/main/java/we/fizz/input/InputConfig.java
@@ -59,7 +59,7 @@ public class InputConfig {
 	public void setFallback(Map<String, String> fallback) {
 		this.fallback = fallback;
 	}
-
+	
 	public void parse(){
 
 	}
diff --git a/src/main/java/we/fizz/input/InputType.java b/src/main/java/we/fizz/input/InputType.java
index b15b7e5..5c7f36b 100644
--- a/src/main/java/we/fizz/input/InputType.java
+++ b/src/main/java/we/fizz/input/InputType.java
@@ -25,12 +25,13 @@ import java.util.Map;
  * @author linwaiwai
  *
  */
+
 public class InputType {
 
     private final String type;
     static private Map<String,InputType > inputs = new HashMap<String,InputType >();
     public InputType(String aType) {
-        this.type = aType;
+         this.type = aType;
         inputs.put(aType, this);
     }
 
diff --git a/src/main/java/we/fizz/input/RequestRPCResponse.java b/src/main/java/we/fizz/input/RequestRPCResponse.java
new file mode 100644
index 0000000..50ef1bd
--- /dev/null
+++ b/src/main/java/we/fizz/input/RequestRPCResponse.java
@@ -0,0 +1,14 @@
+package we.fizz.input;
+
+import org.springframework.http.HttpStatus;
+
+public class RequestRPCResponse extends RPCResponse {
+    private HttpStatus statusCode;
+    public void setStatus(HttpStatus statusCode) {
+        this.statusCode = statusCode;
+    }
+
+    public HttpStatus getStatusCode() {
+        return statusCode;
+    }
+}
diff --git a/src/main/java/we/fizz/input/extension/request/RequestInput.java b/src/main/java/we/fizz/input/extension/request/RequestInput.java
index c115855..e8246c4 100644
--- a/src/main/java/we/fizz/input/extension/request/RequestInput.java
+++ b/src/main/java/we/fizz/input/extension/request/RequestInput.java
@@ -58,11 +58,7 @@ public class RequestInput extends RPCInput implements IInput{
 	static public InputType TYPE = new InputType("REQUEST");
 	private InputType type;
 	protected Map<String, Object> dataMapping;
-	protected Map<String, Object> request = new HashMap<>();
-	protected Map<String, Object> response = new HashMap<>();
 
-	private static final String FALLBACK_MODE_STOP = "stop";
-	private static final String FALLBACK_MODE_CONTINUE = "continue";
 
 	private static final String CONTENT_TYPE_JSON = "application/json";
 	private static final String CONTENT_TYPE_XML = "application/xml";
diff --git a/src/main/java/we/proxy/dubbo/ApacheDubboGenericProxy.java b/src/main/java/we/proxy/dubbo/ApacheDubboGenericProxy.java
new file mode 100644
index 0000000..a15eec8
--- /dev/null
+++ b/src/main/java/we/proxy/dubbo/ApacheDubboGenericProxy.java
@@ -0,0 +1,70 @@
+package we.proxy.dubbo;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.dubbo.config.ApplicationConfig;
+import org.apache.dubbo.config.ReferenceConfig;
+import org.apache.dubbo.config.RegistryConfig;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.service.GenericException;
+import org.apache.dubbo.rpc.service.GenericService;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+import we.fizz.exception.FizzException;
+import we.fizz.input.DubboInputConfig;
+
+import javax.annotation.PostConstruct;
+import java.util.HashMap;
+import java.util.concurrent.CompletableFuture;
+
+@Service
+public class ApacheDubboGenericProxy {
+
+    @PostConstruct
+    public void afterPropertiesSet() {
+
+    }
+
+    public ReferenceConfig<GenericService> createReferenceConfig(String serviceName){
+        ApplicationConfig applicationConfig = new ApplicationConfig();
+        applicationConfig.setName("fizz_proxy");
+        RegistryConfig registryConfig = new RegistryConfig();
+        registryConfig.setAddress("zookeeper://127.0.0.1:2181");
+        ReferenceConfig<GenericService> referenceConfig = new ReferenceConfig<>();
+        referenceConfig.setInterface(serviceName);
+        applicationConfig.setRegistry(registryConfig);
+        referenceConfig.setApplication(applicationConfig);
+        referenceConfig.setGeneric(true);
+        referenceConfig.setAsync(true);
+        referenceConfig.setTimeout(7000);
+        return referenceConfig;
+    }
+
+    /**
+     * Generic invoke.
+     *
+     * @param body     the json string body
+     * @param interfaceDeclaration the interface declaration
+     * @return the object
+     * @throws FizzException the fizz exception
+     */
+    public Mono<Object> send(final String body, final DubboInterfaceDeclaration interfaceDeclaration, HashMap<String, String> attachments ) {
+
+        RpcContext.getContext().setAttachments(attachments);
+        ReferenceConfig<GenericService> reference = createReferenceConfig(interfaceDeclaration.getServiceName());
+        reference.setTimeout(interfaceDeclaration.getTimeout());
+        GenericService genericService = reference.get();
+        Pair<String[], Object[]> pair;
+        if (DubboUtils.isEmpty(body)) {
+            pair = new ImmutablePair<String[], Object[]>(new String[]{}, new Object[]{});
+        } else {
+            pair = DubboUtils.parseDubboParam(body, interfaceDeclaration.getParameterTypes());
+
+        }
+        CompletableFuture<Object> future = genericService.$invokeAsync(interfaceDeclaration.getMethod(), pair.getLeft(), pair.getRight());
+        return Mono.fromFuture(future.thenApply(ret -> {
+            return ret;
+        })).onErrorMap(exception -> exception instanceof GenericException ? new FizzException(((GenericException) exception).getExceptionMessage()) : new FizzException(exception));
+    }
+
+}
diff --git a/src/main/java/we/proxy/dubbo/DubboInterfaceDeclaration.java b/src/main/java/we/proxy/dubbo/DubboInterfaceDeclaration.java
new file mode 100644
index 0000000..f660dc4
--- /dev/null
+++ b/src/main/java/we/proxy/dubbo/DubboInterfaceDeclaration.java
@@ -0,0 +1,44 @@
+package we.proxy.dubbo;
+
+public class DubboInterfaceDeclaration {
+    private String parameterTypes;
+    private String method;
+    private String serviceName;
+    private int timeout;
+
+    public DubboInterfaceDeclaration() {
+    }
+
+
+    public String getParameterTypes() {
+        return parameterTypes;
+    }
+    // call method name
+    public String getMethod() {
+        return method;
+    }
+    // service name
+    public String getServiceName() {
+        return serviceName;
+    }
+
+    public void setServiceName(String serviceName) {
+        this.serviceName = serviceName;
+    }
+
+    public void setMethod(String method) {
+        this.method = method;
+    }
+
+    public void setParameterTypes(String parameterTypes) {
+        this.parameterTypes = parameterTypes;
+    }
+
+    public void setTimeout(int timeout) {
+        this.timeout = timeout;
+    }
+
+    public Integer getTimeout() {
+        return this.timeout;
+    }
+}
diff --git a/src/main/java/we/proxy/dubbo/DubboUtils.java b/src/main/java/we/proxy/dubbo/DubboUtils.java
new file mode 100644
index 0000000..c04dfa1
--- /dev/null
+++ b/src/main/java/we/proxy/dubbo/DubboUtils.java
@@ -0,0 +1,40 @@
+package we.proxy.dubbo;
+
+import com.alibaba.fastjson.JSON;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+public class DubboUtils {
+    public static boolean isEmpty(final String body) {
+        return null == body || "".equals(body) || "{}".equals(body) || "null".equals(body);
+    }
+    /*
+     * body json string
+     */
+    public static Pair<String[], Object[]> parseDubboParam(String body, final String parameterTypes) {
+
+        Map<String, Object> paramMap = (Map<String,Object>) JSON.parse(body);
+        String[] parameter = StringUtils.split(parameterTypes, ',');
+        if (parameter.length == 1 && !isBaseType(parameter[0])) {
+            return new ImmutablePair<>(parameter, new Object[]{paramMap});
+        }
+        List<Object> list = new LinkedList<>();
+        for (String key : paramMap.keySet()) {
+            Object obj = paramMap.get(key);
+            if (obj != null) {
+                list.add(obj);
+            }
+        }
+        Object[] objects = list.toArray();
+        return new ImmutablePair<>(parameter, objects);
+    }
+
+    private static boolean isBaseType(String type) {
+        return type.startsWith("java") || type.startsWith("[Ljava");
+    }
+}