[ISSUE #5208] develop mcp protocol

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/mcp/McpRetryConfig.java

@@ -0,0 +1,33 @@
+/*
+ * 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.common.config.connector.mcp;
+
+import lombok.Data;
+
+@Data
+public class McpRetryConfig {
+    // maximum number of retries, default 2, minimum 0
+    private int maxRetries = 2;
+
+    // retry interval, default 1000ms
+    private int interval = 1000;
+
+    // Default value is false, indicating that only requests with network-level errors will be retried.
+    // If set to true, all failed requests will be retried, including network-level errors and non-2xx responses.
+    private boolean retryOnNonSuccess = false;
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/mcp/McpSinkConfig.java

@@ -0,0 +1,31 @@
+/*
+ * 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.common.config.connector.mcp;
+
+import org.apache.eventmesh.common.config.connector.SinkConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class McpSinkConfig extends SinkConfig {
+
+    public SinkConnectorConfig connectorConfig;
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/mcp/McpSourceConfig.java

@@ -0,0 +1,31 @@
+/*
+ * 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.common.config.connector.mcp;
+
+import org.apache.eventmesh.common.config.connector.SourceConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class McpSourceConfig extends SourceConfig {
+
+    public SourceConnectorConfig connectorConfig;
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/mcp/SinkConnectorConfig.java

@@ -0,0 +1,76 @@
+/*
+ * 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.common.config.connector.mcp;
+
+
+import lombok.Data;
+
+@Data
+public class SinkConnectorConfig {
+
+    private String connectorName;
+
+    private String[] urls;
+
+    // keepAlive, default true
+    private boolean keepAlive = true;
+
+    // timeunit: ms, default 60000ms
+    private int keepAliveTimeout = 60 * 1000; // Keep units consistent
+
+    // timeunit: ms, default 5000ms, recommended scope: 5000ms - 10000ms
+    private int connectionTimeout = 5000;
+
+    // timeunit: ms, default 5000ms
+    private int idleTimeout = 5000;
+
+    // maximum number of HTTP/1 connections a client will pool, default 50
+    private int maxConnectionPoolSize = 50;
+
+    // retry config
+    private McpRetryConfig retryConfig = new McpRetryConfig();
+
+    private String deliveryStrategy = "ROUND_ROBIN";
+
+    private boolean skipDeliverException = false;
+
+    // managed pipelining param, default true
+    private boolean isParallelized = true;
+
+    private int parallelism = 2;
+
+
+    /**
+     * Fill default values if absent (When there are multiple default values for a field)
+     *
+     * @param config SinkConnectorConfig
+     */
+    public static void populateFieldsWithDefaults(SinkConnectorConfig config) {
+        /*
+         * set default values for idleTimeout
+         * recommended scope: common(5s - 10s), webhook(15s - 30s)
+         */
+        final int commonHttpIdleTimeout = 5000;
+
+        // Set default values for idleTimeout
+        if (config.getIdleTimeout() == 0) {
+            config.setIdleTimeout(commonHttpIdleTimeout);
+        }
+
+    }
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/mcp/SourceConnectorConfig.java

@@ -0,0 +1,63 @@
+/*
+ * 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.common.config.connector.mcp;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import lombok.Data;
+
+@Data
+public class SourceConnectorConfig {
+
+    private String connectorName;
+
+    private String path = "/";
+
+    private int port;
+
+    // timeunit: ms, default 5000ms
+    private int idleTimeout = 5000;
+
+    /**
+     * <ul>
+     *     <li>The maximum size allowed for form attributes when Content-Type is application/x-www-form-urlencoded or multipart/form-data </li>
+     *     <li>Default is 1MB (1024 * 1024 bytes). </li>
+     *     <li>If you receive a "size exceed allowed maximum capacity" error, you can increase this value. </li>
+     *     <li>Note: This applies only when handling form data submissions.</li>
+     * </ul>
+     */
+    private int maxFormAttributeSize = 1024 * 1024;
+
+    // max size of the queue, default 1000
+    private int maxStorageSize = 1000;
+
+    // batch size, default 10
+    private int batchSize = 10;
+
+    // protocol, default CloudEvent
+    private String protocol = "Mcp";
+
+    // extra config, e.g. GitHub secret
+    private Map<String, String> extraConfig = new HashMap<>();
+
+    // data consistency enabled, default true
+    private boolean dataConsistencyEnabled = false;
+
+    private String forwardPath;
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/JsonUtils.java

@@ -162,6 +162,17 @@ public static <T> T parseObject(byte[] bytes, Class<T> clazz) {
         }
     }
 
+    public static <T> T parseObject(String text, TypeReference<T> typeReference) {
+        if (StringUtils.isEmpty(text)) {
+            return null;
+        }
+        try {
+            return OBJECT_MAPPER.readValue(text, typeReference);
+        } catch (JsonProcessingException e) {
+            throw new JsonException("deserialize json string to object error", e);
+        }
+    }
+
     /**
      * parse json string to object.
      *

eventmesh-connectors/eventmesh-connector-http/src/main/resources/server-config.yml

@@ -16,4 +16,4 @@
 #
 
 sourceEnable: true
-sinkEnable: false
+sinkEnable: true

eventmesh-connectors/eventmesh-connector-mcp/build.gradle

@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+dependencies {
+    api project(":eventmesh-openconnect:eventmesh-openconnect-java")
+    implementation project(":eventmesh-common")
+    implementation project(":eventmesh-connectors:eventmesh-connector-http")
+    implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-api")
+    implementation "io.cloudevents:cloudevents-core"
+    implementation "com.google.guava:guava"
+    implementation "io.cloudevents:cloudevents-json-jackson"
+    implementation ("io.grpc:grpc-protobuf:1.68.0") {
+        exclude group: "com.google.protobuf", module: "protobuf-java"
+    }
+    implementation 'io.cloudevents:cloudevents-http-vertx:3.0.0'
+    implementation 'io.vertx:vertx-web:4.5.8'
+    implementation 'io.vertx:vertx-web-client:4.5.9'
+    implementation 'dev.failsafe:failsafe:3.3.2'
+
+
+    testImplementation 'org.apache.httpcomponents.client5:httpclient5:5.4'
+    testImplementation 'org.apache.httpcomponents.client5:httpclient5-fluent:5.4'
+    testImplementation 'org.mock-server:mockserver-netty:5.15.0'
+    implementation 'io.netty:netty-codec-http:4.1.114.Final'
+    compileOnly 'org.projectlombok:lombok'
+    annotationProcessor 'org.projectlombok:lombok'
+}

eventmesh-connectors/eventmesh-connector-mcp/gradle.properties

@@ -0,0 +1,18 @@
+#
+# 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.
+#
+pluginType=connector
+pluginName=mcp

eventmesh-connectors/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/config/McpServerConfig.java

@@ -0,0 +1,34 @@
+/*
+ * 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.connector.mcp.config;
+
+import org.apache.eventmesh.common.config.connector.Config;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class McpServerConfig extends Config {
+
+    private boolean sourceEnable;
+
+    private boolean sinkEnable;
+}