This is an automated email from the ASF dual-hosted git repository.
chenguangsheng 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 3bf4d693e [ISSUE #3553] Add eventmesh-storage-knative module
new 10485b8d6 Merge pull request #3554 from xwm1992/add-storage-knative
3bf4d693e is described below
commit 3bf4d693eb97787f6707f396125c047e5da4e3fe
Author: xwm1992 <[email protected]>
AuthorDate: Wed Mar 29 17:13:03 2023 +0800
[ISSUE #3553] Add eventmesh-storage-knative module
---
.../eventmesh-storage-knative/build.gradle | 40 +++++
.../eventmesh-storage-knative/gradle.properties | 18 +++
.../knative/cloudevent/KnativeMessageFactory.java | 47 ++++++
.../impl/KnativeBinaryMessageReader.java | 57 +++++++
.../knative/cloudevent/impl/KnativeHeaders.java | 31 ++++
.../cloudevent/impl/KnativeMessageWriter.java | 80 ++++++++++
.../knative/config/ClientConfiguration.java | 51 ++++++
.../StorageResourceServiceKnativeImpl.java | 32 ++++
.../storage/knative/consumer/DefaultConsumer.java | 75 +++++++++
.../knative/consumer/KnativeConsumerImpl.java | 105 ++++++++++++
.../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/knative/utils/CloudEventUtils.java | 33 ++++
.../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 +++
settings.gradle | 1 +
28 files changed, 1406 insertions(+)
diff --git a/eventmesh-storage-plugin/eventmesh-storage-knative/build.gradle
b/eventmesh-storage-plugin/eventmesh-storage-knative/build.gradle
new file mode 100644
index 000000000..7bdfac7e7
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-knative/build.gradle
@@ -0,0 +1,40 @@
+/*
+ * 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-sdk-java")
+ testImplementation project(":eventmesh-sdk-java")
+
+ implementation project(":eventmesh-runtime")
+ testImplementation project(":eventmesh-runtime")
+
+ 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
new file mode 100644
index 000000000..8ba4f21ce
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-knative/gradle.properties
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+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/cloudevent/KnativeMessageFactory.java
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/KnativeMessageFactory.java
new file mode 100644
index 000000000..91ed60778
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/KnativeMessageFactory.java
@@ -0,0 +1,47 @@
+/*
+ * 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
new file mode 100644
index 000000000..8a3060cab
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeBinaryMessageReader.java
@@ -0,0 +1,57 @@
+/*
+ * 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
new file mode 100644
index 000000000..1f998a4b2
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeHeaders.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.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
new file mode 100644
index 000000000..5c8df1b5f
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/cloudevent/impl/KnativeMessageWriter.java
@@ -0,0 +1,80 @@
+/*
+ * 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.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+
+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 (message != null && message.getData() != null) {
+ return message.getData().toString();
+ }
+ throw CloudEventRWException.newOther(new
IllegalArgumentException("message data is null"));
+ }
+
+ @Override
+ public CloudEventContextWriter withContextAttribute(String name, String
value) throws CloudEventRWException {
+ return null;
+ }
+
+ @Override
+ public String setEvent(EventFormat format, 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
new file mode 100644
index 000000000..5c0630470
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/config/ClientConfiguration.java
@@ -0,0 +1,51 @@
+/*
+ * 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/connector/StorageResourceServiceKnativeImpl.java
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/connector/StorageResourceServiceKnativeImpl.java
new file mode 100644
index 000000000..9fd46e8b6
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/connector/StorageResourceServiceKnativeImpl.java
@@ -0,0 +1,32 @@
+/*
+ * 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.connector;
+
+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/consumer/DefaultConsumer.java
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/DefaultConsumer.java
new file mode 100644
index 000000000..2130ed852
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/DefaultConsumer.java
@@ -0,0 +1,75 @@
+/*
+ * 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
new file mode 100644
index 000000000..4b820d95f
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java
@@ -0,0 +1,105 @@
+/*
+ * 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.cloudevents.CloudEvent;
+
+@Config(field = "clientConfiguration")
+public class KnativeConsumerImpl implements Consumer {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KnativeConsumerImpl.class);
+
+ private transient PullConsumerImpl pullConsumer;
+
+
+ /**
+ * Unified configuration class corresponding to knative-client.properties
+ */
+ private ClientConfiguration clientConfiguration;
+
+ private static final Logger logger =
LoggerFactory.getLogger(KnativeConsumerImpl.class);
+
+ @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
new file mode 100644
index 000000000..c3378d05f
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/PullConsumerImpl.java
@@ -0,0 +1,177 @@
+/*
+ * 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
new file mode 100644
index 000000000..7c829afb2
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/domain/NonStandardKeys.java
@@ -0,0 +1,27 @@
+/*
+ * 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
new file mode 100644
index 000000000..0c0fb9b12
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyContext.java
@@ -0,0 +1,30 @@
+/*
+ * 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
new file mode 100644
index 000000000..3a2e2f30e
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshConsumeConcurrentlyStatus.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.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
new file mode 100644
index 000000000..6a0512fe3
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/patch/EventMeshMessageListenerConcurrently.java
@@ -0,0 +1,24 @@
+/*
+ * 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
new file mode 100644
index 000000000..0ef390dcf
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/AbstractProducer.java
@@ -0,0 +1,67 @@
+/*
+ * 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
new file mode 100644
index 000000000..7817436e2
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImpl.java
@@ -0,0 +1,101 @@
+/*
+ * 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
new file mode 100644
index 000000000..fce4bf67a
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/producer/ProducerImpl.java
@@ -0,0 +1,128 @@
+/*
+ * 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.exception.OnExceptionContext;
+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/utils/CloudEventUtils.java
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/utils/CloudEventUtils.java
new file mode 100644
index 000000000..9b2b4c96a
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/utils/CloudEventUtils.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.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.consumer.Consumer
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
new file mode 100644
index 000000000..6b10bacce
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+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
new file mode 100644
index 000000000..bc5f8b41e
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+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
new file mode 100644
index 000000000..06c107570
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+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
new file mode 100644
index 000000000..8fec1c430
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/resources/knative-client.properties
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#######################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
new file mode 100644
index 000000000..9ff668a1c
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/config/ClientConfigurationTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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(config.getEmurl(), "127.0.0.1");
+ Assert.assertEquals(config.getServiceAddr(),
"cloudevents-player.default.127.0.0.1.sslip.io");
+ }
+}
\ 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
new file mode 100644
index 000000000..d2888ffb9
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImplTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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
new file mode 100644
index 000000000..0ae77db30
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/java/org/apache/eventmesh/storage/knative/producer/KnativeProducerImplTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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
new file mode 100644
index 000000000..8fec1c430
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/test/resources/knative-client.properties
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#######################knative-client##################
+eventMesh.server.knative.service=127.0.0.1;cloudevents-player.default.127.0.0.1.sslip.io
diff --git a/settings.gradle b/settings.gradle
index 39e84e29a..488af68c3 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -26,6 +26,7 @@ include 'eventmesh-spi'
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'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
