[ISSUE apache#367] Refactor connector

    1.support load plugin from eventMeshPluginDir
    2.remove connector plugin from runtime

build.gradle

@@ -305,6 +305,7 @@ subprojects {
             new File(projectDir, '../dist/apps').mkdirs()
             new File(projectDir, '../dist/conf').mkdirs()
             new File(projectDir, '../dist/lib').mkdirs()
+            new File(projectDir, '../dist/plugin/connector').mkdirs()
         }
 
         doLast {
@@ -313,6 +314,7 @@ subprojects {
                 from project.jar.getArchivePath()
                 exclude 'eventmesh-common*.jar'
                 exclude 'eventmesh-connector-api*.jar'
+                exclude 'eventmesh-connector-plugin*.jar'
                 exclude 'eventmesh-starter*.jar'
                 exclude 'eventmesh-test*.jar'
                 exclude 'eventmesh-sdk*.jar'
@@ -335,6 +337,11 @@ subprojects {
                 exclude 'commons-collections-3.2.2.jar'
             }
 
+            copy {
+                into '../dist/plugin/connector'
+                from '../eventmesh-connector-plugin/dist/apps'
+            }
+
             copy {
                 into '../dist/bin'
                 from '../eventmesh-runtime/bin'

eventmesh-connector-plugin/build.gradle

@@ -0,0 +1,16 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */

eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/config/ConfigurationWrapper.java → eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/config/ConfigurationWrapper.java

@@ -73,4 +73,4 @@ private void load() {
     public String getProp(String key) {
         return StringUtils.isEmpty(key) ? null : properties.getProperty(key, null);
     }
-}
+}

eventmesh-connector-rocketmq/src/test/java/org/apache/rocketmq/consumer/PushConsumerImplTest.java → eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/test/java/org/apache/rocketmq/consumer/PushConsumerImplTest.java

@@ -23,7 +23,6 @@
 import java.util.Collections;
 import java.util.Properties;
 
-import io.openmessaging.api.Action;
 import io.openmessaging.api.AsyncConsumeContext;
 import io.openmessaging.api.AsyncMessageListener;
 import io.openmessaging.api.Consumer;

eventmesh-connector-rocketmq/src/main/resources/META-INF/services/org.apache.eventmesh.api.consumer.MeshMQPushConsumer → eventmesh-connector-plugin/gradle.properties

@@ -1,3 +1,4 @@
+#
 # Licensed to the Apache Software Foundation (ASF) under one or more
 # contributor license agreements.  See the NOTICE file distributed with
 # this work for additional information regarding copyright ownership.
@@ -12,5 +13,9 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
-org.apache.eventmesh.connector.rocketmq.consumer.RocketMQConsumerImpl
+#
+group=org.apache.eventmesh
+version=1.2.0-SNAPSHOT
+jdk=1.8
+mavenUserName=
+mavenPassword=

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQConsumerWrapper.java

@@ -19,7 +19,6 @@
 
 import java.util.List;
 import java.util.Properties;
-import java.util.ServiceLoader;
 
 import io.openmessaging.api.AsyncMessageListener;
 import io.openmessaging.api.Message;
@@ -52,24 +51,11 @@ public void unsubscribe(String topic) throws Exception {
     }
 
     public synchronized void init(Properties keyValue) throws Exception {
-        meshMQPushConsumer = getMeshMQPushConsumer();
-        if (meshMQPushConsumer == null) {
-            logger.error("can't load the meshMQPushConsumer plugin, please check.");
-            throw new RuntimeException("doesn't load the meshMQPushConsumer plugin, please check.");
-        }
 
         meshMQPushConsumer.init(keyValue);
         inited.compareAndSet(false, true);
     }
 
-    private MeshMQPushConsumer getMeshMQPushConsumer() {
-        ServiceLoader<MeshMQPushConsumer> meshMQPushConsumerServiceLoader = ServiceLoader.load(MeshMQPushConsumer.class);
-        if (meshMQPushConsumerServiceLoader.iterator().hasNext()) {
-            return meshMQPushConsumerServiceLoader.iterator().next();
-        }
-        return null;
-    }
-
     public synchronized void start() throws Exception {
         meshMQPushConsumer.start();
         started.compareAndSet(false, true);

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQProducerWrapper.java

@@ -18,7 +18,6 @@
 package org.apache.eventmesh.runtime.core.plugin;
 
 import java.util.Properties;
-import java.util.ServiceLoader;
 
 import io.openmessaging.api.Message;
 import io.openmessaging.api.SendCallback;
@@ -47,24 +46,10 @@ public synchronized void init(Properties keyValue) throws Exception {
             return;
         }
 
-        meshMQProducer = getSpiMeshMQProducer();
-        if (meshMQProducer == null) {
-            logger.error("can't load the meshMQProducer plugin, please check.");
-            throw new RuntimeException("doesn't load the meshMQProducer plugin, please check.");
-        }
-
         meshMQProducer.init(keyValue);
         inited.compareAndSet(false, true);
     }
 
-    private MeshMQProducer getSpiMeshMQProducer() {
-        ServiceLoader<MeshMQProducer> meshMQProducerServiceLoader = ServiceLoader.load(MeshMQProducer.class);
-        if (meshMQProducerServiceLoader.iterator().hasNext()) {
-            return meshMQProducerServiceLoader.iterator().next();
-        }
-        return null;
-    }
-
     public synchronized void start() throws Exception {
         if (started.get()) {
             return;

eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionFactory.java

@@ -18,11 +18,43 @@
 package org.apache.eventmesh.spi;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.spi.loader.ExtensionClassLoader;
+import org.apache.eventmesh.spi.loader.JarExtensionClassLoader;
+import org.apache.eventmesh.spi.loader.MetaInfExtensionClassLoader;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * The extension fetching factory, all extension plugins should be fetched by this factory.
+ * And all the extension plugins defined in eventmesh should have {@link EventMeshSPI} annotation.
+ */
 public enum EventMeshExtensionFactory {
     ;
 
+    private static final Logger logger = LoggerFactory.getLogger(EventMeshExtensionFactory.class);
+
+    private static final List<ExtensionClassLoader> extensionClassLoaders = new ArrayList<>();
+
+    static {
+        extensionClassLoaders.add(new MetaInfExtensionClassLoader());
+        extensionClassLoaders.add(new JarExtensionClassLoader());
+    }
+
+    private static final ConcurrentHashMap<String, Object> EXTENSION_INSTANCE_CACHE =
+            new ConcurrentHashMap<>(16);
+
+    /**
+     * @param extensionType extension plugin class type
+     * @param extensionName extension instance name
+     * @param <T>           the type of the plugin
+     * @return plugin instance
+     */
+    @SuppressWarnings("unchecked")
     public static <T> T getExtension(Class<T> extensionType, String extensionName) {
         if (extensionType == null) {
             throw new ExtensionException("extensionType is null");
@@ -33,6 +65,29 @@ public static <T> T getExtension(Class<T> extensionType, String extensionName) {
         if (!extensionType.isInterface() || !extensionType.isAnnotationPresent(EventMeshSPI.class)) {
             throw new ExtensionException(String.format("extensionType:%s is invalided", extensionType));
         }
-        return EventMeshExtensionLoader.getExtension(extensionType, extensionName);
+        if (!EXTENSION_INSTANCE_CACHE.containsKey(extensionName)) {
+            synchronized (EventMeshExtensionFactory.class) {
+                initializeExtension(extensionType, extensionName);
+            }
+        }
+        return (T) EXTENSION_INSTANCE_CACHE.get(extensionName);
+    }
+
+    private static <T> void initializeExtension(Class<T> extensionType, String extensionName) {
+        for (ExtensionClassLoader extensionClassLoader : extensionClassLoaders) {
+            Map<String, Class<?>> extensionClassMap = extensionClassLoader.loadExtensionClass(extensionType);
+            Class<?> instanceClass = extensionClassMap.get(extensionName);
+            if (instanceClass != null) {
+                try {
+                    Object extensionObj = instanceClass.newInstance();
+                    logger.info("initialize extension instance success, extensionType: {}, extensionName: {}",
+                            extensionType, extensionName);
+                    EXTENSION_INSTANCE_CACHE.put(extensionName, extensionObj);
+                } catch (InstantiationException | IllegalAccessException e) {
+                    throw new ExtensionException("Extension initialize error", e);
+                }
+            }
+        }
     }
+
 }

eventmesh-spi/src/main/java/org/apache/eventmesh/spi/loader/ExtensionClassLoader.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.spi.loader;
+
+import java.util.Map;
+
+/**
+ * Load extension class
+ * <ul>
+ *     <li>{@link MetaInfExtensionClassLoader}</li>
+ *     <li>{@link JarExtensionClassLoader}</li>
+ * </ul>
+ */
+public interface ExtensionClassLoader {
+
+    /**
+     * load
+     *
+     * @param extensionType extension type class
+     * @param <T>           extension type
+     * @return extension instance name to extension instance class
+     */
+    <T> Map<String, Class<?>> loadExtensionClass(Class<T> extensionType);
+}