refactor RPC and Registry

Author: kimmking

07rpc/rpc01/client-rest.http

@@ -0,0 +1 @@
+http://127.0.0.1:8080/api/hello

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/annotation/RpcfxReference.java

@@ -10,7 +10,7 @@
  */
 @Documented
 @Retention(RetentionPolicy.RUNTIME)
-@Target(ElementType.TYPE)
+@Target(ElementType.FIELD)
 @Inherited
 public @interface RpcfxReference {
 

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/api/RpcContext.java

@@ -0,0 +1,25 @@
+package io.kimmking.rpcfx.api;
+
+import io.kimmking.rpcfx.meta.ProviderMeta;
+import lombok.Getter;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Description for this class.
+ *
+ * @Author : kimmking(kimmking@apache.org)
+ * @create 2024/1/13 20:34
+ */
+public class RpcContext {
+
+    @Getter
+    private MultiValueMap<String, ProviderMeta> providerHolder = new LinkedMultiValueMap<>();
+
+    @Getter
+    private Map<String, Object> consumerHolder = new HashMap<>();
+
+}

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/api/RpcfxResolver.java

@@ -0,0 +1,60 @@
+package io.kimmking.rpcfx.consumer;
+
+import io.kimmking.rpcfx.annotation.RpcfxReference;
+import io.kimmking.rpcfx.api.RpcContext;
+import io.kimmking.rpcfx.stub.StubSkeletonHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.PropertyValues;
+import org.springframework.beans.factory.config.InstantiationAwareBeanPostProcessor;
+import org.springframework.stereotype.Component;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Description for this class.
+ *
+ * @Author : kimmking(kimmking@apache.org)
+ * @create 2024/1/13 23:26
+ */
+@Slf4j
+@Component
+public class ConsumerBootstrap implements Closeable, InstantiationAwareBeanPostProcessor {
+
+    private RpcContext rpcContext = new RpcContext();
+
+    private String scanPackage = "io.kimmking";
+
+    @Override
+    public void close() throws IOException {
+
+    }
+
+    @Override
+    public PropertyValues postProcessProperties(PropertyValues pvs, Object bean, String beanName) throws BeansException {
+        if (bean.getClass().getPackage().getName().startsWith(scanPackage)) {
+            Field[] declaredFields = bean.getClass().getDeclaredFields();
+            List<Field> consumers = Arrays.stream(declaredFields).filter(field -> field.isAnnotationPresent(RpcfxReference.class)).collect(Collectors.toList());
+
+            consumers.stream().forEach(consumer -> {
+                Object consumer1 = createConsumer(consumer.getType());
+                try {
+                    consumer.setAccessible(true);
+                    consumer.set(bean, consumer1);
+                } catch (IllegalAccessException e) {
+                    log.error(e.getMessage(), e);
+                }
+            });
+        }
+        return null;
+    }
+
+    private <T> T createConsumer(Class<T> clazz) {
+        return StubSkeletonHelper.createConsumer(clazz, rpcContext);
+    }
+}

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/consumer/ConsumerBootstrap.java

@@ -1,7 +1,8 @@
-package io.kimmking.rpcfx.client;
+package io.kimmking.rpcfx.consumer;
 
 import com.alibaba.fastjson.JSON;
 import io.kimmking.rpcfx.api.*;
+import io.kimmking.rpcfx.stub.StubSkeletonHelper;
 import okhttp3.*;
 
 import java.io.IOException;
@@ -35,6 +36,10 @@ public <T> RpcfxInvocationHandler(Class<T> serviceClass, List<String> invokers,
     @Override
     public Object invoke(Object proxy, Method method, Object[] params) throws Throwable {
 
+        if (!StubSkeletonHelper.checkRpcMethod(method)){
+            return null ;
+        }
+
         List<String> urls = router.route(invokers);
 //            System.out.println("router.route => ");
 //            urls.forEach(System.out::println);
@@ -85,9 +90,6 @@ private RpcfxResponse post(RpcfxRequest req, String url) throws IOException {
         String reqJson = JSON.toJSONString(req);
 //            System.out.println("req json: "+reqJson);
 
-        // 1.可以复用client
-        // 2.尝试使用httpclient或者netty client
-
         final Request request = new Request.Builder()
                 .url(url)
                 .post(RequestBody.create(JSONTYPE, reqJson))

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/client/RpcfxInvocationHandler.java renamed to 07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/consumer/RpcfxInvocationHandler.java

@@ -1,54 +1,50 @@
-package io.kimmking.rpcfx.client;
+package io.kimmking.rpcfx.consumer;
 
 
-import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.parser.ParserConfig;
 import io.kimmking.rpcfx.api.*;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
-import okhttp3.RequestBody;
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.cache.TreeCache;
 import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
 import org.apache.curator.framework.recipes.cache.TreeCacheListener;
 import org.apache.curator.retry.ExponentialBackoffRetry;
-import org.apache.zookeeper.CreateMode;
 
-import java.io.IOException;
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
-import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.List;
 
-public final class Rpcfx {
+public final class RpcfxInvoker {
 
     static {
         ParserConfig.getGlobalInstance().addAccept("io.kimmking");
     }
+    CuratorFramework client;
+    String zkUrl = null;
 
-    public static <T, filters> T createFromRegistry(final Class<T> serviceClass, final String zkUrl, Router router, LoadBalancer loadBalance, Filter filter) {
+    public RpcfxInvoker(String zkUrl) {
+        this.zkUrl = zkUrl; //"localhost:2181"
+        this.start();
+    }
+
+    public void start() {
+        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
+        client = CuratorFrameworkFactory.builder().connectString(this.zkUrl).namespace("rpcfx").retryPolicy(retryPolicy).build();
+        client.start();
+    }
+
+    public void stop() {
+        client.close();
+    }
 
-        // 加filte之一
+    public <T> T createFromRegistry(final Class<T> serviceClass, Router router, LoadBalancer loadBalance, Filter filter) {
 
         String service = serviceClass.getCanonicalName();//"io.kimking.rpcfx.demo.api.UserService";
         System.out.println("====> "+service);
         List<String> invokers = new ArrayList<>();
 
-        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
-        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181").namespace("rpcfx").retryPolicy(retryPolicy).build();
-        client.start();
-
         try {
-//        ServiceProviderDesc userServiceSesc = ServiceProviderDesc.builder()
-//                .host(InetAddress.getLocalHost().getHostAddress())
-//                .port(8082).serviceClass(service).build();
-        // String userServiceSescJson = JSON.toJSONString(userServiceSesc);
-
 
             if ( null == client.checkExists().forPath("/" + service)) {
                 return null;
@@ -69,25 +65,13 @@ public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCac
             ex.printStackTrace();
         }
 
-//
-//
-//		// register service
-//		// xxx "io.kimmking.rpcfx.demo.api.UserService"
-//
-
-
-        // curator Provider list from zk
-
-        // 1. 简单：从zk拿到服务提供的列表
-        // 2. 挑战：监听zk的临时节点，根据事件更新这个list（注意，需要做个全局map保持每个服务的提供者List）
-
         return (T) create(serviceClass, invokers, router, loadBalance, filter);
 
     }
 
 
 
-    private static void fetchInvokers(CuratorFramework client, String service, List<String> invokers) throws Exception {
+    private void fetchInvokers(CuratorFramework client, String service, List<String> invokers) throws Exception {
         List<String> services = client.getChildren().forPath("/" + service);
         invokers.clear();
         for (String svc : services) {
@@ -97,10 +81,11 @@ private static void fetchInvokers(CuratorFramework client, String service, List<
         }
     }
 
-    private static <T> Object create(Class<T> serviceClass, List<String> invokers, Router router, LoadBalancer loadBalance, Filter... filters) {
-        // 0. 替换动态代理 -> 字节码生成
-        return (T) Proxy.newProxyInstance(Rpcfx.class.getClassLoader(), new Class[]{serviceClass},
-                new RpcfxInvocationHandler(serviceClass, invokers, router, loadBalance, filters));
+    private <T> T create(Class<T> serviceClass, List<String> invokers, Router router, LoadBalancer loadBalance, Filter... filters) {
+        RpcfxInvocationHandler invocationHandler
+                = new RpcfxInvocationHandler(serviceClass, invokers, router, loadBalance, filters);
+        return (T) Proxy.newProxyInstance(RpcfxInvoker.class.getClassLoader(),
+                new Class[]{serviceClass}, invocationHandler);
     }
 
 }

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/client/Rpcfx.java renamed to 07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/consumer/RpcfxInvoker.java

@@ -0,0 +1,10 @@
+package io.kimmking.rpcfx.discovery;
+
+/**
+ * Description for this class.
+ *
+ * @Author : kimmking(kimmking@apache.org)
+ * @create 2024/1/13 20:17
+ */
+public class DiscoveryClient {
+}

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/discovery/DiscoveryClient.java

@@ -0,0 +1,19 @@
+package io.kimmking.rpcfx.meta;
+
+import lombok.Data;
+
+import java.lang.reflect.Method;
+
+/**
+ * @author lirui
+ */
+@Data
+public class ProviderMeta {
+
+    private Object serviceImpl;
+
+    private Method method;
+
+    private String methodSign;
+
+}