Resolve conflicts with storage-api branch

Author: githublaohu

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/build.gradle

@@ -23,10 +23,16 @@ dependencies {
     api "io.dropwizard.metrics:metrics-healthchecks"
     api "io.dropwizard.metrics:metrics-annotation"
     api "io.dropwizard.metrics:metrics-json"
+    
+    implementation 'io.cloudevents:cloudevents-json-jackson:2.4.0'
 
     compileOnly 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'
 
+	implementation "org.mockito:mockito-core"
+    implementation "org.powermock:powermock-module-junit4"
+    implementation "org.powermock:powermock-api-mockito2"
+	
     testCompileOnly 'org.projectlombok:lombok'
     testAnnotationProcessor 'org.projectlombok:lombok'
 

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/gradle.properties

@@ -13,4 +13,6 @@
 # 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.
-#
+#
+pluginType=connector
+pluginName=storage

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/CloudEventUtils.java

@@ -20,24 +20,31 @@
 import org.apache.eventmesh.api.connector.storage.data.CloudEventInfo;
 
 import java.lang.reflect.Field;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
 import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventFormat;
 import io.cloudevents.core.impl.BaseCloudEvent;
+import io.cloudevents.core.provider.EventFormatProvider;
 import io.cloudevents.core.v03.CloudEventV03;
 import io.cloudevents.core.v1.CloudEventV1;
 
 public class CloudEventUtils {
+	
+	public static EventFormat eventFormat = EventFormatProvider.getInstance().resolveFormat("application/cloudevents+json");
 
     private static Field CLOUD_EVENT_EXTENSIONS_FIELD;
 
     static {
         try {
-            CLOUD_EVENT_EXTENSIONS_FIELD = BaseCloudEvent.class.getField("extensions");
+            CLOUD_EVENT_EXTENSIONS_FIELD = BaseCloudEvent.class.getDeclaredField("extensions");
             CLOUD_EVENT_EXTENSIONS_FIELD.setAccessible(true);
         } catch (NoSuchFieldException | SecurityException e) {
-            e.printStackTrace();
+            throw new RuntimeException(e.getMessage() , e );
         }
     }
 
@@ -50,22 +57,48 @@ public static CloudEvent setValue(CloudEvent cloudEvent, String key, Object valu
                 extensions.put(key, value);
                 return cloudEvent;
             } catch (Exception e) {
-                e.printStackTrace();
+                throw new RuntimeException(e);
             }
         }
         return null;
     }
