You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by lr...@apache.org on 2023/09/07 02:09:06 UTC
[eventmesh] branch master updated: remove duplicated moudle (#4429)
This is an automated email from the ASF dual-hosted git repository.
lrhkobe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new ba694b437 remove duplicated moudle (#4429)
ba694b437 is described below
commit ba694b43771f824d9e510d92460d6ebd5c51cf9c
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Thu Sep 7 10:09:00 2023 +0800
remove duplicated moudle (#4429)
Co-authored-by: easoncchen(陈广胜) <ea...@webank.com>
---
.../eventmesh-storage-knative/build.gradle | 37 ----
.../eventmesh-storage-knative/gradle.properties | 18 --
.../storage/knative/admin/KnativeAdmin.java | 56 -----
.../storage/knative/admin/KnativeAdminAdaptor.java | 85 --------
.../knative/cloudevent/KnativeMessageFactory.java | 47 -----
.../impl/KnativeBinaryMessageReader.java | 57 -----
.../knative/cloudevent/impl/KnativeHeaders.java | 31 ---
.../cloudevent/impl/KnativeMessageWriter.java | 84 --------
.../knative/config/ClientConfiguration.java | 51 -----
.../storage/knative/consumer/DefaultConsumer.java | 75 -------
.../knative/consumer/KnativeConsumerImpl.java | 100 ---------
.../storage/knative/consumer/PullConsumerImpl.java | 177 ----------------
.../storage/knative/domain/NonStandardKeys.java | 27 ---
.../patch/EventMeshConsumeConcurrentlyContext.java | 30 ---
.../patch/EventMeshConsumeConcurrentlyStatus.java | 33 ---
.../EventMeshMessageListenerConcurrently.java | 24 ---
.../storage/knative/producer/AbstractProducer.java | 67 ------
.../knative/producer/KnativeProducerImpl.java | 101 ---------
.../storage/knative/producer/ProducerImpl.java | 128 -----------
.../storage/StorageResourceServiceKnativeImpl.java | 32 ---
.../storage/knative/utils/CloudEventUtils.java | 33 ---
.../eventmesh/org.apache.eventmesh.api.admin.Admin | 16 --
.../org.apache.eventmesh.api.consumer.Consumer | 16 --
.../org.apache.eventmesh.api.producer.Producer | 16 --
...he.eventmesh.api.storage.StorageResourceService | 16 --
.../src/main/resources/knative-client.properties | 18 --
.../knative/config/ClientConfigurationTest.java | 51 -----
.../knative/consumer/KnativeConsumerImplTest.java | 48 -----
.../knative/producer/KnativeProducerImplTest.java | 66 ------
.../src/test/resources/knative-client.properties | 18 --
.../eventmesh-storage-mongodb/build.gradle | 47 -----
.../eventmesh-storage-mongodb/gradle.properties | 18 --
.../storage/mongodb/admin/MongodbAdmin.java | 56 -----
.../storage/mongodb/admin/MongodbAdminAdaptor.java | 85 --------
.../mongodb/client/MongodbClientManager.java | 44 ----
.../storage/mongodb/config/ConfigKey.java | 25 ---
.../mongodb/config/ConfigurationHolder.java | 39 ----
.../storage/mongodb/constant/MongodbConstants.java | 37 ----
.../storage/mongodb/consumer/MongodbConsumer.java | 90 --------
.../consumer/MongodbReplicaSetConsumer.java | 140 -------------
.../consumer/MongodbStandaloneConsumer.java | 178 ----------------
.../mongodb/exception/MongodbStorageException.java | 31 ---
.../storage/mongodb/producer/MongodbProducer.java | 99 ---------
.../producer/MongodbReplicaSetProducer.java | 143 -------------
.../producer/MongodbStandaloneProducer.java | 157 --------------
.../storage/MongodbStorageResourceService.java | 32 ---
.../mongodb/utils/MongodbCloudEventUtil.java | 75 -------
.../storage/mongodb/utils/MongodbSequenceUtil.java | 54 -----
.../eventmesh/org.apache.eventmesh.api.admin.Admin | 16 --
.../org.apache.eventmesh.api.consumer.Consumer | 16 --
.../org.apache.eventmesh.api.producer.Producer | 16 --
...he.eventmesh.api.storage.StorageResourceService | 16 --
.../src/main/resources/mongodb-client.properties | 29 ---
.../storage/mongodb/MongodbConnectorTest.java | 79 -------
.../eventmesh/storage/mongodb/MongodbServer.java | 62 ------
.../storage/mongodb/MongodbServerTest.java | 71 -------
.../src/test/resources/mongodb-client.properties | 29 ---
.../eventmesh-storage-pravega/README.md | 19 --
.../eventmesh-storage-pravega/build.gradle | 39 ----
.../eventmesh-storage-pravega/gradle.properties | 20 --
.../eventmesh-storage-pravega/pravega-storage.jpg | Bin 143684 -> 0 bytes
.../storage/pravega/PravegaConsumerImpl.java | 109 ----------
.../storage/pravega/PravegaProducerImpl.java | 121 -----------
.../pravega/PravegaStorageResourceServiceImpl.java | 43 ----
.../storage/pravega/admin/PravegaAdmin.java | 56 -----
.../storage/pravega/admin/PravegaAdminAdaptor.java | 85 --------
.../storage/pravega/client/PravegaClient.java | 233 ---------------------
.../pravega/client/PravegaCloudEventWriter.java | 73 -------
.../storage/pravega/client/PravegaEvent.java | 77 -------
.../storage/pravega/client/SubscribeTask.java | 85 --------
.../pravega/config/PravegaStorageConfig.java | 67 ------
.../pravega/exception/PravegaStorageException.java | 31 ---
.../eventmesh/org.apache.eventmesh.api.admin.Admin | 16 --
.../org.apache.eventmesh.api.consumer.Consumer | 16 --
.../org.apache.eventmesh.api.producer.Producer | 16 --
...he.eventmesh.api.storage.StorageResourceService | 16 --
.../src/main/resources/pravega-storage.properties | 26 ---
.../storage/pravega/client/PravegaClientTest.java | 188 -----------------
.../pravega/config/PravegaStorageConfigTest.java | 60 ------
.../src/test/resources/pravega-storage.properties | 26 ---
settings.gradle | 3 -
81 files changed, 4667 deletions(-)
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/build.gradle b/eventmesh-storage-plugin/eventmesh-storage-knative/build.gradle
deleted file mode 100644
index 92d9a8e30..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/build.gradle
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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 {
- implementation 'org.asynchttpclient:async-http-client'
-
- implementation project(":eventmesh-sdks:eventmesh-sdk-java")
- testImplementation project(":eventmesh-sdks:eventmesh-sdk-java")
-
- implementation project(":eventmesh-common")
- testImplementation project(":eventmesh-common")
-
- implementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
- testImplementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
-
- testImplementation "org.powermock:powermock-module-junit4"
-
- compileOnly 'org.projectlombok:lombok:1.18.22'
- annotationProcessor 'org.projectlombok:lombok:1.18.22'
-
- testCompileOnly 'org.projectlombok:lombok:1.18.22'
- testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/gradle.properties b/eventmesh-storage-plugin/eventmesh-storage-knative/gradle.properties
deleted file mode 100644
index 8ba4f21ce..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/gradle.properties
+++ /dev/null
@@ -1,18 +0,0 @@
-# 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.
-#
-knative_version=1.5
-pluginType=storage
-pluginName=knative
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/admin/KnativeAdmin.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/admin/KnativeAdmin.java
deleted file mode 100644
index 86c935a1b..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/admin/KnativeAdmin.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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.storage.knative.admin;
-
-import org.apache.eventmesh.api.admin.AbstractAdmin;
-import org.apache.eventmesh.api.admin.TopicProperties;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import io.cloudevents.CloudEvent;
-
-public class KnativeAdmin extends AbstractAdmin {
-
- public KnativeAdmin() {
- super(new AtomicBoolean(false));
- }
-
- @Override
- public List<TopicProperties> getTopic() {
- // TODO implement admin functions
- return new ArrayList<>();
- }
-
- @Override
- public void createTopic(String topicName) {
- }
-
- @Override
- public void deleteTopic(String topicName) {
- }
-
- @Override
- public void publish(CloudEvent cloudEvent) throws Exception {
- }
-
- @Override
- public void shutdown() {
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/admin/KnativeAdminAdaptor.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/admin/KnativeAdminAdaptor.java
deleted file mode 100644
index f666e0973..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/admin/KnativeAdminAdaptor.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * 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.storage.knative.admin;
-
-import org.apache.eventmesh.api.admin.Admin;
-import org.apache.eventmesh.api.admin.TopicProperties;
-
-import java.util.List;
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-public class KnativeAdminAdaptor implements Admin {
-
- private final KnativeAdmin admin;
-
- public KnativeAdminAdaptor() {
- admin = new KnativeAdmin();
- }
-
- @Override
- public boolean isStarted() {
- return admin.isStarted();
- }
-
- @Override
- public boolean isClosed() {
- return admin.isClosed();
- }
-
- @Override
- public void start() {
- admin.start();
- }
-
- @Override
- public void shutdown() {
- admin.shutdown();
- }
-
- @Override
- public void init(Properties properties) throws Exception {
- admin.init(properties);
- }
-
- @Override
- public List<TopicProperties> getTopic() throws Exception {
- return admin.getTopic();
- }
-
- @Override
- public void createTopic(String topicName) {
- admin.createTopic(topicName);
- }
-
- @Override
- public void deleteTopic(String topicName) {
- admin.deleteTopic(topicName);
- }
-
- @Override
- public List<CloudEvent> getEvent(String topicName, int offset, int length) throws Exception {
- return admin.getEvent(topicName, offset, length);
- }
-
- @Override
- public void publish(CloudEvent cloudEvent) throws Exception {
- admin.publish(cloudEvent);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/KnativeMessageFactory.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/KnativeMessageFactory.java
deleted file mode 100644
index 91ed60778..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/KnativeMessageFactory.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.storage.knative.cloudevent;
-
-import org.apache.eventmesh.storage.knative.cloudevent.impl.KnativeMessageWriter;
-
-import java.nio.charset.StandardCharsets;
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public final class KnativeMessageFactory {
-
- private KnativeMessageFactory() {
- // prevent instantiation
- }
-
- public static String createReader(final CloudEvent message) {
- if (message.getData() == null) {
- log.warn("CloudEvent message's data is null.");
- return "";
- }
- return new String(message.getData().toBytes(), StandardCharsets.UTF_8);
- }
-
- public static KnativeMessageWriter createWriter(final Properties properties) {
- return new KnativeMessageWriter(properties);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeBinaryMessageReader.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeBinaryMessageReader.java
deleted file mode 100644
index 8a3060cab..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeBinaryMessageReader.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.storage.knative.cloudevent.impl;
-
-import java.util.Locale;
-import java.util.function.BiConsumer;
-
-import io.cloudevents.SpecVersion;
-import io.cloudevents.core.data.BytesCloudEventData;
-import io.cloudevents.core.message.impl.BaseGenericBinaryMessageReaderImpl;
-
-public class KnativeBinaryMessageReader extends BaseGenericBinaryMessageReaderImpl<String, String> {
-
- protected KnativeBinaryMessageReader(SpecVersion version, byte[] body) {
- super(version, body != null && body.length > 0 ? BytesCloudEventData.wrap(body) : (BytesCloudEventData) null);
- }
-
- @Override
- protected boolean isContentTypeHeader(String key) {
- return KnativeHeaders.CONTENT_TYPE.equals(key);
- }
-
- @Override
- protected boolean isCloudEventsHeader(String key) {
- return true;
- }
-
- @Override
- protected String toCloudEventsKey(String key) {
- return key.toLowerCase(Locale.getDefault());
- }
-
- @Override
- protected void forEachHeader(BiConsumer<String, String> fn) {
-
- }
-
- @Override
- protected String toCloudEventsValue(String value) {
- return value;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeHeaders.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeHeaders.java
deleted file mode 100644
index 1f998a4b2..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeHeaders.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.storage.knative.cloudevent.impl;
-
-public class KnativeHeaders {
-
- public static final String CONTENT_TYPE = "Content-Type";
-
- public static final String CE_ID = "Ce-Id";
-
- public static final String CE_SPECVERSION = "Ce-Specversion";
-
- public static final String CE_TYPE = "Ce-Type";
-
- public static final String CE_SOURCE = "Ce-Source";
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeMessageWriter.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeMessageWriter.java
deleted file mode 100644
index 273ee30d6..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeMessageWriter.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.storage.knative.cloudevent.impl;
-
-import org.apache.commons.lang3.ObjectUtils;
-
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-import java.util.Properties;
-
-import javax.annotation.Nullable;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.CloudEventData;
-import io.cloudevents.SpecVersion;
-import io.cloudevents.core.format.EventFormat;
-import io.cloudevents.core.message.MessageWriter;
-import io.cloudevents.core.v1.CloudEventBuilder;
-import io.cloudevents.rw.CloudEventContextWriter;
-import io.cloudevents.rw.CloudEventRWException;
-import io.cloudevents.rw.CloudEventWriter;
-
-public class KnativeMessageWriter implements MessageWriter<CloudEventWriter<String>, String>, CloudEventWriter<String> {
-
- private transient CloudEvent message;
-
- public CloudEvent getMessage() {
- return this.message;
- }
-
- public KnativeMessageWriter(Properties properties) {
- String s = "{ \"msg\": [\"" + properties.get("data") + "\"]}";
- this.message = new CloudEventBuilder()
- .withId(properties.getProperty(KnativeHeaders.CE_ID))
- .withSource(URI.create(properties.getProperty(KnativeHeaders.CE_SOURCE)))
- .withType(properties.getProperty(KnativeHeaders.CE_TYPE))
- .withDataContentType(properties.getProperty(KnativeHeaders.CONTENT_TYPE))
- .withData(s.getBytes(StandardCharsets.UTF_8))
- .build();
- }
-
- @Override
- public String end(CloudEventData data) throws CloudEventRWException {
- return data.toString();
- }
-
- @Override
- public String end() throws CloudEventRWException {
- if (ObjectUtils.allNotNull(message, message.getData())) {
- return message.getData().toString();
- }
- throw CloudEventRWException.newOther(new IllegalArgumentException("message data is null"));
- }
-
- @Override
- public CloudEventContextWriter withContextAttribute(@Nullable String name, @Nullable String value) throws CloudEventRWException {
- return null;
- }
-
- @Override
- public String setEvent(@Nullable EventFormat format, @Nullable byte[] value) throws CloudEventRWException {
- return null;
- }
-
- @Override
- public CloudEventWriter<String> create(SpecVersion version) throws CloudEventRWException {
- return null;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/config/ClientConfiguration.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/config/ClientConfiguration.java
deleted file mode 100644
index 5c0630470..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/config/ClientConfiguration.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.storage.knative.config;
-
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.common.config.ConfigFiled;
-import org.apache.eventmesh.common.config.convert.converter.ListConverter.ListConverterSemi;
-
-import java.util.List;
-
-import lombok.Data;
-
-@Data
-@Config(prefix = "eventMesh.server.knative", path = "classPath://knative-client.properties")
-public class ClientConfiguration {
-
- @ConfigFiled(field = "service", converter = ListConverterSemi.class)
- public List<String> service;
-
- /**
- * In keeping with the old way of configuration parsing, the value is taken from the service field [0]
- */
- @ConfigFiled(reload = true)
- public String emurl = "";
-
- /**
- * In keeping with the old way of configuration parsing, the value is taken from the service field [1]
- */
- @ConfigFiled(reload = true)
- public String serviceAddr = "";
-
- public void reload() {
- emurl = this.service.get(0);
- serviceAddr = this.service.get(1);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/DefaultConsumer.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/DefaultConsumer.java
deleted file mode 100644
index 2130ed852..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/DefaultConsumer.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.storage.knative.consumer;
-
-import static org.asynchttpclient.Dsl.asyncHttpClient;
-
-import org.apache.eventmesh.storage.knative.patch.EventMeshMessageListenerConcurrently;
-
-import java.util.concurrent.TimeUnit;
-
-import org.asynchttpclient.AsyncHttpClient;
-import org.asynchttpclient.ListenableFuture;
-import org.asynchttpclient.Response;
-import org.asynchttpclient.util.HttpConstants;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class DefaultConsumer {
-
- private static final Logger LOG = LoggerFactory.getLogger(DefaultConsumer.class);
-
- private transient AsyncHttpClient asyncHttpClient;
-
-
- private transient EventMeshMessageListenerConcurrently messageListener;
-
- public EventMeshMessageListenerConcurrently getMessageListener() {
- return messageListener;
- }
-
- public DefaultConsumer() throws Exception {
- this.asyncHttpClient = asyncHttpClient();
- }
-
- public String pullMessage(String topic, String subscribeUrl) throws Exception {
- Preconditions.checkNotNull(topic, "Subscribe item cannot be null");
- Preconditions.checkNotNull(subscribeUrl, "SubscribeUrl cannot be null");
-
- // Get event message via HTTP:
- String responseBody;
- ListenableFuture<Response> execute = asyncHttpClient.prepareGet("http://" + subscribeUrl + "/" + topic).execute();
- Response response = execute.get(10, TimeUnit.SECONDS);
-
- if (response.getStatusCode() == HttpConstants.ResponseStatusCodes.OK_200) {
- responseBody = response.getResponseBody();
- LOG.info(responseBody);
- return responseBody;
- }
- throw new IllegalStateException("HTTP response code error: " + response.getStatusCode());
- }
-
- public void registerMessageListener(EventMeshMessageListenerConcurrently messageListener) {
- this.messageListener = messageListener;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java
deleted file mode 100644
index fa7171031..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.storage.knative.consumer;
-
-import org.apache.eventmesh.api.AbstractContext;
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.consumer.Consumer;
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.storage.knative.config.ClientConfiguration;
-
-import java.util.List;
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Config(field = "clientConfiguration")
-@Slf4j
-public class KnativeConsumerImpl implements Consumer {
-
- private transient PullConsumerImpl pullConsumer;
-
- /**
- * Unified configuration class corresponding to knative-client.properties
- */
- private ClientConfiguration clientConfiguration;
-
- @Override
- public synchronized void init(Properties properties) throws Exception {
- // Load parameters from properties file:
- properties.put("emUrl", clientConfiguration.getEmurl());
- properties.put("serviceAddr", clientConfiguration.getServiceAddr());
-
- pullConsumer = new PullConsumerImpl(properties);
- }
-
- @Override
- public void subscribe(String topic) {
- pullConsumer.subscribe(topic);
- }
-
- @Override
- public void unsubscribe(String topic) {
- try {
- pullConsumer.unsubscribe(topic);
- } catch (Exception e) {
- log.error("unsubscribe error", e);
- }
- }
-
- @Override
- public void registerEventListener(EventListener listener) {
- pullConsumer.registerEventListener(listener);
- }
-
- @Override
- public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context) {
- pullConsumer.updateOffset(cloudEvents, context);
- }
-
- @Override
- public boolean isStarted() {
- return pullConsumer.isStarted();
- }
-
- @Override
- public boolean isClosed() {
- return pullConsumer.isClosed();
- }
-
- @Override
- public void start() {
- pullConsumer.start();
- }
-
- @Override
- public void shutdown() {
- pullConsumer.shutdown();
- }
-
- public ClientConfiguration getClientConfiguration() {
- return this.clientConfiguration;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/PullConsumerImpl.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/PullConsumerImpl.java
deleted file mode 100644
index c3378d05f..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/PullConsumerImpl.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
- * 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.storage.knative.consumer;
-
-import org.apache.eventmesh.api.AbstractContext;
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.EventMeshAction;
-import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
-import org.apache.eventmesh.common.protocol.SubscriptionItem;
-import org.apache.eventmesh.common.protocol.SubscriptionMode;
-import org.apache.eventmesh.common.protocol.SubscriptionType;
-import org.apache.eventmesh.storage.knative.domain.NonStandardKeys;
-import org.apache.eventmesh.storage.knative.patch.EventMeshConsumeConcurrentlyContext;
-import org.apache.eventmesh.storage.knative.patch.EventMeshConsumeConcurrentlyStatus;
-import org.apache.eventmesh.storage.knative.patch.EventMeshMessageListenerConcurrently;
-
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.cloudevents.CloudEvent;
-
-import com.google.common.collect.Lists;
-
-public class PullConsumerImpl {
-
- private static final Logger LOG = LoggerFactory.getLogger(PullConsumerImpl.class);
-
- private final transient DefaultConsumer defaultConsumer;
-
- // Topics to subscribe:
- private transient List<SubscriptionItem> topicList = null;
- private final transient ConcurrentHashMap<String, AtomicLong> offsetMap;
- private final transient AtomicBoolean started = new AtomicBoolean(false);
- private final transient Properties properties;
-
- // Store received message:
- private transient ConcurrentMap<String /* topic */, String /* responseBody */> subscriptionInner;
- private transient EventListener eventListener;
-
-
-
- public PullConsumerImpl(final Properties properties) throws Exception {
- this.properties = properties;
- this.topicList = Lists.newArrayList();
- this.subscriptionInner = new ConcurrentHashMap<String, String>();
- this.offsetMap = new ConcurrentHashMap<>();
- defaultConsumer = new DefaultConsumer();
-
- // Register listener:
- defaultConsumer.registerMessageListener(new ClusteringMessageListener());
- }
-
- public void subscribe(String topic) {
- // Subscribe topics:
- try {
- // Add topic to topicList:
- topicList.add(new SubscriptionItem(topic, SubscriptionMode.CLUSTERING, SubscriptionType.ASYNC));
- // Pull event messages iteratively:
- topicList.forEach(
- item -> {
- try {
- subscriptionInner.put(item.getTopic(),
- defaultConsumer.pullMessage(item.getTopic(), properties.getProperty("serviceAddr")));
- } catch (Exception e) {
- LOG.error("store received message error", e);
- }
- }
- );
- } catch (Exception e) {
- LOG.error("other error", e);
- }
- }
-
- public void unsubscribe(String topic) {
- try {
- // Unsubscribe topic:
- topicList.remove(topic);
- } catch (Exception e) {
- LOG.error("unsubscribe topic error", e);
- }
- }
-
- // todo: offset
- public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context) {
- cloudEvents.forEach(cloudEvent -> this.updateOffset(
- cloudEvent.getSubject(), (Long) cloudEvent.getExtension("offset"))
- );
- }
-
- public void updateOffset(String topicMetadata, Long offset) {
- offsetMap.computeIfPresent(topicMetadata, (k, v) -> {
- v.set(offset);
- return v;
- });
- }
-
- public void start() {
- this.started.set(true);
- }
-
- public synchronized void shutdown() {
- this.started.set(false);
- }
-
- public boolean isStarted() {
- return this.started.get();
- }
-
- public boolean isClosed() {
- return !this.isStarted();
- }
-
- public void registerEventListener(EventListener listener) {
- this.eventListener = listener;
- }
-
- // todo: load balancer cluser and broadcast
- private class ClusteringMessageListener extends EventMeshMessageListenerConcurrently {
- @Override
- public EventMeshConsumeConcurrentlyStatus handleMessage(CloudEvent cloudEvent, EventMeshConsumeConcurrentlyContext context) {
- final Properties contextProperties = new Properties();
- contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.RECONSUME_LATER.name());
-
- EventMeshAsyncConsumeContext eventMeshAsyncConsumeContext = new EventMeshAsyncConsumeContext() {
- @Override
- public void commit(EventMeshAction action) {
- switch (action) {
- case CommitMessage:
- contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS,
- EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS.name());
- break;
- case ReconsumeLater:
- contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS,
- EventMeshConsumeConcurrentlyStatus.RECONSUME_LATER.name());
- break;
- case ManualAck:
- contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS,
- EventMeshConsumeConcurrentlyStatus.CONSUME_FINISH.name());
- break;
- default:
- break;
- }
- }
- };
-
- eventMeshAsyncConsumeContext.setAbstractContext((AbstractContext) context);
-
- // Consume received message:
- eventListener.consume(cloudEvent, eventMeshAsyncConsumeContext);
-
- return EventMeshConsumeConcurrentlyStatus.valueOf(
- contextProperties.getProperty(NonStandardKeys.MESSAGE_CONSUME_STATUS));
- }
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/domain/NonStandardKeys.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/domain/NonStandardKeys.java
deleted file mode 100644
index 7c829afb2..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/domain/NonStandardKeys.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.storage.knative.domain;
-
-/**
- * NonStandardKeys
- */
-public class NonStandardKeys {
-
- public static final String MESSAGE_CONSUME_STATUS = "em.message.consume.status";
-
-}
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyContext.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyContext.java
deleted file mode 100644
index 0c0fb9b12..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyContext.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.storage.knative.patch;
-
-public class EventMeshConsumeConcurrentlyContext {
- private boolean manualAck = true;
-
- public boolean isManualAck() {
- return manualAck;
- }
-
- public void setManualAck(boolean manualAck) {
- this.manualAck = manualAck;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyStatus.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyStatus.java
deleted file mode 100644
index 3a2e2f30e..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyStatus.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.storage.knative.patch;
-
-public enum EventMeshConsumeConcurrentlyStatus {
- /**
- * Success consumption
- */
- CONSUME_SUCCESS,
- /**
- * Failure consumption,later try to consume
- */
- RECONSUME_LATER,
- /**
- * Success consumption but ack later manually
- */
- CONSUME_FINISH
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshMessageListenerConcurrently.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshMessageListenerConcurrently.java
deleted file mode 100644
index 6a0512fe3..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshMessageListenerConcurrently.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.storage.knative.patch;
-
-import io.cloudevents.CloudEvent;
-
-public abstract class EventMeshMessageListenerConcurrently {
- public abstract EventMeshConsumeConcurrentlyStatus handleMessage(CloudEvent cloudEvent, EventMeshConsumeConcurrentlyContext context);
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/AbstractProducer.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/AbstractProducer.java
deleted file mode 100644
index 0ef390dcf..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/AbstractProducer.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.storage.knative.producer;
-
-import static org.asynchttpclient.Dsl.asyncHttpClient;
-
-import org.apache.eventmesh.api.exception.StorageRuntimeException;
-import org.apache.eventmesh.api.producer.Producer;
-
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.asynchttpclient.AsyncHttpClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.cloudevents.CloudEvent;
-
-public abstract class AbstractProducer implements Producer {
-
- protected static final Logger LOG = LoggerFactory.getLogger(AbstractProducer.class);
-
- protected final transient AtomicBoolean started = new AtomicBoolean(false);
- protected final transient Properties properties;
- private transient AsyncHttpClient asyncHttpClient;
-
- AbstractProducer(final Properties properties) {
- this.properties = properties;
- this.asyncHttpClient = asyncHttpClient();
- }
-
- StorageRuntimeException checkProducerException(CloudEvent cloudEvent, Throwable e) {
- if (cloudEvent.getData() == null) {
- return new StorageRuntimeException(String.format("CloudEvent message data does not exist.", e));
- }
- return new StorageRuntimeException(String.format("Unknown connector runtime exception.", e));
- }
-
- @Override
- public boolean isStarted() {
- return started.get();
- }
-
- @Override
- public boolean isClosed() {
- return !this.isStarted();
- }
-
- public AsyncHttpClient getAsyncHttpClient() {
- return asyncHttpClient;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImpl.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImpl.java
deleted file mode 100644
index 7817436e2..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImpl.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * 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.storage.knative.producer;
-
-import org.apache.eventmesh.api.RequestReplyCallback;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.exception.StorageRuntimeException;
-import org.apache.eventmesh.api.producer.Producer;
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.storage.knative.config.ClientConfiguration;
-
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-@Config(field = "clientConfiguration")
-public class KnativeProducerImpl implements Producer {
-
- private transient ProducerImpl producer;
-
- /**
- * Unified configuration class corresponding to knative-client.properties
- */
- private ClientConfiguration clientConfiguration;
-
- @Override
- public synchronized void init(Properties properties) throws Exception {
- // Load parameters from properties file:
- properties.put("url", clientConfiguration.getServiceAddr());
- producer = new ProducerImpl(properties);
- }
-
- @Override
- public void publish(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- producer.sendAsync(cloudEvent, sendCallback);
- }
-
- @Override
- public boolean isStarted() {
- return producer.isStarted();
- }
-
- @Override
- public boolean isClosed() {
- return producer.isClosed();
- }
-
- @Override
- public void start() {
- throw new StorageRuntimeException("Start is not supported");
- }
-
- @Override
- public void shutdown() {
- throw new StorageRuntimeException("Shutdown is not supported");
- }
-
- @Override
- public void sendOneway(CloudEvent cloudEvent) {
- throw new StorageRuntimeException("SendOneWay is not supported");
- }
-
- @Override
- public void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout) throws Exception {
- throw new StorageRuntimeException("Request is not supported");
- }
-
- @Override
- public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- throw new StorageRuntimeException("Reply is not supported");
- }
-
- @Override
- public void checkTopicExist(String topic) throws Exception {
- throw new StorageRuntimeException("CheckTopicExist is not supported");
- }
-
- @Override
- public void setExtFields() {
- throw new StorageRuntimeException("SetExtFields is not supported");
- }
-
- public ClientConfiguration getClientConfiguration() {
- return this.clientConfiguration;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/ProducerImpl.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/ProducerImpl.java
deleted file mode 100644
index c36045555..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/ProducerImpl.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * 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.storage.knative.producer;
-
-import org.apache.eventmesh.api.RequestReplyCallback;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.exception.OnExceptionContext;
-import org.apache.eventmesh.api.exception.StorageRuntimeException;
-import org.apache.eventmesh.storage.knative.cloudevent.KnativeMessageFactory;
-import org.apache.eventmesh.storage.knative.cloudevent.impl.KnativeHeaders;
-import org.apache.eventmesh.storage.knative.utils.CloudEventUtils;
-
-import java.io.IOException;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-import org.asynchttpclient.ListenableFuture;
-import org.asynchttpclient.Response;
-import org.asynchttpclient.util.HttpConstants;
-
-import io.cloudevents.CloudEvent;
-
-public class ProducerImpl extends AbstractProducer {
-
- public ProducerImpl(final Properties properties) throws IOException {
- super(properties);
- }
-
- public Properties attributes() {
- return properties;
- }
-
- public void send(CloudEvent cloudEvent, SendCallback sendCallback) {
- // Set HTTP header, body and send CloudEvent message:
- try {
- ListenableFuture<Response> execute = super.getAsyncHttpClient().preparePost("http://" + this.attributes().getProperty("url"))
- .addHeader(KnativeHeaders.CONTENT_TYPE, cloudEvent.getDataContentType())
- .addHeader(KnativeHeaders.CE_ID, cloudEvent.getId())
- .addHeader(KnativeHeaders.CE_SPECVERSION, String.valueOf(cloudEvent.getSpecVersion()))
- .addHeader(KnativeHeaders.CE_TYPE, cloudEvent.getType())
- .addHeader(KnativeHeaders.CE_SOURCE, String.valueOf(cloudEvent.getSource()))
- .setBody(KnativeMessageFactory.createReader(cloudEvent))
- .execute();
-
- Response response = execute.get(10, TimeUnit.SECONDS);
- if (response.getStatusCode() == HttpConstants.ResponseStatusCodes.OK_200) {
- sendCallback.onSuccess(CloudEventUtils.convertSendResult(cloudEvent));
- return;
- }
- throw new IllegalStateException("HTTP response code error: " + response.getStatusCode());
- } catch (Exception e) {
- StorageRuntimeException onsEx = ProducerImpl.this.checkProducerException(cloudEvent, e);
- OnExceptionContext context = new OnExceptionContext();
- context.setTopic(KnativeMessageFactory.createReader(cloudEvent));
- context.setException(onsEx);
- sendCallback.onException(context);
- }
- }
-
- public void sendAsync(CloudEvent cloudEvent, SendCallback sendCallback) {
- try {
- this.send(cloudEvent, sendCallback);
- } catch (Exception e) {
- //LOG.error("Send cloudevent message Exception", e);
- throw new StorageRuntimeException("Send cloudevent message Exception.", e);
- }
- }
-
- @Override
- public void init(Properties properties) throws Exception {
- new ProducerImpl(properties);
- }
-
- @Override
- public void publish(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- this.sendAsync(cloudEvent, sendCallback);
- }
-
- @Override
- public void sendOneway(CloudEvent cloudEvent) {
- throw new StorageRuntimeException("SendOneWay is not supported");
- }
-
- @Override
- public void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout) throws Exception {
- throw new StorageRuntimeException("Request is not supported");
- }
-
- @Override
- public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- throw new StorageRuntimeException("Reply is not supported");
- }
-
- @Override
- public void checkTopicExist(String topic) throws Exception {
- throw new StorageRuntimeException("CheckTopicExist is not supported");
- }
-
- @Override
- public void setExtFields() {
- throw new StorageRuntimeException("SetExtFields is not supported");
- }
-
- @Override
- public void start() {
- started.set(true);
- }
-
- @Override
- public void shutdown() {
- started.set(false);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/storage/StorageResourceServiceKnativeImpl.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/storage/StorageResourceServiceKnativeImpl.java
deleted file mode 100644
index fabc3d330..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/storage/StorageResourceServiceKnativeImpl.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.storage.knative.storage;
-
-import org.apache.eventmesh.api.storage.StorageResourceService;
-
-public class StorageResourceServiceKnativeImpl implements StorageResourceService {
- @Override
- public void init() throws Exception {
-
- }
-
- @Override
- public void release() throws Exception {
-
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/utils/CloudEventUtils.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/utils/CloudEventUtils.java
deleted file mode 100644
index 9b2b4c96a..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/utils/CloudEventUtils.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.storage.knative.utils;
-
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.storage.knative.cloudevent.KnativeMessageFactory;
-
-import io.cloudevents.CloudEvent;
-
-public class CloudEventUtils {
-
- public static SendResult convertSendResult(CloudEvent cloudEvent) {
- SendResult sendResult = new SendResult();
- sendResult.setTopic(KnativeMessageFactory.createReader(cloudEvent));
- sendResult.setMessageId(cloudEvent.getId());
- return sendResult;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
deleted file mode 100644
index 07c24919a..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-knative=org.apache.eventmesh.storage.knative.admin.KnativeAdminAdaptor
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
deleted file mode 100644
index 6b10bacce..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-knative=org.apache.eventmesh.storage.knative.consumer.KnativeConsumerImpl
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
deleted file mode 100644
index bc5f8b41e..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-knative=org.apache.eventmesh.storage.knative.producer.KnativeProducerImpl
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
deleted file mode 100644
index 06c107570..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-knative=org.apache.eventmesh.storage.knative.storage.StorageResourceServiceKnativeImpl
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/knative-client.properties b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/knative-client.properties
deleted file mode 100644
index 8fec1c430..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/knative-client.properties
+++ /dev/null
@@ -1,18 +0,0 @@
-#
-# 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.
-#
-#######################knative-client##################
-eventMesh.server.knative.service=127.0.0.1;cloudevents-player.default.127.0.0.1.sslip.io
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/config/ClientConfigurationTest.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/config/ClientConfigurationTest.java
deleted file mode 100644
index 334b9b4de..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/config/ClientConfigurationTest.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.storage.knative.config;
-
-import org.apache.eventmesh.api.factory.StoragePluginFactory;
-import org.apache.eventmesh.storage.knative.consumer.KnativeConsumerImpl;
-import org.apache.eventmesh.storage.knative.producer.KnativeProducerImpl;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-public class ClientConfigurationTest {
-
- @Test
- public void getConfigWhenKnativeConsumerInit() {
- KnativeConsumerImpl consumer =
- (KnativeConsumerImpl) StoragePluginFactory.getMeshMQPushConsumer("knative");
-
- ClientConfiguration config = consumer.getClientConfiguration();
- assertConfig(config);
- }
-
- @Test
- public void getConfigWhenKnativeProducerInit() {
- KnativeProducerImpl producer =
- (KnativeProducerImpl) StoragePluginFactory.getMeshMQProducer("knative");
-
- ClientConfiguration config = producer.getClientConfiguration();
- assertConfig(config);
- }
-
- private void assertConfig(ClientConfiguration config) {
- Assert.assertEquals("127.0.0.1", config.getEmurl());
- Assert.assertEquals("cloudevents-player.default.127.0.0.1.sslip.io", config.getServiceAddr());
- }
-}
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImplTest.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImplTest.java
deleted file mode 100644
index d2888ffb9..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImplTest.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.storage.knative.consumer;
-
-import org.apache.eventmesh.api.factory.StoragePluginFactory;
-
-import java.util.Properties;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-public class KnativeConsumerImplTest {
-
- @Test
- public void testSubscribe() throws Exception {
- Properties properties = new Properties();
- final String topic = "messages";
- properties.put("topic", topic);
-
- // Create a Knative consumer:
- KnativeConsumerImpl knativeConsumer =
- (KnativeConsumerImpl) StoragePluginFactory.getMeshMQPushConsumer("knative");
-
- try {
- knativeConsumer.init(properties);
-
- // Subscribe:
- knativeConsumer.subscribe(properties.getProperty("topic"));
- } catch (Exception e) {
- Assert.fail(e.getMessage());
- }
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImplTest.java b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImplTest.java
deleted file mode 100644
index 0ae77db30..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImplTest.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.storage.knative.producer;
-
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.api.exception.OnExceptionContext;
-import org.apache.eventmesh.api.factory.StoragePluginFactory;
-import org.apache.eventmesh.storage.knative.cloudevent.KnativeMessageFactory;
-import org.apache.eventmesh.storage.knative.cloudevent.impl.KnativeHeaders;
-
-import java.util.Properties;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-public class KnativeProducerImplTest {
-
- @Test
- public void testPublish() throws Exception {
- Properties properties = new Properties();
-
- properties.put(KnativeHeaders.CONTENT_TYPE, "application/json");
- properties.put(KnativeHeaders.CE_ID, "1234");
- properties.put(KnativeHeaders.CE_SPECVERSION, "1.0");
- properties.put(KnativeHeaders.CE_TYPE, "some-type");
- properties.put(KnativeHeaders.CE_SOURCE, "java-client");
- properties.put("data", "Hello Knative from EventMesh!");
-
- // Create a Knative producer:
- KnativeProducerImpl knativeProducer =
- (KnativeProducerImpl) StoragePluginFactory.getMeshMQProducer("knative");
-
- try {
- knativeProducer.init(properties);
-
- // Publish an event message:
- knativeProducer.publish(KnativeMessageFactory.createWriter(properties).getMessage(), new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- }
-
- @Override
- public void onException(OnExceptionContext context) {
- }
- });
- } catch (Exception e) {
- Assert.fail(e.getMessage());
- }
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/resources/knative-client.properties b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/resources/knative-client.properties
deleted file mode 100644
index 8fec1c430..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/resources/knative-client.properties
+++ /dev/null
@@ -1,18 +0,0 @@
-#
-# 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.
-#
-#######################knative-client##################
-eventMesh.server.knative.service=127.0.0.1;cloudevents-player.default.127.0.0.1.sslip.io
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/build.gradle b/eventmesh-storage-plugin/eventmesh-storage-mongodb/build.gradle
deleted file mode 100644
index 5d4e671bd..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/build.gradle
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.
- */
-
-configurations {
- implementation.exclude group: 'ch.qos.logback', module: 'logback-classic'
- implementation.exclude group: 'log4j', module: 'log4j'
-}
-
-dependencies {
- implementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
- implementation project(":eventmesh-common")
-
- testImplementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
- testImplementation project(":eventmesh-common")
-
- implementation 'org.mongodb:mongodb-driver:3.12.11'
- testImplementation 'org.mongodb:mongodb-driver:3.12.11'
-
- implementation 'io.cloudevents:cloudevents-json-jackson'
- testImplementation 'io.cloudevents:cloudevents-json-jackson'
-
- testImplementation "org.mockito:mockito-core"
- testImplementation "org.powermock:powermock-module-junit4"
- testImplementation "org.powermock:powermock-api-mockito2"
-
- compileOnly 'org.projectlombok:lombok:1.18.22'
- annotationProcessor 'org.projectlombok:lombok:1.18.22'
-
- testCompileOnly 'org.projectlombok:lombok:1.18.22'
- testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
-
- testImplementation 'de.bwaldvogel:mongo-java-server:1.42.0'
-}
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/gradle.properties b/eventmesh-storage-plugin/eventmesh-storage-mongodb/gradle.properties
deleted file mode 100644
index dac8f0000..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/gradle.properties
+++ /dev/null
@@ -1,18 +0,0 @@
-# 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=storage
-pluginName=mongodb
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/admin/MongodbAdmin.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/admin/MongodbAdmin.java
deleted file mode 100644
index b3c54d0c5..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/admin/MongodbAdmin.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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.storage.mongodb.admin;
-
-import org.apache.eventmesh.api.admin.AbstractAdmin;
-import org.apache.eventmesh.api.admin.TopicProperties;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import io.cloudevents.CloudEvent;
-
-public class MongodbAdmin extends AbstractAdmin {
-
- public MongodbAdmin() {
- super(new AtomicBoolean(false));
- }
-
- @Override
- public List<TopicProperties> getTopic() {
- // TODO implement admin functions
- return new ArrayList<>();
- }
-
- @Override
- public void createTopic(String topicName) {
- }
-
- @Override
- public void deleteTopic(String topicName) {
- }
-
- @Override
- public void publish(CloudEvent cloudEvent) throws Exception {
- }
-
- @Override
- public void shutdown() {
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/admin/MongodbAdminAdaptor.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/admin/MongodbAdminAdaptor.java
deleted file mode 100644
index 48b55525d..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/admin/MongodbAdminAdaptor.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * 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.storage.mongodb.admin;
-
-import org.apache.eventmesh.api.admin.Admin;
-import org.apache.eventmesh.api.admin.TopicProperties;
-
-import java.util.List;
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-public class MongodbAdminAdaptor implements Admin {
-
- private final MongodbAdmin admin;
-
- public MongodbAdminAdaptor() {
- admin = new MongodbAdmin();
- }
-
- @Override
- public boolean isStarted() {
- return admin.isStarted();
- }
-
- @Override
- public boolean isClosed() {
- return admin.isClosed();
- }
-
- @Override
- public void start() {
- admin.start();
- }
-
- @Override
- public void shutdown() {
- admin.shutdown();
- }
-
- @Override
- public void init(Properties properties) throws Exception {
- admin.init(properties);
- }
-
- @Override
- public List<TopicProperties> getTopic() throws Exception {
- return admin.getTopic();
- }
-
- @Override
- public void createTopic(String topicName) {
- admin.createTopic(topicName);
- }
-
- @Override
- public void deleteTopic(String topicName) {
- admin.deleteTopic(topicName);
- }
-
- @Override
- public List<CloudEvent> getEvent(String topicName, int offset, int length) throws Exception {
- return admin.getEvent(topicName, offset, length);
- }
-
- @Override
- public void publish(CloudEvent cloudEvent) throws Exception {
- admin.publish(cloudEvent);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/client/MongodbClientManager.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/client/MongodbClientManager.java
deleted file mode 100644
index 41aa54485..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/client/MongodbClientManager.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.storage.mongodb.client;
-
-import com.mongodb.ConnectionString;
-import com.mongodb.client.MongoClient;
-import com.mongodb.client.MongoClients;
-
-public class MongodbClientManager {
- /**
- * create mongodb client
- *
- * @param url url, like: mongodb://root:root@192.168.74.143:27018,192.168.74.143:27019
- * @return mongodb client
- */
- public static MongoClient createMongodbClient(String url) {
- ConnectionString connectionString = new ConnectionString(url);
- return MongoClients.create(connectionString);
- }
-
- /**
- * close mongodb client
- *
- * @param mongoClient mongodb client
- */
- public static void closeMongodbClient(MongoClient mongoClient) {
- mongoClient.close();
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/config/ConfigKey.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/config/ConfigKey.java
deleted file mode 100644
index e5949f819..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/config/ConfigKey.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.storage.mongodb.config;
-
-public class ConfigKey {
-
- public static final String STANDALONE = "STANDALONE";
-
- public static final String REPLICA_SET = "REPLICA_SET";
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/config/ConfigurationHolder.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/config/ConfigurationHolder.java
deleted file mode 100644
index f42dc2cda..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/config/ConfigurationHolder.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.storage.mongodb.config;
-
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.common.config.ConfigFiled;
-
-import lombok.Data;
-
-@Data
-@Config(prefix = "eventMesh.server.mongodb", path = "classPath://mongodb-client.properties")
-public class ConfigurationHolder {
- @ConfigFiled(field = "connectorType")
- private String connectorType;
-
- @ConfigFiled(field = "url")
- private String url;
-
- @ConfigFiled(field = "database")
- private String database;
-
- @ConfigFiled(field = "collection")
- private String collection;
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/constant/MongodbConstants.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/constant/MongodbConstants.java
deleted file mode 100644
index 23a2570db..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/constant/MongodbConstants.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.storage.mongodb.constant;
-
-public class MongodbConstants {
- public static final String TOPIC = "flag";
- public static final String CAPPED_COL_TOPIC_FN = "topic";
- public static final String CAPPED_COL_CURSOR_FN = "ts";
- public static final String SEQUENCE_COLLECTION_NAME = "pub_sub_seq";
- public static final String SEQUENCE_KEY_FN = "topic";
- public static final String SEQUENCE_VALUE_FN = "value";
- public static final String CLOUD_EVENT_DOC_VERSION = "version";
- public static final String CLOUD_EVENT_DOC_DATA = "data";
- public static final String CLOUD_EVENT_DOC_ID = "id";
- public static final String CLOUD_EVENT_DOC_SOURCE = "source";
- public static final String CLOUD_EVENT_DOC_TYPE = "type";
- public static final String CLOUD_EVENT_DOC_DATACONTENTTYPE = "datacontenttype";
- public static final String CLOUD_EVENT_DOC_SUBJECT = "subject";
-
-
-
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbConsumer.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbConsumer.java
deleted file mode 100644
index 0484b7b4a..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbConsumer.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.storage.mongodb.consumer;
-
-import org.apache.eventmesh.api.AbstractContext;
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.consumer.Consumer;
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.storage.mongodb.config.ConfigKey;
-import org.apache.eventmesh.storage.mongodb.config.ConfigurationHolder;
-
-import java.util.List;
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-@Config(field = "configurationHolder")
-public class MongodbConsumer implements Consumer {
-
- private ConfigurationHolder configurationHolder;
-
- private Consumer consumer;
-
- @Override
- public boolean isStarted() {
- return consumer.isStarted();
- }
-
- @Override
- public boolean isClosed() {
- return consumer.isClosed();
- }
-
- @Override
- public void start() {
- consumer.start();
- }
-
- @Override
- public void shutdown() {
- consumer.shutdown();
- }
-
- @Override
- public void init(Properties keyValue) throws Exception {
- String connectorType = configurationHolder.getConnectorType();
- if (connectorType.equals(ConfigKey.STANDALONE)) {
- consumer = new MongodbStandaloneConsumer(configurationHolder);
- }
- if (connectorType.equals(ConfigKey.REPLICA_SET)) {
- consumer = new MongodbReplicaSetConsumer(configurationHolder);
- }
- consumer.init(keyValue);
- }
-
- @Override
- public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context) {
- consumer.updateOffset(cloudEvents, context);
- }
-
- @Override
- public void subscribe(String topic) throws Exception {
- consumer.subscribe(topic);
- }
-
- @Override
- public void unsubscribe(String topic) {
- consumer.unsubscribe(topic);
- }
-
- @Override
- public void registerEventListener(EventListener listener) {
- consumer.registerEventListener(listener);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbReplicaSetConsumer.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbReplicaSetConsumer.java
deleted file mode 100644
index d67899bc6..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbReplicaSetConsumer.java
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
- * 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.storage.mongodb.consumer;
-
-import org.apache.eventmesh.api.AbstractContext;
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.EventMeshAction;
-import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
-import org.apache.eventmesh.api.consumer.Consumer;
-import org.apache.eventmesh.storage.mongodb.client.MongodbClientManager;
-import org.apache.eventmesh.storage.mongodb.config.ConfigurationHolder;
-import org.apache.eventmesh.storage.mongodb.utils.MongodbCloudEventUtil;
-
-import java.util.List;
-import java.util.Properties;
-
-import org.bson.Document;
-
-import io.cloudevents.CloudEvent;
-
-import com.mongodb.client.ChangeStreamIterable;
-import com.mongodb.client.MongoChangeStreamCursor;
-import com.mongodb.client.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.model.changestream.ChangeStreamDocument;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class MongodbReplicaSetConsumer implements Consumer {
- private final ConfigurationHolder configurationHolder;
-
- private MongoClient mongoClient;
-
- private MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor;
-
- private volatile boolean started = false;
-
- private EventListener eventListener;
-
- public MongodbReplicaSetConsumer(ConfigurationHolder configurationHolder) {
- this.configurationHolder = configurationHolder;
- }
-
- @Override
- public boolean isStarted() {
- return started;
- }
-
- @Override
- public boolean isClosed() {
- return !isStarted();
- }
-
- @Override
- public void start() {
- if (!started) {
- started = true;
- }
- }
-
- @Override
- public void shutdown() {
- if (started) {
- try {
- if (this.mongoClient != null) {
- MongodbClientManager.closeMongodbClient(this.mongoClient);
- }
- if (this.cursor != null) {
- this.cursor.close();
- }
- } finally {
- started = false;
- }
- }
- }
-
- @Override
- public void init(Properties keyValue) {
- this.mongoClient = MongodbClientManager.createMongodbClient(configurationHolder.getUrl());
- }
-
- @Override
- public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context) {
-
- }
-
- @Override
- public void subscribe(String topic) {
- MongoCollection<Document> collection = mongoClient
- .getDatabase(configurationHolder.getDatabase()).getCollection(configurationHolder.getCollection());
- ChangeStreamIterable<Document> changeStreamDocuments = collection.watch();
- this.cursor = changeStreamDocuments.cursor();
- this.handle();
- }
-
- @Override
- public void unsubscribe(String topic) {
- this.cursor.close();
- }
-
- @Override
- public void registerEventListener(EventListener listener) {
- this.eventListener = listener;
- }
-
- private void handle() {
- while (this.cursor.hasNext()) {
- ChangeStreamDocument<Document> next = cursor.next();
- Document fullDocument = next.getFullDocument();
- if (fullDocument != null) {
- CloudEvent cloudEvent = MongodbCloudEventUtil.convertToCloudEvent(fullDocument);
- final EventMeshAsyncConsumeContext consumeContext = new EventMeshAsyncConsumeContext() {
- @Override
- public void commit(EventMeshAction action) {
- log.info("[MongodbReplicaSetConsumer] Mongodb consumer context commit.");
- }
- };
- if (eventListener != null) {
- eventListener.consume(cloudEvent, consumeContext);
- }
- }
- }
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbStandaloneConsumer.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbStandaloneConsumer.java
deleted file mode 100644
index 9722923ae..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/consumer/MongodbStandaloneConsumer.java
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * 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.storage.mongodb.consumer;
-
-import org.apache.eventmesh.api.AbstractContext;
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.EventMeshAction;
-import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
-import org.apache.eventmesh.api.consumer.Consumer;
-import org.apache.eventmesh.common.ThreadPoolFactory;
-import org.apache.eventmesh.storage.mongodb.client.MongodbClientManager;
-import org.apache.eventmesh.storage.mongodb.config.ConfigurationHolder;
-import org.apache.eventmesh.storage.mongodb.constant.MongodbConstants;
-import org.apache.eventmesh.storage.mongodb.utils.MongodbCloudEventUtil;
-
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.bson.Document;
-
-import io.cloudevents.CloudEvent;
-
-import com.mongodb.client.FindIterable;
-import com.mongodb.client.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoDatabase;
-
-import lombok.extern.slf4j.Slf4j;
-@Slf4j
-@SuppressWarnings("all")
-public class MongodbStandaloneConsumer implements Consumer {
-
- private final ConfigurationHolder configurationHolder;
-
- private volatile boolean started = false;
-
- private EventListener eventListener;
-
- private MongoClient client;
-
- private MongoDatabase db;
-
- private MongoCollection<Document> cappedCol;
-
- private SubTask task = new SubTask();
-
- private final ThreadPoolExecutor executor = ThreadPoolFactory.createThreadPoolExecutor(
- Runtime.getRuntime().availableProcessors() * 2,
- Runtime.getRuntime().availableProcessors() * 2,
- "EventMesh-Mongodb-StandaloneConsumer-");
-
- public MongodbStandaloneConsumer(ConfigurationHolder configurationHolder) {
- this.configurationHolder = configurationHolder;
- }
-
- @Override
- public boolean isStarted() {
- return started;
- }
-
- @Override
- public boolean isClosed() {
- return !isStarted();
- }
-
- @Override
- public void start() {
- if (!started) {
- started = true;
- }
- }
-
- @Override
- public void shutdown() {
- if (started) {
- try {
- MongodbClientManager.closeMongodbClient(this.client);
- } finally {
- started = false;
- }
- }
- }
-
- @Override
- public void init(Properties keyValue) {
- this.client = MongodbClientManager.createMongodbClient(configurationHolder.getUrl());
- this.db = client.getDatabase(configurationHolder.getDatabase());
- this.cappedCol = db.getCollection(configurationHolder.getCollection());
- //create index
- Document index = new Document(MongodbConstants.CAPPED_COL_CURSOR_FN, 1)
- .append(MongodbConstants.CAPPED_COL_TOPIC_FN, 1);
- this.cappedCol.createIndex(index);
- }
-
- @Override
- public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context) {
-
- }
-
- @Override
- public void subscribe(String topic) {
- task = new SubTask();
- executor.execute(task);
- }
-
- @Override
- public void unsubscribe(String topic) {
- task.stop();
- }
-
- @Override
- public void registerEventListener(EventListener listener) {
- this.eventListener = listener;
- }
-
- private FindIterable<Document> getCursor(MongoCollection<Document> collection, String topic, int lastId) {
- Document index = new Document("$gt", lastId);
- Document ts = new Document(MongodbConstants.CAPPED_COL_CURSOR_FN, index);
-
- Document spec = ts.append(MongodbConstants.CAPPED_COL_TOPIC_FN, topic);
- FindIterable<Document> documents = collection.find(spec);
- return documents;
- }
-
- private class SubTask implements Runnable {
- private final AtomicBoolean stop = new AtomicBoolean(false);
-
- public void run() {
- int lastId = -1;
- while (!stop.get()) {
- try {
- FindIterable<Document> cur = getCursor(cappedCol, MongodbConstants.TOPIC, lastId);
- for (Document obj : cur) {
- CloudEvent cloudEvent = MongodbCloudEventUtil.convertToCloudEvent(obj);
- final EventMeshAsyncConsumeContext consumeContext = new EventMeshAsyncConsumeContext() {
- @Override
- public void commit(EventMeshAction action) {
- log.info("[MongodbStandaloneConsumer] Mongodb consumer context commit.");
- }
- };
- if (eventListener != null) {
- eventListener.consume(cloudEvent, consumeContext);
- }
- try {
- lastId = (int) ((Double) obj.get(MongodbConstants.CAPPED_COL_CURSOR_FN)).doubleValue();
- } catch (ClassCastException ce) {
- lastId = (Integer) obj.get(MongodbConstants.CAPPED_COL_CURSOR_FN);
- }
- }
- } catch (Exception ex) {
- log.error("[MongodbStandaloneConsumer] thread run happen exception.", ex);
- }
- Thread.yield();
- }
- }
-
- public void stop() {
- stop.set(true);
- }
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/exception/MongodbStorageException.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/exception/MongodbStorageException.java
deleted file mode 100644
index e9dfe6794..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/exception/MongodbStorageException.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.storage.mongodb.exception;
-
-import org.apache.eventmesh.api.exception.StorageRuntimeException;
-
-public class MongodbStorageException extends StorageRuntimeException {
-
- public MongodbStorageException(String message) {
- super(message);
- }
-
- public MongodbStorageException(Throwable throwable) {
- super(throwable);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbProducer.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbProducer.java
deleted file mode 100644
index 84e6fa7a8..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbProducer.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
- * 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.storage.mongodb.producer;
-
-import org.apache.eventmesh.api.RequestReplyCallback;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.producer.Producer;
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.storage.mongodb.config.ConfigKey;
-import org.apache.eventmesh.storage.mongodb.config.ConfigurationHolder;
-
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-@Config(field = "configurationHolder")
-public class MongodbProducer implements Producer {
-
- private ConfigurationHolder configurationHolder;
-
- private Producer producer;
-
- @Override
- public boolean isStarted() {
- return producer.isStarted();
- }
-
- @Override
- public boolean isClosed() {
- return producer.isClosed();
- }
-
- @Override
- public void start() {
- producer.start();
- }
-
- @Override
- public void shutdown() {
- producer.shutdown();
- }
-
- @Override
- public void init(Properties properties) throws Exception {
- String connectorType = configurationHolder.getConnectorType();
- if (connectorType.equals(ConfigKey.STANDALONE)) {
- producer = new MongodbStandaloneProducer(configurationHolder);
- }
- if (connectorType.equals(ConfigKey.REPLICA_SET)) {
- producer = new MongodbReplicaSetProducer(configurationHolder);
- }
- producer.init(properties);
- }
-
- @Override
- public void publish(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- producer.publish(cloudEvent, sendCallback);
- }
-
- @Override
- public void sendOneway(CloudEvent cloudEvent) {
- producer.sendOneway(cloudEvent);
- }
-
- @Override
- public void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout) throws Exception {
- producer.request(cloudEvent, rrCallback, timeout);
- }
-
- @Override
- public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- return producer.reply(cloudEvent, sendCallback);
- }
-
- @Override
- public void checkTopicExist(String topic) throws Exception {
- producer.checkTopicExist(topic);
- }
-
- @Override
- public void setExtFields() {
- producer.setExtFields();
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbReplicaSetProducer.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbReplicaSetProducer.java
deleted file mode 100644
index 458896800..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbReplicaSetProducer.java
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
- * 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.storage.mongodb.producer;
-
-import org.apache.eventmesh.api.RequestReplyCallback;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.api.exception.OnExceptionContext;
-import org.apache.eventmesh.api.producer.Producer;
-import org.apache.eventmesh.storage.mongodb.client.MongodbClientManager;
-import org.apache.eventmesh.storage.mongodb.config.ConfigurationHolder;
-import org.apache.eventmesh.storage.mongodb.exception.MongodbStorageException;
-import org.apache.eventmesh.storage.mongodb.utils.MongodbCloudEventUtil;
-
-import java.util.Properties;
-
-import org.bson.Document;
-
-import io.cloudevents.CloudEvent;
-
-import com.mongodb.client.MongoClient;
-import com.mongodb.client.MongoCollection;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class MongodbReplicaSetProducer implements Producer {
- private volatile boolean started = false;
-
- private final ConfigurationHolder configurationHolder;
-
- private MongoClient mongoClient;
-
- public MongodbReplicaSetProducer(ConfigurationHolder configurationHolder) {
- this.configurationHolder = configurationHolder;
- }
-
- @Override
- public boolean isStarted() {
- return started;
- }
-
- @Override
- public boolean isClosed() {
- return !isStarted();
- }
-
- @Override
- public void start() {
- if (!started) {
- started = true;
- }
- }
-
- @Override
- public void shutdown() {
- if (started) {
- try {
- if (this.mongoClient != null) {
- MongodbClientManager.closeMongodbClient(this.mongoClient);
- }
- } finally {
- started = false;
- }
- }
- }
-
- @Override
- public void init(Properties properties) {
- this.mongoClient = MongodbClientManager.createMongodbClient(configurationHolder.getUrl());
- }
-
- @Override
- public void publish(CloudEvent cloudEvent, SendCallback sendCallback) {
- try {
- Document document = MongodbCloudEventUtil.convertToDocument(cloudEvent);
- MongoCollection<Document> collection = mongoClient
- .getDatabase(configurationHolder.getDatabase()).getCollection(configurationHolder.getCollection());
- collection.insertOne(document);
-
- SendResult sendResult = new SendResult();
- sendResult.setTopic(cloudEvent.getSubject());
- sendResult.setMessageId(cloudEvent.getId());
- sendCallback.onSuccess(sendResult);
- } catch (Exception ex) {
- log.error("[MongodbReplicaSetProducer] publish happen exception.", ex);
- sendCallback.onException(
- OnExceptionContext.builder()
- .topic(cloudEvent.getSubject())
- .messageId(cloudEvent.getId())
- .exception(new MongodbStorageException(ex))
- .build()
- );
- }
- }
-
- @Override
- public void sendOneway(CloudEvent cloudEvent) {
- try {
- Document document = MongodbCloudEventUtil.convertToDocument(cloudEvent);
- MongoCollection<Document> collection = mongoClient
- .getDatabase(configurationHolder.getDatabase()).getCollection(configurationHolder.getCollection());
- collection.insertOne(document);
- } catch (Exception ex) {
- log.error("[MongodbReplicaSetProducer] sendOneway happen exception.", ex);
- }
- }
-
- @Override
- public void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout) {
-
- }
-
- @Override
- public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback) {
- return false;
- }
-
- @Override
- public void checkTopicExist(String topic) {
-
- }
-
- @Override
- public void setExtFields() {
-
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbStandaloneProducer.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbStandaloneProducer.java
deleted file mode 100644
index 5a8e99812..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/producer/MongodbStandaloneProducer.java
+++ /dev/null
@@ -1,157 +0,0 @@
-/*
- * 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.storage.mongodb.producer;
-
-import org.apache.eventmesh.api.RequestReplyCallback;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.api.exception.OnExceptionContext;
-import org.apache.eventmesh.api.producer.Producer;
-import org.apache.eventmesh.storage.mongodb.client.MongodbClientManager;
-import org.apache.eventmesh.storage.mongodb.config.ConfigurationHolder;
-import org.apache.eventmesh.storage.mongodb.constant.MongodbConstants;
-import org.apache.eventmesh.storage.mongodb.exception.MongodbStorageException;
-import org.apache.eventmesh.storage.mongodb.utils.MongodbCloudEventUtil;
-import org.apache.eventmesh.storage.mongodb.utils.MongodbSequenceUtil;
-
-import java.util.Properties;
-
-import org.bson.Document;
-
-import io.cloudevents.CloudEvent;
-
-import com.mongodb.client.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoDatabase;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@SuppressWarnings("all")
-public class MongodbStandaloneProducer implements Producer {
-
- private final ConfigurationHolder configurationHolder;
-
- private volatile boolean started = false;
-
- private MongoClient client;
-
- private MongoDatabase db;
-
- private MongoCollection<Document> cappedCol;
-
- private MongodbSequenceUtil mongodbSequenceUtil;
-
- public MongodbStandaloneProducer(ConfigurationHolder configurationHolder) {
- this.configurationHolder = configurationHolder;
- }
-
- @Override
- public boolean isStarted() {
- return started;
- }
-
- @Override
- public boolean isClosed() {
- return !isStarted();
- }
-
- @Override
- public void start() {
- if (!started) {
- started = true;
- }
- }
-
- @Override
- public void shutdown() {
- if (started) {
- try {
- MongodbClientManager.closeMongodbClient(this.client);
- } finally {
- started = false;
- }
- }
- }
-
- @Override
- public void init(Properties keyValue) {
- this.client = MongodbClientManager.createMongodbClient(configurationHolder.getUrl());
- this.db = client.getDatabase(configurationHolder.getDatabase());
- this.cappedCol = db.getCollection(configurationHolder.getCollection());
- this.mongodbSequenceUtil = new MongodbSequenceUtil(configurationHolder);
- }
-
- @Override
- public void publish(CloudEvent cloudEvent, SendCallback sendCallback) {
- try {
- Document doc = MongodbCloudEventUtil.convertToDocument(cloudEvent);
- int i = mongodbSequenceUtil.getNextSeq(MongodbConstants.TOPIC);
- doc.append(MongodbConstants.CAPPED_COL_TOPIC_FN, MongodbConstants.TOPIC)
- .append(MongodbConstants.CAPPED_COL_CURSOR_FN, i);
- cappedCol.insertOne(doc);
-
- SendResult sendResult = new SendResult();
- sendResult.setTopic(cloudEvent.getSubject());
- sendResult.setMessageId(cloudEvent.getId());
- sendCallback.onSuccess(sendResult);
- } catch (Exception ex) {
- log.error("[MongodbStandaloneProducer] publish happen exception.", ex);
- sendCallback.onException(
- OnExceptionContext.builder()
- .topic(cloudEvent.getSubject())
- .messageId(cloudEvent.getId())
- .exception(new MongodbStorageException(ex))
- .build()
- );
- }
- }
-
- @Override
- public void sendOneway(CloudEvent cloudEvent) {
- try {
- Document doc = MongodbCloudEventUtil.convertToDocument(cloudEvent);
- int i = mongodbSequenceUtil.getNextSeq(MongodbConstants.TOPIC);
- doc.append(MongodbConstants.CAPPED_COL_TOPIC_FN, MongodbConstants.TOPIC)
- .append(MongodbConstants.CAPPED_COL_CURSOR_FN, i);
- cappedCol.insertOne(doc);
- } catch (Exception ex) {
- log.error("[MongodbStandaloneProducer] sendOneway happen exception.", ex);
- }
- }
-
- @Override
- public void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout) throws Exception {
-
- }
-
- @Override
- public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- return false;
- }
-
- @Override
- public void checkTopicExist(String topic) throws Exception {
-
- }
-
- @Override
- public void setExtFields() {
-
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/storage/MongodbStorageResourceService.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/storage/MongodbStorageResourceService.java
deleted file mode 100644
index 8dcffc335..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/storage/MongodbStorageResourceService.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.storage.mongodb.storage;
-
-import org.apache.eventmesh.api.storage.StorageResourceService;
-
-public class MongodbStorageResourceService implements StorageResourceService {
- @Override
- public void init() throws Exception {
-
- }
-
- @Override
- public void release() throws Exception {
-
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/utils/MongodbCloudEventUtil.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/utils/MongodbCloudEventUtil.java
deleted file mode 100644
index c18c711ef..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/utils/MongodbCloudEventUtil.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.storage.mongodb.utils;
-
-import org.apache.eventmesh.common.Constants;
-import org.apache.eventmesh.storage.mongodb.constant.MongodbConstants;
-import org.apache.eventmesh.storage.mongodb.exception.MongodbStorageException;
-
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-
-import org.bson.Document;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.SpecVersion;
-import io.cloudevents.core.builder.CloudEventBuilder;
-
-
-public class MongodbCloudEventUtil {
- public static CloudEvent convertToCloudEvent(Document document) {
- document.remove("_id");
- String versionStr = document.getString(MongodbConstants.CLOUD_EVENT_DOC_VERSION);
- SpecVersion version = SpecVersion.valueOf(versionStr);
- CloudEventBuilder builder;
- switch (version) {
- case V03:
- builder = CloudEventBuilder.v03();
- break;
- case V1:
- builder = CloudEventBuilder.v1();
- break;
- default:
- throw new MongodbStorageException(String.format("CloudEvent version %s does not support.", version));
- }
- builder.withData(document.remove(MongodbConstants.CLOUD_EVENT_DOC_DATA).toString().getBytes(Constants.DEFAULT_CHARSET))
- .withId(document.remove(MongodbConstants.CLOUD_EVENT_DOC_ID).toString())
- .withSource(URI.create(document.remove(MongodbConstants.CLOUD_EVENT_DOC_SOURCE).toString()))
- .withType(document.remove(MongodbConstants.CLOUD_EVENT_DOC_TYPE).toString())
- .withDataContentType(document.remove(MongodbConstants.CLOUD_EVENT_DOC_DATACONTENTTYPE).toString())
- .withSubject(document.remove(MongodbConstants.CLOUD_EVENT_DOC_SUBJECT).toString());
- document.forEach((key, value) -> builder.withExtension(key, value.toString()));
-
- return builder.build();
- }
-
- public static Document convertToDocument(CloudEvent cloudEvent) {
- Document document = new Document();
- document.put(MongodbConstants.CLOUD_EVENT_DOC_VERSION, cloudEvent.getSpecVersion().name());
- document.put(MongodbConstants.CLOUD_EVENT_DOC_DATA, cloudEvent.getData() == null
- ? null : new String(cloudEvent.getData().toBytes(), StandardCharsets.UTF_8));
- document.put(MongodbConstants.CLOUD_EVENT_DOC_ID, cloudEvent.getId());
- document.put(MongodbConstants.CLOUD_EVENT_DOC_SOURCE, cloudEvent.getSource().toString());
- document.put(MongodbConstants.CLOUD_EVENT_DOC_TYPE, cloudEvent.getType());
- document.put(MongodbConstants.CLOUD_EVENT_DOC_DATACONTENTTYPE, cloudEvent.getDataContentType());
- document.put(MongodbConstants.CLOUD_EVENT_DOC_SUBJECT, cloudEvent.getSubject());
- cloudEvent.getExtensionNames().forEach(key -> document.put(key, cloudEvent.getExtension(key)));
-
- return document;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/utils/MongodbSequenceUtil.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/utils/MongodbSequenceUtil.java
deleted file mode 100644
index 9eb58aa1e..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/java/org/apache/eventmesh/storage/mongodb/utils/MongodbSequenceUtil.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.storage.mongodb.utils;
-
-import org.apache.eventmesh.storage.mongodb.client.MongodbClientManager;
-import org.apache.eventmesh.storage.mongodb.config.ConfigurationHolder;
-import org.apache.eventmesh.storage.mongodb.constant.MongodbConstants;
-
-import org.bson.Document;
-
-import com.mongodb.BasicDBObject;
-import com.mongodb.client.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoDatabase;
-import com.mongodb.client.model.FindOneAndUpdateOptions;
-import com.mongodb.client.model.ReturnDocument;
-
-@SuppressWarnings("all")
-public class MongodbSequenceUtil {
- private final MongoClient mongoClient;
- private final MongoDatabase db;
- private final MongoCollection<Document> seqCol;
-
- public MongodbSequenceUtil(ConfigurationHolder configurationHolder) {
- mongoClient = MongodbClientManager.createMongodbClient(configurationHolder.getUrl());
- db = mongoClient.getDatabase(configurationHolder.getDatabase());
- seqCol = db.getCollection(MongodbConstants.SEQUENCE_COLLECTION_NAME);
- }
-
- public int getNextSeq(String topic) {
- Document query = new Document(MongodbConstants.SEQUENCE_KEY_FN, topic);
- Document update = new Document("$inc", new BasicDBObject(MongodbConstants.SEQUENCE_VALUE_FN, 1));
- FindOneAndUpdateOptions options = new FindOneAndUpdateOptions();
- options.upsert(true);
- options.returnDocument(ReturnDocument.AFTER);
- Document result = seqCol.findOneAndUpdate(query, update, options);
- return (int) (Integer) result.get(MongodbConstants.SEQUENCE_VALUE_FN);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
deleted file mode 100644
index 41e799399..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-mongodb=org.apache.eventmesh.storage.mongodb.admin.MongodbAdminAdaptor
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
deleted file mode 100644
index a405ee7bf..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-mongodb=org.apache.eventmesh.storage.mongodb.consumer.MongodbConsumer
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
deleted file mode 100644
index f7c627f98..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-mongodb=org.apache.eventmesh.storage.mongodb.producer.MongodbProducer
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
deleted file mode 100644
index ea0ae7939..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-mongodb=org.apache.eventmesh.storage.mongodb.storage.MongodbStorageResourceService
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/mongodb-client.properties b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/mongodb-client.properties
deleted file mode 100644
index 70afeb8db..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/main/resources/mongodb-client.properties
+++ /dev/null
@@ -1,29 +0,0 @@
-#
-# 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.
-#
-
-####################### mongodb server ##################
-eventMesh.server.mongodb.connectorType=STANDALONE
-##like: mongodb://root:root@127.0.0.1:27018,127.0.0.1:27019
-eventMesh.server.mongodb.url=mongodb://127.0.0.1:27018
-eventMesh.server.mongodb.database=db
-eventMesh.server.mongodb.collection=col
-
-#eventMesh.server.mongodb.connectorType=REPLICA_SET
-##like: mongodb://root:root@127.0.0.1:27018,127.0.0.1:27019
-#eventMesh.server.mongodb.url=mongodb://root:root@127.0.0.1:27018,127.0.0.1:27019
-#eventMesh.server.mongodb.database=db
-#eventMesh.server.mongodb.collection=col
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbConnectorTest.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbConnectorTest.java
deleted file mode 100644
index 8ba1ce3cd..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbConnectorTest.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * 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.storage.mongodb;
-
-import org.apache.eventmesh.api.EventMeshAction;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.api.exception.OnExceptionContext;
-
-import java.net.URI;
-import java.time.OffsetDateTime;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.builder.CloudEventBuilder;
-
-public class MongodbConnectorTest extends MongodbServer {
- @Test
- public void subscribe() throws Exception {
- final int expectedCount = 5;
- final CountDownLatch downLatch = new CountDownLatch(expectedCount);
-
- mongodbConsumer.registerEventListener((cloudEvent, context) -> {
- downLatch.countDown();
- context.commit(EventMeshAction.CommitMessage);
- Assert.assertEquals(cloudEvent.getSubject(), "topic");
- });
-
- mongodbConsumer.subscribe("topic");
-
- Thread.sleep(1000);
- for (int i = 0; i < expectedCount; i++) {
- CloudEvent cloudEvent = CloudEventBuilder.v1()
- .withId(String.valueOf(i))
- .withTime(OffsetDateTime.now())
- .withSource(URI.create("testsource"))
- .withSubject("topic")
- .withType(String.class.getCanonicalName())
- .withDataContentType("text/plain")
- .withData("data".getBytes())
- .build();
-
- mongodbProducer.publish(cloudEvent, new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- Assert.assertEquals(cloudEvent.getId(), sendResult.getMessageId());
- Assert.assertEquals(cloudEvent.getSubject(), sendResult.getTopic());
- }
-
- @Override
- public void onException(OnExceptionContext context) {
- Assert.assertEquals(cloudEvent.getId(), context.getMessageId());
- Assert.assertEquals(cloudEvent.getSubject(), context.getTopic());
- }
- });
- }
-
- Assert.assertTrue(downLatch.await(5, TimeUnit.MINUTES));
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbServer.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbServer.java
deleted file mode 100644
index 51094442a..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbServer.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * 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.storage.mongodb;
-
-import org.apache.eventmesh.api.factory.StoragePluginFactory;
-import org.apache.eventmesh.storage.mongodb.consumer.MongodbConsumer;
-import org.apache.eventmesh.storage.mongodb.producer.MongodbProducer;
-
-import java.util.Properties;
-
-import org.junit.After;
-import org.junit.Before;
-
-import de.bwaldvogel.mongo.MongoServer;
-import de.bwaldvogel.mongo.backend.memory.MemoryBackend;
-
-public class MongodbServer {
- private MongoServer server1;
- private MongoServer server2;
- public MongodbConsumer mongodbConsumer;
- public MongodbProducer mongodbProducer;
-
- @Before
- public void setup() throws Exception {
- server1 = new MongoServer(new MemoryBackend());
- server1.bind("127.0.0.1", 27018);
- server2 = new MongoServer(new MemoryBackend());
- server2.bind("127.0.0.1", 27019);
-
- mongodbConsumer = (MongodbConsumer) StoragePluginFactory.getMeshMQPushConsumer("mongodb");
- mongodbConsumer.init(new Properties());
- mongodbConsumer.start();
-
- mongodbProducer = (MongodbProducer) StoragePluginFactory.getMeshMQProducer("mongodb");
- mongodbProducer.init(new Properties());
- mongodbProducer.start();
- }
-
- @After
- public void shutdown() {
- mongodbConsumer.shutdown();
- mongodbProducer.shutdown();
-
- server1.shutdown();
- server2.shutdown();
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbServerTest.java b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbServerTest.java
deleted file mode 100644
index 5d3e6cdfb..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/java/org/apache/eventmesh/storage/mongodb/MongodbServerTest.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.storage.mongodb;
-
-import org.bson.Document;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import com.mongodb.client.MongoClient;
-import com.mongodb.client.MongoClients;
-import com.mongodb.client.MongoCollection;
-
-import de.bwaldvogel.mongo.MongoServer;
-import de.bwaldvogel.mongo.backend.memory.MemoryBackend;
-
-public class MongodbServerTest {
- private MongoCollection<Document> collection;
- private MongoClient client;
- private MongoServer server;
-
- @Before
- public void setUp() {
- server = new MongoServer(new MemoryBackend());
-
- // optionally:
- // server.enableSsl(key, keyPassword, certificate);
- // server.enableOplog();
-
- // bind on a random local port
- server.bind("127.0.0.1", 27011);
- String connectionString = server.getConnectionString();
-
- client = MongoClients.create(connectionString);
- collection = client.getDatabase("testdb").getCollection("testcollection");
- }
-
- @After
- public void tearDown() {
- client.close();
- server.shutdown();
- }
-
- @Test
- public void testSimpleInsertQuery() {
- Assert.assertEquals(collection.countDocuments(), 0);
-
- // creates the database and collection in memory and insert the object
- Document obj = new Document("_id", 1).append("key", "value");
- collection.insertOne(obj);
-
- Assert.assertEquals(collection.countDocuments(), 1L);
- Assert.assertEquals(collection.find().first(), obj);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/resources/mongodb-client.properties b/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/resources/mongodb-client.properties
deleted file mode 100644
index 70afeb8db..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-mongodb/src/test/resources/mongodb-client.properties
+++ /dev/null
@@ -1,29 +0,0 @@
-#
-# 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.
-#
-
-####################### mongodb server ##################
-eventMesh.server.mongodb.connectorType=STANDALONE
-##like: mongodb://root:root@127.0.0.1:27018,127.0.0.1:27019
-eventMesh.server.mongodb.url=mongodb://127.0.0.1:27018
-eventMesh.server.mongodb.database=db
-eventMesh.server.mongodb.collection=col
-
-#eventMesh.server.mongodb.connectorType=REPLICA_SET
-##like: mongodb://root:root@127.0.0.1:27018,127.0.0.1:27019
-#eventMesh.server.mongodb.url=mongodb://root:root@127.0.0.1:27018,127.0.0.1:27019
-#eventMesh.server.mongodb.database=db
-#eventMesh.server.mongodb.collection=col
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/README.md b/eventmesh-storage-plugin/eventmesh-storage-pravega/README.md
deleted file mode 100644
index 92306965b..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/README.md
+++ /dev/null
@@ -1,19 +0,0 @@
-# Pravega Connector
-Pravega connector of EventMesh supports taking Pravega as event storage. [Pravega](https://cncf.pravega.io/) is a storage system that exposes Stream as the main primitive for continuous and unbounded data and meets the requirement of EventMesh connector module.
-
-## The Design of Pravega Connector
-Pravega has 3 level storage concepts, which are **Event**, **Stream**, **Scope**. An **Event** is represented as a set of bytes within a Stream. A **Stream** is a durable, elastic, append-only, unbounded sequence of bytes having good performance and strong consistency. A **Scope** acts as a namespace for Stream names and all Stream names are unique within their Scope.
-
-Naturally, Pravega connector makes each topic correspond to a pravega stream. Because `EventStreamWriter` and `EventStreamReader` must assign to a special stream in a special scope, each topic corresponds to an `EventStreamWriter` and a `ReaderGroup`(only contains one `EventStreamReader`) as well. What's more, `SubscribeTask` thread will read every topic stream continuously.
-
-**Note**:
-- Publish success will respond `-1` as messageId since writing to Pravega will not return message offset of the stream. If you want to use EventMesh with Pravega, please ensure messageId is unimportant.
-- Name and ID in Pravega must be `0-9`, `A-Z`, `a-z`, `.`, `-`.
- - Topic must be legal in Pravega.
- - ReaderGroup name format is `String.format("%s-%s", consumerGroup, topic)` for cluster message model, while random UUID for broadcast message model.
- - Reader name format is `String.format("%s-reader", instanceName).replaceAll("\\(", "-").replaceAll("\\)", "-")`. The `instanceName` consists of `consumerGroup`, config `eventMesh.server.cluster`, EventMesh version and PID.
-
-
-
-## Active Pravega Connector
-Referring to doc [_Build and Load Plugins_](https://eventmesh.apache.org/docs/installation/runtime#13-build-and-load-plugins) to load `:eventmesh-connector-plugin:eventmesh-connector-pravega` plugin module, and edit `eventMesh.connector.plugin.type=pravega` in config file `eventmesh.properties` to active pravega connector.
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/build.gradle b/eventmesh-storage-plugin/eventmesh-storage-pravega/build.gradle
deleted file mode 100644
index 253c06720..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/build.gradle
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.
- */
-
-configurations {
- implementation.exclude group: 'ch.qos.logback', module: 'logback-classic'
- implementation.exclude group: 'log4j', module: 'log4j'
-}
-
-dependencies {
- implementation project(":eventmesh-common")
- implementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
- implementation("io.pravega:pravega-client:$pravega_version") {
- exclude group: 'io.netty', module: 'netty-codec-http2:4.1.30.Final'
- }
- // use newer version to avoid CVE
- implementation 'io.netty:netty-codec-http2'
-
- testImplementation "org.testcontainers:testcontainers:1.17.3"
-
- compileOnly 'org.projectlombok:lombok'
- annotationProcessor 'org.projectlombok:lombok'
-
- testCompileOnly 'org.projectlombok:lombok'
- testAnnotationProcessor 'org.projectlombok:lombok'
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/gradle.properties b/eventmesh-storage-plugin/eventmesh-storage-pravega/gradle.properties
deleted file mode 100644
index 0bd75ae16..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/gradle.properties
+++ /dev/null
@@ -1,20 +0,0 @@
-# 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.
-#
-
-pravega_version=0.11.0
-
-pluginType=storage
-pluginName=pravega
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/pravega-storage.jpg b/eventmesh-storage-plugin/eventmesh-storage-pravega/pravega-storage.jpg
deleted file mode 100644
index 20cb82960..000000000
Binary files a/eventmesh-storage-plugin/eventmesh-storage-pravega/pravega-storage.jpg and /dev/null differ
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
deleted file mode 100644
index 2e8b2b693..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * 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.storage.pravega;
-
-import org.apache.eventmesh.api.AbstractContext;
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.consumer.Consumer;
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.storage.pravega.client.PravegaClient;
-import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
-import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
-
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import io.cloudevents.CloudEvent;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@Config(field = "pravegaConnectorConfig")
-public class PravegaConsumerImpl implements Consumer {
-
- private static final AtomicBoolean started = new AtomicBoolean(false);
-
- /**
- * Unified configuration class corresponding to pravega-storage.properties
- */
- private PravegaStorageConfig pravegaConnectorConfig;
-
- private boolean isBroadcast;
- private String instanceName;
- private String consumerGroup;
- private PravegaClient client;
- private EventListener eventListener;
-
- @Override
- public void init(Properties keyValue) throws Exception {
- isBroadcast = Boolean.parseBoolean(keyValue.getProperty("isBroadcast", "false"));
- instanceName = keyValue.getProperty("instanceName", "");
- consumerGroup = keyValue.getProperty("consumerGroup", "");
-
- client = PravegaClient.getInstance(pravegaConnectorConfig);
- }
-
- @Override
- public void start() {
- started.compareAndSet(false, true);
- }
-
- @Override
- public void shutdown() {
- started.compareAndSet(true, false);
- }
-
- @Override
- public boolean isStarted() {
- return started.get();
- }
-
- @Override
- public boolean isClosed() {
- return !started.get();
- }
-
- @Override
- public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void subscribe(String topic) {
- if (!client.subscribe(topic, isBroadcast, consumerGroup, instanceName, eventListener)) {
- throw new PravegaStorageException(String.format("subscribe topic[%s] fail.", topic));
- }
- }
-
- @Override
- public void unsubscribe(String topic) {
- if (!client.unsubscribe(topic, isBroadcast, consumerGroup)) {
- throw new PravegaStorageException(String.format("unsubscribe topic[%s] fail.", topic));
- }
- }
-
- @Override
- public void registerEventListener(EventListener listener) {
- this.eventListener = listener;
- }
-
- public PravegaStorageConfig getClientConfiguration() {
- return this.pravegaConnectorConfig;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaProducerImpl.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaProducerImpl.java
deleted file mode 100644
index 707858d50..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaProducerImpl.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * 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.storage.pravega;
-
-import org.apache.eventmesh.api.RequestReplyCallback;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.api.exception.OnExceptionContext;
-import org.apache.eventmesh.api.exception.StorageRuntimeException;
-import org.apache.eventmesh.api.producer.Producer;
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.storage.pravega.client.PravegaClient;
-import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
-import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
-
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import io.cloudevents.CloudEvent;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@Config(field = "pravegaConnectorConfig")
-public class PravegaProducerImpl implements Producer {
-
- private final AtomicBoolean started = new AtomicBoolean(false);
- private PravegaClient client;
-
- /**
- * Unified configuration class corresponding to pravega-storage.properties
- */
- private PravegaStorageConfig pravegaConnectorConfig;
-
- @Override
- public void init(Properties properties) throws Exception {
- client = PravegaClient.getInstance(pravegaConnectorConfig);
- }
-
- @Override
- public void start() {
- started.compareAndSet(false, true);
- }
-
- @Override
- public void shutdown() {
- started.compareAndSet(true, false);
- }
-
- @Override
- public boolean isStarted() {
- return started.get();
- }
-
- @Override
- public boolean isClosed() {
- return !started.get();
- }
-
- @Override
- public void publish(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- try {
- SendResult sendResult = client.publish(cloudEvent.getSubject(), cloudEvent);
- sendCallback.onSuccess(sendResult);
- } catch (Exception e) {
- log.error("send message error, topic: {}", cloudEvent.getSubject());
- OnExceptionContext onExceptionContext = OnExceptionContext.builder()
- .messageId("-1")
- .topic(cloudEvent.getSubject())
- .exception(new StorageRuntimeException(e))
- .build();
- sendCallback.onException(onExceptionContext);
- }
- }
-
- @Override
- public void sendOneway(CloudEvent cloudEvent) {
- client.publish(cloudEvent.getSubject(), cloudEvent);
- }
-
- @Override
- public void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout) throws Exception {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback) throws Exception {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void checkTopicExist(String topic) throws Exception {
- if (!client.checkTopicExist(topic)) {
- throw new PravegaStorageException(String.format("topic:%s is not exist", topic));
- }
- }
-
- @Override
- public void setExtFields() {
-
- }
-
- public PravegaStorageConfig getClientConfiguration() {
- return this.pravegaConnectorConfig;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaStorageResourceServiceImpl.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaStorageResourceServiceImpl.java
deleted file mode 100644
index 9e785c5e9..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaStorageResourceServiceImpl.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.storage.pravega;
-
-import org.apache.eventmesh.api.storage.StorageResourceService;
-import org.apache.eventmesh.storage.pravega.client.PravegaClient;
-import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class PravegaStorageResourceServiceImpl implements StorageResourceService {
-
- /**
- * Unified configuration class corresponding to pravega-storage.properties
- */
- private PravegaStorageConfig pravegaConnectorConfig = new PravegaStorageConfig();
-
- @Override
- public void init() throws Exception {
- PravegaClient.getInstance(pravegaConnectorConfig).start();
- }
-
- @Override
- public void release() {
- PravegaClient.getInstance().shutdown();
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/admin/PravegaAdmin.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/admin/PravegaAdmin.java
deleted file mode 100644
index 8f4a87372..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/admin/PravegaAdmin.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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.storage.pravega.admin;
-
-import org.apache.eventmesh.api.admin.AbstractAdmin;
-import org.apache.eventmesh.api.admin.TopicProperties;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import io.cloudevents.CloudEvent;
-
-public class PravegaAdmin extends AbstractAdmin {
-
- public PravegaAdmin() {
- super(new AtomicBoolean(false));
- }
-
- @Override
- public List<TopicProperties> getTopic() {
- // TODO implement admin functions
- return new ArrayList<>();
- }
-
- @Override
- public void createTopic(String topicName) {
- }
-
- @Override
- public void deleteTopic(String topicName) {
- }
-
- @Override
- public void publish(CloudEvent cloudEvent) throws Exception {
- }
-
- @Override
- public void shutdown() {
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/admin/PravegaAdminAdaptor.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/admin/PravegaAdminAdaptor.java
deleted file mode 100644
index faa53539f..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/admin/PravegaAdminAdaptor.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * 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.storage.pravega.admin;
-
-import org.apache.eventmesh.api.admin.Admin;
-import org.apache.eventmesh.api.admin.TopicProperties;
-
-import java.util.List;
-import java.util.Properties;
-
-import io.cloudevents.CloudEvent;
-
-public class PravegaAdminAdaptor implements Admin {
-
- private final PravegaAdmin admin;
-
- public PravegaAdminAdaptor() {
- admin = new PravegaAdmin();
- }
-
- @Override
- public boolean isStarted() {
- return admin.isStarted();
- }
-
- @Override
- public boolean isClosed() {
- return admin.isClosed();
- }
-
- @Override
- public void start() {
- admin.start();
- }
-
- @Override
- public void shutdown() {
- admin.shutdown();
- }
-
- @Override
- public void init(Properties properties) throws Exception {
- admin.init(properties);
- }
-
- @Override
- public List<TopicProperties> getTopic() throws Exception {
- return admin.getTopic();
- }
-
- @Override
- public void createTopic(String topicName) {
- admin.createTopic(topicName);
- }
-
- @Override
- public void deleteTopic(String topicName) {
- admin.deleteTopic(topicName);
- }
-
- @Override
- public List<CloudEvent> getEvent(String topicName, int offset, int length) throws Exception {
- return admin.getEvent(topicName, offset, length);
- }
-
- @Override
- public void publish(CloudEvent cloudEvent) throws Exception {
- admin.publish(cloudEvent);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaClient.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaClient.java
deleted file mode 100644
index 401a19ea9..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaClient.java
+++ /dev/null
@@ -1,233 +0,0 @@
-/*
- * 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.storage.pravega.client;
-
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
-import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
-
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import io.cloudevents.CloudEvent;
-import io.pravega.client.ClientConfig;
-import io.pravega.client.EventStreamClientFactory;
-import io.pravega.client.admin.ReaderGroupManager;
-import io.pravega.client.admin.StreamManager;
-import io.pravega.client.stream.EventStreamReader;
-import io.pravega.client.stream.EventStreamWriter;
-import io.pravega.client.stream.EventWriterConfig;
-import io.pravega.client.stream.ReaderConfig;
-import io.pravega.client.stream.ReaderGroupConfig;
-import io.pravega.client.stream.StreamConfiguration;
-import io.pravega.client.stream.impl.ByteArraySerializer;
-import io.pravega.shared.NameUtils;
-import io.pravega.shared.security.auth.DefaultCredentials;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class PravegaClient {
-
- private final PravegaStorageConfig config;
- private final StreamManager streamManager;
- private final EventStreamClientFactory clientFactory;
- private final ReaderGroupManager readerGroupManager;
- private final Map<String, EventStreamWriter<byte[]>> writerMap = new ConcurrentHashMap<>();
- private final Map<String, SubscribeTask> subscribeTaskMap = new ConcurrentHashMap<>();
-
- private static PravegaClient instance;
-
- public static PravegaClient getInstance() {
- return instance;
- }
-
- public static PravegaClient getInstance(PravegaStorageConfig config) {
- if (instance == null) {
- instance = new PravegaClient(config);
- }
-
- return instance;
- }
-
- private PravegaClient(PravegaStorageConfig config) {
- this.config = config;
- streamManager = StreamManager.create(config.getControllerURI());
- ClientConfig.ClientConfigBuilder clientConfigBuilder = ClientConfig.builder().controllerURI(config.getControllerURI());
- if (config.isAuthEnabled()) {
- clientConfigBuilder.credentials(new DefaultCredentials(config.getPassword(), config.getUsername()));
- }
- if (config.isTlsEnable()) {
- clientConfigBuilder.trustStore(config.getTruststore()).validateHostName(false);
- }
- ClientConfig clientConfig = clientConfigBuilder.build();
- clientFactory = EventStreamClientFactory.withScope(config.getScope(), clientConfig);
- readerGroupManager = ReaderGroupManager.withScope(config.getScope(), clientConfig);
- }
-
- protected static PravegaClient getNewInstance(PravegaStorageConfig config) {
- return new PravegaClient(config);
- }
-
- public void start() {
- if (createScope()) {
- log.info("Create Pravega scope[{}] success.", config.getScope());
- } else {
- log.info("Pravega scope[{}] has already been created.", config.getScope());
- }
- }
-
- public void shutdown() {
- subscribeTaskMap.forEach((topic, task) -> task.stopRead());
- subscribeTaskMap.clear();
- writerMap.forEach((topic, writer) -> writer.close());
- writerMap.clear();
- readerGroupManager.close();
- clientFactory.close();
- streamManager.close();
- }
-
- /**
- * Publish CloudEvent to Pravega stream named topic. Note that the messageId in SendResult is always -1 since {@link
- * EventStreamWriter#writeEvent(Object)} just return {@link java.util.concurrent.CompletableFuture} with {@link Void} which couldn't get
- * messageId.
- *
- * @param topic topic
- * @param cloudEvent cloudEvent
- * @return SendResult whose messageId is always -1
- */
- public SendResult publish(String topic, CloudEvent cloudEvent) {
- if (!createStream(topic)) {
- log.debug("stream[{}] has already been created.", topic);
- }
- EventStreamWriter<byte[]> writer = writerMap.computeIfAbsent(topic, k -> createWrite(topic));
- PravegaCloudEventWriter cloudEventWriter = new PravegaCloudEventWriter(topic);
- PravegaEvent pravegaEvent = cloudEventWriter.writeBinary(cloudEvent);
- try {
- writer.writeEvent(PravegaEvent.toByteArray(pravegaEvent)).get(5, TimeUnit.SECONDS);
- } catch (ExecutionException | InterruptedException | TimeoutException e) {
- log.error(String.format("Write topic[%s] fail.", topic), e);
- throw new PravegaStorageException(String.format("Write topic[%s] fail.", topic));
- }
- SendResult sendResult = new SendResult();
- sendResult.setTopic(topic);
- // set -1 as messageId since writeEvent method doesn't return it.
- sendResult.setMessageId("-1");
- return sendResult;
-
- }
-
- public boolean subscribe(String topic, boolean isBroadcast, String consumerGroup, String instanceName, EventListener listener) {
- if (subscribeTaskMap.containsKey(topic)) {
- return true;
- }
- String readerGroupName = buildReaderGroupName(isBroadcast, consumerGroup, topic);
- createReaderGroup(topic, readerGroupName);
- String readerId = buildReaderId(instanceName);
- EventStreamReader<byte[]> reader = createReader(readerId, readerGroupName);
- SubscribeTask subscribeTask = new SubscribeTask(topic, reader, listener);
- subscribeTask.start();
- subscribeTaskMap.put(topic, subscribeTask);
- return true;
- }
-
- public boolean unsubscribe(String topic, boolean isBroadcast, String consumerGroup) {
- if (!subscribeTaskMap.containsKey(topic)) {
- return true;
- }
- if (!isBroadcast) {
- deleteReaderGroup(buildReaderGroupName(false, consumerGroup, topic));
- }
- subscribeTaskMap.remove(topic).stopRead();
- return true;
- }
-
- public boolean checkTopicExist(String topic) {
- return streamManager.checkStreamExists(config.getScope(), topic);
- }
-
- private boolean createScope() {
- return streamManager.createScope(config.getScope());
- }
-
- private boolean createStream(String topic) {
- StreamConfiguration streamConfiguration = StreamConfiguration.builder().build();
- return streamManager.createStream(config.getScope(), topic, streamConfiguration);
- }
-
- private EventStreamWriter<byte[]> createWrite(String topic) {
- return clientFactory.createEventWriter(topic, new ByteArraySerializer(), EventWriterConfig.builder().build());
- }
-
- private String buildReaderGroupName(boolean isBroadcast, String consumerGroup, String topic) {
- if (isBroadcast) {
- return UUID.randomUUID().toString();
- } else {
- return String.format("%s-%s", consumerGroup, topic);
- }
- }
-
- private String buildReaderId(String instanceName) {
- return String.format("%s-reader", instanceName).replace("\\(", "-").replace("\\)", "-");
- }
-
- private void createReaderGroup(String topic, String readerGroupName) {
- if (!checkTopicExist(topic)) {
- createStream(topic);
- }
- ReaderGroupConfig readerGroupConfig =
- ReaderGroupConfig.builder()
- .stream(NameUtils.getScopedStreamName(config.getScope(), topic))
- .retentionType(ReaderGroupConfig.StreamDataRetention.AUTOMATIC_RELEASE_AT_LAST_CHECKPOINT)
- .build();
- readerGroupManager.createReaderGroup(readerGroupName, readerGroupConfig);
- }
-
- private void deleteReaderGroup(String readerGroup) {
- readerGroupManager.deleteReaderGroup(readerGroup);
- }
-
- private EventStreamReader<byte[]> createReader(String readerId, String readerGroup) {
- return clientFactory.createReader(readerId, readerGroup, new ByteArraySerializer(), ReaderConfig.builder().build());
- }
-
- protected StreamManager getStreamManager() {
- return streamManager;
- }
-
- protected EventStreamClientFactory getClientFactory() {
- return clientFactory;
- }
-
- protected ReaderGroupManager getReaderGroupManager() {
- return readerGroupManager;
- }
-
- public Map<String, EventStreamWriter<byte[]>> getWriterMap() {
- return writerMap;
- }
-
- protected Map<String, SubscribeTask> getSubscribeTaskMap() {
- return subscribeTaskMap;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaCloudEventWriter.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaCloudEventWriter.java
deleted file mode 100644
index 25c59edea..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaCloudEventWriter.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.storage.pravega.client;
-
-import java.nio.charset.StandardCharsets;
-
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
-import io.cloudevents.CloudEventData;
-import io.cloudevents.SpecVersion;
-import io.cloudevents.core.format.EventFormat;
-import io.cloudevents.core.message.MessageWriter;
-import io.cloudevents.rw.CloudEventContextWriter;
-import io.cloudevents.rw.CloudEventRWException;
-import io.cloudevents.rw.CloudEventWriter;
-
-public class PravegaCloudEventWriter
- implements MessageWriter<CloudEventWriter<PravegaEvent>, PravegaEvent>, CloudEventWriter<PravegaEvent> {
-
- private final PravegaEvent pravegaEvent;
-
- public PravegaCloudEventWriter(String topic) {
- pravegaEvent = new PravegaEvent();
- pravegaEvent.setTopic(topic);
- pravegaEvent.setCreateTimestamp(System.currentTimeMillis());
- }
-
- @Override
- public PravegaEvent setEvent(@Nullable EventFormat format, @Nonnull byte[] value) throws CloudEventRWException {
- pravegaEvent.setData(new String(value, StandardCharsets.UTF_8));
- return pravegaEvent;
- }
-
- @Override
- public PravegaEvent end(CloudEventData data) throws CloudEventRWException {
- pravegaEvent.setData(new String(data.toBytes(), StandardCharsets.UTF_8));
- return pravegaEvent;
- }
-
- @Override
- public PravegaEvent end() throws CloudEventRWException {
- pravegaEvent.setData("");
- return pravegaEvent;
- }
-
- @Override
- public CloudEventContextWriter withContextAttribute(@Nonnull String name, @Nonnull String value) throws CloudEventRWException {
- pravegaEvent.getExtensions().put(name, value);
- return this;
- }
-
- @Override
- public CloudEventWriter<PravegaEvent> create(SpecVersion version) throws CloudEventRWException {
- pravegaEvent.setVersion(version);
- return this;
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaEvent.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaEvent.java
deleted file mode 100644
index 4ae85de92..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaEvent.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.storage.pravega.client;
-
-import org.apache.eventmesh.common.utils.JsonUtils;
-import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
-
-import java.io.Serializable;
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Map;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.SpecVersion;
-import io.cloudevents.core.builder.CloudEventBuilder;
-
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@NoArgsConstructor
-public class PravegaEvent implements Serializable {
-
- private static final long serialVersionUID = 0L;
-
- private SpecVersion version;
- private String topic;
- private String data;
- private Map<String, String> extensions = new HashMap<>();
- private long createTimestamp;
-
- public static byte[] toByteArray(PravegaEvent pravegaEvent) {
- return JsonUtils.toJSONString(pravegaEvent).getBytes(StandardCharsets.UTF_8);
- }
-
- public static PravegaEvent getFromByteArray(byte[] body) {
- return JsonUtils.parseObject(new String(body, StandardCharsets.UTF_8), PravegaEvent.class);
- }
-
- public CloudEvent convertToCloudEvent() {
- CloudEventBuilder builder;
- switch (version) {
- case V03:
- builder = CloudEventBuilder.v03();
- break;
- case V1:
- builder = CloudEventBuilder.v1();
- break;
- default:
- throw new PravegaStorageException(String.format("CloudEvent version %s does not support.", version));
- }
- builder.withData(data.getBytes(StandardCharsets.UTF_8))
- .withId(extensions.remove("id"))
- .withSource(URI.create(extensions.remove("source")))
- .withType(extensions.remove("type"))
- .withDataContentType(extensions.remove("datacontenttype"))
- .withSubject(extensions.remove("subject"));
- extensions.forEach(builder::withExtension);
- return builder.build();
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/SubscribeTask.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/SubscribeTask.java
deleted file mode 100644
index 4f26ff85e..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/SubscribeTask.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * 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.storage.pravega.client;
-
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.api.EventMeshAction;
-import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import io.cloudevents.CloudEvent;
-import io.pravega.client.stream.EventRead;
-import io.pravega.client.stream.EventStreamReader;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class SubscribeTask extends Thread {
-
- private final EventStreamReader<byte[]> reader;
- private final EventListener listener;
- private final AtomicBoolean running = new AtomicBoolean(true);
- private final AtomicBoolean continueRead = new AtomicBoolean(true);
-
- public SubscribeTask(String name, EventStreamReader<byte[]> reader, EventListener listener) {
- super(name);
- this.reader = reader;
- this.listener = listener;
- }
-
- @Override
- public void run() {
- CloudEvent cloudEvent = null;
- while (running.get()) {
- if (continueRead.get()) {
- EventRead<byte[]> event = reader.readNextEvent(2000);
-
- byte[] eventByteArray = event.getEvent();
- if (eventByteArray == null) {
- continue;
- }
- PravegaEvent pravegaEvent = PravegaEvent.getFromByteArray(eventByteArray);
- cloudEvent = pravegaEvent.convertToCloudEvent();
-
- }
- listener.consume(cloudEvent, new PravegaEventMeshAsyncConsumeContext());
- }
- }
-
- public void stopRead() {
- running.compareAndSet(true, false);
- }
-
- private class PravegaEventMeshAsyncConsumeContext extends EventMeshAsyncConsumeContext {
-
- @Override
- public void commit(EventMeshAction action) {
- switch (action) {
- case CommitMessage:
- case ReconsumeLater:
- continueRead.set(false);
- break;
- case ManualAck:
- continueRead.set(true);
- break;
- default:
- }
- }
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfig.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfig.java
deleted file mode 100644
index 646d021f0..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfig.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.storage.pravega.config;
-
-import org.apache.eventmesh.common.config.Config;
-import org.apache.eventmesh.common.config.ConfigFiled;
-
-import org.apache.commons.lang3.StringUtils;
-
-import java.net.URI;
-
-import lombok.Getter;
-import lombok.Setter;
-
-@Getter
-@Setter
-@Config(prefix = "eventMesh.server.pravega", path = "classPath://pravega-storage.properties")
-public class PravegaStorageConfig {
-
- @ConfigFiled(field = "controller.uri")
- private URI controllerURI = URI.create("tcp://127.0.0.1:9090");
-
- @ConfigFiled(field = "scope")
- private String scope = "eventmesh-pravega";
-
- @ConfigFiled(field = "clientPool.size")
- private int clientPoolSize = 8;
-
- @ConfigFiled(field = "queue.size")
- private int queueSize = 512;
-
- @ConfigFiled(field = "authEnabled", reload = true)
- private boolean authEnabled = false;
-
- @ConfigFiled(field = "username")
- private String username = "";
-
- @ConfigFiled(field = "password")
- private String password = "";
-
- @ConfigFiled(field = "tlsEnabled")
- private boolean tlsEnable = false;
-
- @ConfigFiled(field = "truststore")
- private String truststore = "";
-
- public void reload() {
- if (!authEnabled && StringUtils.isNotBlank(username) && StringUtils.isNotBlank(password)) {
- authEnabled = true;
- }
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/exception/PravegaStorageException.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/exception/PravegaStorageException.java
deleted file mode 100644
index 50d23cbba..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/exception/PravegaStorageException.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.storage.pravega.exception;
-
-import org.apache.eventmesh.api.exception.StorageRuntimeException;
-
-public class PravegaStorageException extends StorageRuntimeException {
-
- public PravegaStorageException(String message) {
- super(message);
- }
-
- public PravegaStorageException(Throwable throwable) {
- super(throwable);
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
deleted file mode 100644
index 585a0029c..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-pravega=org.apache.eventmesh.storage.pravega.admin.PravegaAdminAdaptor
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
deleted file mode 100644
index 2d68877d3..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-pravega=org.apache.eventmesh.storage.pravega.PravegaConsumerImpl
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
deleted file mode 100644
index b35a24cd4..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-pravega=org.apache.eventmesh.storage.pravega.PravegaProducerImpl
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
deleted file mode 100644
index 64222b4dc..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-pravega=org.apache.eventmesh.storage.pravega.PravegaStorageResourceServiceImpl
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/pravega-storage.properties b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/pravega-storage.properties
deleted file mode 100644
index 25fa6f0af..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/resources/pravega-storage.properties
+++ /dev/null
@@ -1,26 +0,0 @@
-#
-# 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.
-#
-#######################pravega-client##################
-eventMesh.server.pravega.controller.uri=tcp://127.0.0.1:9090
-eventMesh.server.pravega.scope=eventmesh-pravega
-eventMesh.server.pravega.authEnabled=false
-eventMesh.server.pravega.username=
-eventMesh.server.pravega.password=
-eventMesh.server.pravega.tlsEnabled=false
-eventMesh.server.pravega.truststore=
-eventMesh.server.pravega.clientPool.size=8
-eventMesh.server.pravega.queue.size=512
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/client/PravegaClientTest.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/client/PravegaClientTest.java
deleted file mode 100644
index 594d1e0fd..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/client/PravegaClientTest.java
+++ /dev/null
@@ -1,188 +0,0 @@
-/*
- * 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.storage.pravega.client;
-
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import org.apache.eventmesh.api.AsyncConsumeContext;
-import org.apache.eventmesh.api.EventListener;
-import org.apache.eventmesh.common.config.ConfigService;
-import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
-
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-import java.util.function.Consumer;
-
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Rule;
-import org.junit.Test;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.utility.DockerImageName;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.builder.CloudEventBuilder;
-import io.pravega.client.admin.StreamManager;
-import io.pravega.client.stream.ReaderGroupNotFoundException;
-
-import com.github.dockerjava.api.command.CreateContainerCmd;
-import com.github.dockerjava.api.model.PortBinding;
-
-@Ignore
-public class PravegaClientTest {
-
- private StreamManager streamManager;
- private PravegaStorageConfig config;
- private URI controllerURI;
-
- @Rule
- public GenericContainer pravega =
- new GenericContainer(DockerImageName.parse("pravega/pravega:0.11.0"))
- .withExposedPorts(9090, 12345).withEnv("HOST_IP", "127.0.0.1").withCommand("standalone")
- .withCreateContainerCmdModifier((Consumer<CreateContainerCmd>) createContainerCmd ->
- createContainerCmd.getHostConfig()
- .withPortBindings(PortBinding.parse("9090:9090"), PortBinding.parse("12345:12345")));
-
- @Before
- public void setUp() {
- String host = pravega.getHost();
- int port = pravega.getFirstMappedPort();
- controllerURI = URI.create(String.format("tcp://%s:%d", host, port));
- streamManager = StreamManager.create(controllerURI);
- }
-
- @Test
- public void startTest() {
- PravegaClient pravegaClient = getNewPravegaClient();
- pravegaClient.start();
- assertTrue(streamManager.checkScopeExists(config.getScope()));
- }
-
- @Test
- public void shutdownTest() {
- PravegaClient pravegaClient = getNewPravegaClient();
- pravegaClient.start();
- pravegaClient.publish("test1", createCloudEvent());
- pravegaClient.publish("test2", createCloudEvent());
- pravegaClient.publish("test3", createCloudEvent());
-
- pravegaClient.shutdown();
- assertTrue(pravegaClient.getSubscribeTaskMap().isEmpty());
- assertTrue(pravegaClient.getWriterMap().isEmpty());
- }
-
- @Test
- public void publishTest() {
- PravegaClient pravegaClient = getNewPravegaClient();
- pravegaClient.start();
- pravegaClient.publish("test1", createCloudEvent());
- pravegaClient.publish("test2", createCloudEvent());
- assertTrue(streamManager.checkStreamExists(config.getScope(), "test1"));
- assertTrue(streamManager.checkStreamExists(config.getScope(), "test2"));
- }
-
- @Test
- public void subscribeTest() {
- PravegaClient pravegaClient = getNewPravegaClient();
- pravegaClient.start();
- pravegaClient.subscribe("test1", false, "consumerGroup", "instanceName", new EventListener() {
- @Override
- public void consume(CloudEvent cloudEvent, AsyncConsumeContext context) {
- // do nothing
- }
- });
- assertNotNull(pravegaClient.getReaderGroupManager().getReaderGroup("test1-consumerGroup"));
- assertTrue(pravegaClient.getSubscribeTaskMap().containsKey("test1"));
- }
-
- @Test
- public void unsubscribeTest() {
- PravegaClient pravegaClient = getNewPravegaClient();
- pravegaClient.unsubscribe("test1", false, "consumerGroup");
-
- pravegaClient.start();
- pravegaClient.subscribe("test1", false, "consumerGroup", "instanceName", new EventListener() {
- @Override
- public void consume(CloudEvent cloudEvent, AsyncConsumeContext context) {
- // do nothing
- }
- });
- pravegaClient.unsubscribe("test1", false, "consumerGroup");
-
- assertFalse(pravegaClient.getSubscribeTaskMap().containsKey("test1"));
- try {
- pravegaClient.getReaderGroupManager().getReaderGroup("test1-consumerGroup");
- } catch (Exception e) {
- assertTrue(e instanceof ReaderGroupNotFoundException);
- return;
- }
- fail();
- }
-
- @Test
- public void checkTopicExistTest() {
- PravegaClient pravegaClient = getNewPravegaClient();
- pravegaClient.start();
- assertFalse(pravegaClient.checkTopicExist("test1"));
-
- pravegaClient.publish("test1", createCloudEvent());
- assertTrue(pravegaClient.checkTopicExist("test1"));
- }
-
- public PravegaClient getNewPravegaClient() {
- ConfigService configService = ConfigService.getInstance();
-
- this.config = configService.buildConfigInstance(PravegaStorageConfig.class);
- this.config.setControllerURI(controllerURI);
- return PravegaClient.getNewInstance(this.config);
- }
-
- private CloudEvent createCloudEvent() {
- String data = "{\"headers\":{\"content-length\":\"36\",\"Accept\":\"*/*\",\"ip\":\"127.0.0.1:51226\",\"User-Agent\":\"curl/7.83.1\","
- + "\"Host\":\"127.0.0.1:10105\",\"source\":\"127.0.0.1:51226\",\"Content-Type\":\"application/json\"},"
- + "\"path\":\"/eventmesh/publish/TEST-TOPIC-HTTP-ASYNC\",\"method\":\"POST\",\"body\":{\"pass\":\"12345678\",\"name\":\"admin\"}}";
- return CloudEventBuilder.v1()
- .withId("c61039e1-7884-4d7f-b72f-6c61160a64fc")
- .withSource(URI.create("source:127.0.0.1:51226"))
- .withType("http_request")
- .withDataContentType("application/json")
- .withSubject("TEST-TOPIC-HTTP-ASYNC")
- .withData(data.getBytes(StandardCharsets.UTF_8))
- .withExtension("reqeventmesh2mqtimestamp", "1659342713460")
- .withExtension("ip", "127.0.0.1:51226")
- .withExtension("idc", "idc")
- .withExtension("protocoldesc", "http")
- .withExtension("pid", 2376)
- .withExtension("env", "env")
- .withExtension("sys", 1234)
- .withExtension("ttl", 4000)
- .withExtension("producergroup", "em-http-producer")
- .withExtension("consumergroup", "em-http-consumer")
- .withExtension("passwd", "pass")
- .withExtension("bizseqno", "249695004068274968410952665702")
- .withExtension("protocoltype", "http")
- .withExtension("msgtype", "persistent")
- .withExtension("uniqueid", "866012286651006371403062105469")
- .withExtension("username", "eventmesh")
- .withExtension("reqc2eventmeshtimestamp", "1659342713460")
- .build();
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfigTest.java b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfigTest.java
deleted file mode 100644
index e7b24e89d..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfigTest.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.storage.pravega.config;
-
-import org.apache.eventmesh.api.factory.StoragePluginFactory;
-import org.apache.eventmesh.storage.pravega.PravegaConsumerImpl;
-import org.apache.eventmesh.storage.pravega.PravegaProducerImpl;
-
-import java.net.URI;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-public class PravegaStorageConfigTest {
-
- @Test
- public void getConfigWhenPravegaConsumerInit() {
- PravegaConsumerImpl consumer =
- (PravegaConsumerImpl) StoragePluginFactory.getMeshMQPushConsumer("pravega");
-
- PravegaStorageConfig config = consumer.getClientConfiguration();
- assertConfig(config);
- }
-
- @Test
- public void getConfigWhenPravegaProducerInit() {
- PravegaProducerImpl producer =
- (PravegaProducerImpl) StoragePluginFactory.getMeshMQProducer("pravega");
-
- PravegaStorageConfig config = producer.getClientConfiguration();
- assertConfig(config);
- }
-
- private void assertConfig(PravegaStorageConfig config) {
- Assert.assertEquals(URI.create("tcp://127.0.0.1:816"), config.getControllerURI());
- Assert.assertEquals("scope-success!!!", config.getScope());
- Assert.assertTrue(config.isAuthEnabled());
- Assert.assertEquals("username-success!!!", config.getUsername());
- Assert.assertEquals("password-success!!!", config.getPassword());
- Assert.assertTrue(config.isTlsEnable());
- Assert.assertEquals("truststore-success!!!", config.getTruststore());
- Assert.assertEquals(816, config.getClientPoolSize());
- Assert.assertEquals(1816, config.getQueueSize());
- }
-}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/resources/pravega-storage.properties b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/resources/pravega-storage.properties
deleted file mode 100644
index 5bb5e5490..000000000
--- a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/test/resources/pravega-storage.properties
+++ /dev/null
@@ -1,26 +0,0 @@
-#
-# 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.
-#
-#######################pravega-client##################
-eventMesh.server.pravega.controller.uri=tcp://127.0.0.1:816
-eventMesh.server.pravega.scope=scope-success!!!
-eventMesh.server.pravega.authEnabled=false
-eventMesh.server.pravega.username=username-success!!!
-eventMesh.server.pravega.password=password-success!!!
-eventMesh.server.pravega.tlsEnabled=true
-eventMesh.server.pravega.truststore=truststore-success!!!
-eventMesh.server.pravega.clientPool.size=816
-eventMesh.server.pravega.queue.size=1816
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
index a1aad7463..ac21f39ff 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -43,10 +43,7 @@ include 'eventmesh-connectors:eventmesh-connector-jdbc'
include 'eventmesh-storage-plugin:eventmesh-storage-api'
include 'eventmesh-storage-plugin:eventmesh-storage-standalone'
include 'eventmesh-storage-plugin:eventmesh-storage-kafka'
-include 'eventmesh-storage-plugin:eventmesh-storage-knative'
-include 'eventmesh-storage-plugin:eventmesh-storage-mongodb'
include 'eventmesh-storage-plugin:eventmesh-storage-pulsar'
-include 'eventmesh-storage-plugin:eventmesh-storage-pravega'
include 'eventmesh-storage-plugin:eventmesh-storage-redis'
include 'eventmesh-storage-plugin:eventmesh-storage-rocketmq'
include 'eventmesh-storage-plugin:eventmesh-storage-rabbitmq'
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org