+    
+	public static String getTableName(CloudEvent cloudEvent) {
+		return cloudEvent.getSubject();
+	}
+	
+	public static String checkConsumerGroupName(String topic) {
+		return topic.replace("-", "_");
+	}
+
+	public static List<Object> getParameterToCloudEvent(CloudEvent cloudEvent) {
+		List<Object> parameterList = new ArrayList<>();
+		String id = (String) cloudEvent.getExtension("cloudeventid");
+		parameterList.add(Objects.isNull(id)?"1":id);// id
+		parameterList.add(getTableName(cloudEvent));// topic
+		parameterList.add("");// cloud_event_storage_node_adress
+		parameterList.add("");// cloud_event_type
+		parameterList.add("");// cloud_event_producer_group_name
+		parameterList.add("");// cloud_event_source
+		parameterList.add("application/cloudevents+json");// cloud_event_content_type
+		// parameterList.add("");//cloud_event_tag
+		parameterList.add("");// cloud_event_extensions
+		String data = new String(CloudEventUtils.eventFormat.serialize(cloudEvent), Charset.forName("UTF-8"));
+		parameterList.add(data);// cloud_event_data
+		parameterList.add("{}");
+		return parameterList;
+	}
 
     public static String getNodeAdress(CloudEvent cloudEvent) {
-        return (String) cloudEvent.getExtension(Constant.NODE_ADDRESS);
+        return (String) cloudEvent.getExtension(Constant.STORAGE_CONFIG_ADDRESS);
     }
 
     public static String getTopic(CloudEvent cloudEvent) {
-        return null;
+        return cloudEvent.getSubject();
     }
 
     public static String getId(CloudEvent cloudEvent) {
-        return null;
+        return "";
     }
 
     public static CloudEvent createCloudEvent(CloudEventInfo cloudEventInfo) {

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/StorageConnectorInfo.java renamed to eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/ConnectorResourceServiceStorageImpl.java

@@ -14,13 +14,22 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.eventmesh.api.connector.storage;
 
-import lombok.Data;
+import org.apache.eventmesh.api.connector.ConnectorResourceService;
+
+public class ConnectorResourceServiceStorageImpl implements ConnectorResourceService{
+
+	@Override
+	public void init() throws Exception {
+		// TODO Auto-generated method stub
+		
+	}
 
-@Data
-public class StorageConnectorInfo {
+	@Override
+	public void release() throws Exception {
+		// TODO Auto-generated method stub
+		
+	}
 
-	private boolean distinguishTopic;
 }

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/Constant.java

@@ -19,5 +19,24 @@
 
 public class Constant {
 
-	public static final String NODE_ADDRESS = "nodeAddress";
+	public static final String STORAGE_CONFIG_ADDRESS = "eventMesh.connector.plugin.storage.nodeaddress";
+	
+	public static final String STORAGE_CONFIG_TYPE = "eventMesh.connector.plugin.storage.type";
+
+	public static final String STORAGE_CONFIG_JDBC_PARAMETER = "eventMesh.connector.plugin.storage.jdbc.parameter";
+
+	public static final String STORAGE_CONFIG_USER_NAME = "eventMesh.connector.plugin.storage.username";
+
+	public static final String STORAGE_CONFIG_PASSWORD = "eventMesh.connector.plugin.storage.password";
+	
+	public static final String STORAGE_CONFIG_JDBC_TYPE = "eventMesh.connector.plugin.storage.jdbc.dbType";
+	
+	public static final String STORAGE_CONFIG_JDBC_MAXACTIVE = "eventMesh.connector.plugin.storage.jdbc.maxActive";
+	
+	public static final String STORAGE_CONFIG_JDBC_MAXWAIT = "eventMesh.connector.plugin.storage.jdbc.maxWait";
+	
+	public static final String STORAGE_ID = "storageid";
+	
+	public static final String STORAGE_NODE_ADDRESS = "address";
+
 }

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/StorageConfig.java

@@ -22,7 +22,7 @@
 @Data
 public class StorageConfig {
 
-    private long pullInterval;
+    private long pullInterval = 200;
 
-    private long pullThresholdForQueue;
+    private long pullThresholdForQueue = 5;
 }

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/StorageConnector.java

@@ -61,6 +61,10 @@ public interface StorageConnector extends LifeCycle {
     public List<CloudEvent> pull(PullRequest pullRequest) throws Exception;
 
     void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context);
+    
+    public default String getTopic(String topic) {
+    	return topic;
+    }
 
     public default int deleteCloudEvent(CloudEvent cloudEvent) {
         return 0;

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/StorageConnectorMetedata.java

@@ -35,5 +35,7 @@ public interface StorageConnectorMetedata {
     public int createTopic(TopicInfo topicInfo) throws Exception;
 
     public int createConsumerGroupInfo(ConsumerGroupInfo consumerGroupInfo) throws Exception;
+    
+    public List<TopicInfo> geTopicInfos(Set<String> topics,String key) throws Exception;
 
 }

Diff for: eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/storage/StorageConnectorProxy.java

@@ -24,18 +24,21 @@
 import org.apache.eventmesh.api.connector.storage.data.TopicInfo;
 import org.apache.eventmesh.api.connector.storage.metadata.RouteHandler;
 import org.apache.eventmesh.api.connector.storage.metadata.StorageMetaServcie;
+import org.apache.eventmesh.api.connector.storage.reply.ReplyOperation;
 import org.apache.eventmesh.api.connector.storage.reply.ReplyOperationService;
 import org.apache.eventmesh.api.connector.storage.reply.RequestReplyInfo;
 import org.apache.eventmesh.api.exception.ConnectorRuntimeException;
 import org.apache.eventmesh.api.exception.OnExceptionContext;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 
 import io.cloudevents.CloudEvent;
+import lombok.Setter;
 
 public class StorageConnectorProxy implements StorageConnector {
 
@@ -45,10 +48,13 @@ public class StorageConnectorProxy implements StorageConnector {
 
     private RouteHandler routeHandler = new RouteHandler();
 
+    @Setter
     private ReplyOperationService replyService;
 
+    @Setter
     private StorageMetaServcie storageMetaServcie;
 
+    @Setter
     private Executor executor;
 
     @Override
@@ -63,7 +69,12 @@ public void shutdown() {
     public void init(Properties properties) throws Exception {
     }
 
+    public  Collection<StorageConnector> getStorageConnectorList(){
+    	return this.storageConnectorByKeyMap.values();
+    }
+    
     public void setConnector(StorageConnector storageConnector, String key) {
+    	routeHandler.addStorageConnector(storageConnector);
         storageConnectorMap.put(storageConnector, key);
         storageConnectorByKeyMap.put(key, storageConnector);
     }
@@ -84,6 +95,7 @@ private void doPublish(CloudEvent cloudEvent, SendCallback sendCallback) {
             if (storageConnector instanceof StorageConnectorMetedata
                 && !storageMetaServcie.isTopic(storageConnector, CloudEventUtils.getTopic(cloudEvent))) {
                 TopicInfo topicInfo = new TopicInfo();
+                topicInfo.setTopicName(CloudEventUtils.getTopic(cloudEvent));
                 StorageConnectorMetedata storageConnectorMetedata = (StorageConnectorMetedata) storageConnector;
                 storageConnectorMetedata.createTopic(topicInfo);
             }
@@ -114,15 +126,18 @@ public void run() {
     public void doRequest(CloudEvent cloudEvent, RequestReplyCallback requestReplyCallback, long timeout) {
         try {
             StorageConnector storageConnector = routeHandler.select();
+            if(!(storageConnector instanceof ReplyOperation)) {
+            	return;
+            }
             String key = storageConnectorMap.get(storageConnector);
-            CloudEventUtils.setValue(cloudEvent, "nodeAddress", key);
+            CloudEventUtils.setValue(cloudEvent, Constant.STORAGE_CONFIG_ADDRESS, key);
             storageConnector.request(cloudEvent, requestReplyCallback, timeout);
-            Long storageId = (Long) cloudEvent.getExtension("storageId");
+            Long storageId = Long.valueOf( cloudEvent.getExtension(Constant.STORAGE_ID).toString());
             RequestReplyInfo requestReplyInfo = new RequestReplyInfo();
             requestReplyInfo.setStorageId(storageId);
             requestReplyInfo.setTimeOut(System.currentTimeMillis() + timeout);
             requestReplyInfo.setRequestReplyCallback(requestReplyCallback);
-            replyService.setRequestReplyInfo(null, cloudEvent.getType(), storageId, requestReplyInfo);
+            replyService.setRequestReplyInfo((ReplyOperation)storageConnector, cloudEvent.getType(), storageId, requestReplyInfo);
         } catch (Exception e) {
             requestReplyCallback.onException(e);
         }