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/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 65e90dbff [ISSUE #3380] Add eventmesh-storage-pulsar module
new ed162b70d Merge pull request #3383 from xwm1992/add-pulsar-storage
65e90dbff is described below
commit 65e90dbff44529fa792ae988d610770916c64c7f
Author: xwm1992 <[email protected]>
AuthorDate: Thu Mar 9 22:06:33 2023 +0800
[ISSUE #3380] Add eventmesh-storage-pulsar module
---
.../eventmesh-storage-pulsar/build.gradle | 45 +++++
.../eventmesh-storage-pulsar/gradle.properties | 18 ++
.../storage/pulsar/client/PulsarClientWrapper.java | 120 ++++++++++++
.../storage/pulsar/config/ClientConfiguration.java | 46 +++++
.../storage/pulsar/constant/PulsarConstant.java | 23 +++
.../pulsar/consumer/PulsarConsumerImpl.java | 208 +++++++++++++++++++++
.../storage/pulsar/producer/AbstractProducer.java | 55 ++++++
.../storage/pulsar/producer/ProducerImpl.java | 83 ++++++++
.../pulsar/producer/PulsarProducerImpl.java | 99 ++++++++++
.../storage/StorageResourceServicePulsarImpl.java | 33 ++++
.../storage/pulsar/utils/CloudEventUtils.java | 32 ++++
.../org.apache.eventmesh.api.consumer.Consumer | 16 ++
.../org.apache.eventmesh.api.producer.Producer | 16 ++
...he.eventmesh.api.storage.StorageResourceService | 16 ++
.../src/main/resources/pulsar-client.properties | 18 ++
.../pulsar/config/ClientConfigurationTest.java | 50 +++++
.../src/test/resources/pulsar-client.properties | 20 ++
settings.gradle | 1 +
18 files changed, 899 insertions(+)
diff --git a/eventmesh-storage/eventmesh-storage-pulsar/build.gradle
b/eventmesh-storage/eventmesh-storage-pulsar/build.gradle
new file mode 100644
index 000000000..eb4da5c65
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-pulsar/build.gradle
@@ -0,0 +1,45 @@
+/*
+ * 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:eventmesh-storage-api")
+ implementation 'org.apache.pulsar:pulsar-client:2.10.1'
+
+ testImplementation project(":eventmesh-storage:eventmesh-storage-api")
+ testImplementation project(":eventmesh-common")
+ testImplementation 'org.apache.pulsar:pulsar-client:2.10.1'
+
+ 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'
+ annotationProcessor 'org.projectlombok:lombok'
+
+ testCompileOnly 'org.projectlombok:lombok'
+ testAnnotationProcessor 'org.projectlombok:lombok'
+}
diff --git a/eventmesh-storage/eventmesh-storage-pulsar/gradle.properties
b/eventmesh-storage/eventmesh-storage-pulsar/gradle.properties
new file mode 100644
index 000000000..da19b5bdd
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-pulsar/gradle.properties
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+pluginType=storage
+pluginName=pulsar
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/client/PulsarClientWrapper.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/client/PulsarClientWrapper.java
new file mode 100644
index 000000000..9ad3f95ae
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/client/PulsarClientWrapper.java
@@ -0,0 +1,120 @@
+/*
+ * 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.pulsar.client;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.storage.pulsar.config.ClientConfiguration;
+import org.apache.eventmesh.storage.pulsar.utils.CloudEventUtils;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.jackson.JsonFormat;
+
+import com.google.common.base.Preconditions;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarClientWrapper {
+
+ private ClientConfiguration config;
+ private PulsarClient pulsarClient;
+ private Map<String, Producer<byte[]>> producerMap = new HashMap<>();
+
+ public PulsarClientWrapper(ClientConfiguration config, Properties
properties) {
+ this.config = config;
+ String token = properties.getProperty(Constants.PRODUCER_TOKEN);
+ try {
+ ClientBuilder clientBuilder = PulsarClient.builder()
+ .serviceUrl(config.getServiceAddr());
+
+ if (config.getAuthPlugin() != null) {
+ Preconditions.checkNotNull(config.getAuthParams(),
+ "Authentication Enabled in pulsar cluster, Please set
authParams in pulsar-client.properties");
+ clientBuilder.authentication(
+ config.getAuthPlugin(),
+ config.getAuthParams()
+ );
+ }
+ if (StringUtils.isNotBlank(token)) {
+ clientBuilder.authentication(
+ AuthenticationFactory.token(token)
+ );
+ }
+
+ this.pulsarClient = clientBuilder.build();
+ } catch (PulsarClientException ex) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Failed to connect pulsar cluster %s with
exception: %s", config.getServiceAddr(), ex.getMessage()));
+ }
+ }
+
+ private Producer<byte[]> createProducer(String topic) {
+ try {
+ return this.pulsarClient.newProducer()
+ .topic(topic)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
+ .sendTimeout(10, TimeUnit.SECONDS)
+ .blockIfQueueFull(true)
+ .create();
+ } catch (PulsarClientException ex) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Failed to create pulsar producer for %s with
exception: %s", topic, ex.getMessage()));
+ }
+ }
+
+ public void publish(CloudEvent cloudEvent, SendCallback sendCallback) {
+ String topic = config.getTopicPrefix() + cloudEvent.getSubject();
+ Producer<byte[]> producer = producerMap.computeIfAbsent(topic, k ->
createProducer(topic));
+ try {
+ byte[] serializedCloudEvent = EventFormatProvider
+ .getInstance()
+ .resolveFormat(JsonFormat.CONTENT_TYPE)
+ .serialize(cloudEvent);
+ producer.sendAsync(serializedCloudEvent).thenAccept(messageId -> {
+
sendCallback.onSuccess(CloudEventUtils.convertSendResult(cloudEvent));
+ });
+ } catch (Exception ex) {
+ log.error("Failed to publish cloudEvent for {} with exception: {}",
+ cloudEvent.getSubject(), ex.getMessage());
+ }
+ }
+
+ public void shutdown() throws PulsarClientException {
+ pulsarClient.close();
+ for (Map.Entry<String, Producer<byte[]>> producerEntry :
producerMap.entrySet()) {
+ producerEntry.getValue().close();
+ }
+ producerMap.clear();
+ }
+
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/config/ClientConfiguration.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/config/ClientConfiguration.java
new file mode 100644
index 000000000..5dc94accc
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/config/ClientConfiguration.java
@@ -0,0 +1,46 @@
+/*
+ * 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.pulsar.config;
+
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.common.config.ConfigFiled;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+@Config(prefix = "eventMesh.server.pulsar", path =
"classPath://pulsar-client.properties")
+public class ClientConfiguration {
+
+ @ConfigFiled(field = "service")
+ private String serviceAddr;
+
+ @ConfigFiled(field = "authPlugin")
+ private String authPlugin;
+
+ @ConfigFiled(field = "authParams")
+ private String authParams;
+
+ /*
+ * the full format of topic needs a prefix, but the prefix cannot be
passed in the url when the topic is carried
+ *
+ * */
+ @ConfigFiled(field = "topicPrefix")
+ private String topicPrefix;
+}
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/constant/PulsarConstant.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/constant/PulsarConstant.java
new file mode 100644
index 000000000..9c14f2aac
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/constant/PulsarConstant.java
@@ -0,0 +1,23 @@
+/*
+ * 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.pulsar.constant;
+
+public class PulsarConstant {
+
+ public static final String KEY_SEPARATOR = "/";
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/consumer/PulsarConsumerImpl.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/consumer/PulsarConsumerImpl.java
new file mode 100644
index 000000000..bafd63745
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/consumer/PulsarConsumerImpl.java
@@ -0,0 +1,208 @@
+/*
+ * 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.pulsar.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.api.exception.StorageConnectorRuntimeException;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.storage.pulsar.config.ClientConfiguration;
+import org.apache.eventmesh.storage.pulsar.constant.PulsarConstant;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventDeserializationException;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.jackson.JsonFormat;
+
+import com.google.common.base.Preconditions;
+
+import lombok.extern.slf4j.Slf4j;
+
+
+@Slf4j
+@Config(field = "clientConfiguration")
+public class PulsarConsumerImpl implements Consumer {
+
+ private final AtomicBoolean started = new AtomicBoolean(false);
+ private Properties properties;
+ private PulsarClient pulsarClient;
+ private EventListener eventListener;
+
+ private ConcurrentHashMap<String,
org.apache.pulsar.client.api.Consumer<byte[]>> consumerMap = new
ConcurrentHashMap<>();
+
+ /**
+ * Unified configuration class corresponding to pulsar-client.properties
+ */
+ private ClientConfiguration clientConfiguration;
+
+ @Override
+ public void init(Properties properties) throws Exception {
+ this.properties = properties;
+ String token = properties.getProperty(Constants.CONSUMER_TOKEN);
+
+ try {
+ ClientBuilder clientBuilder = PulsarClient.builder()
+ .serviceUrl(clientConfiguration.getServiceAddr());
+
+ if (clientConfiguration.getAuthPlugin() != null) {
+ Preconditions.checkNotNull(clientConfiguration.getAuthParams(),
+ "Authentication Enabled in pulsar cluster, Please set
authParams in pulsar-client.properties");
+ clientBuilder.authentication(
+ clientConfiguration.getAuthPlugin(),
+ clientConfiguration.getAuthParams()
+ );
+ }
+ if (StringUtils.isNotBlank(token)) {
+ clientBuilder.authentication(
+ AuthenticationFactory.token(token)
+ );
+ }
+
+ this.pulsarClient = clientBuilder.build();
+ } catch (Exception ex) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Failed to connect pulsar with exception: %s",
ex.getMessage()));
+ }
+ }
+
+ @Override
+ public void start() {
+ this.started.compareAndSet(false, true);
+ }
+
+ @Override
+ public void subscribe(String topic) throws Exception {
+ String subTopic = clientConfiguration.getTopicPrefix() + topic;
+ if (pulsarClient == null) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Cann't find the pulsar client for topic: %s",
subTopic));
+ }
+
+ EventMeshAsyncConsumeContext consumeContext = new
EventMeshAsyncConsumeContext() {
+ @Override
+ public void commit(EventMeshAction action) {
+ log.debug("message action: {} for topic: {}", action.name(),
subTopic);
+ }
+ };
+
+ SubscriptionType type = SubscriptionType.Shared;
+
+ String consumerKey = topic + PulsarConstant.KEY_SEPARATOR +
properties.getProperty(Constants.CONSUMER_GROUP)
+ + PulsarConstant.KEY_SEPARATOR +
properties.getProperty(Constants.CLIENT_ADDRESS);
+ org.apache.pulsar.client.api.Consumer<byte[]> consumer =
pulsarClient.newConsumer()
+ .topic(subTopic)
+ .subscriptionName(properties.getProperty(Constants.CONSUMER_GROUP))
+ .subscriptionMode(SubscriptionMode.Durable)
+ .subscriptionType(type)
+ .messageListener(
+ (MessageListener<byte[]>) (ackConsumer, msg) -> {
+ CloudEvent cloudEvent = EventFormatProvider
+ .getInstance()
+ .resolveFormat(JsonFormat.CONTENT_TYPE)
+ .deserialize(msg.getData());
+ eventListener.consume(cloudEvent, consumeContext);
+ try {
+ ackConsumer.acknowledge(msg);
+ } catch (PulsarClientException ex) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Failed to unsubscribe the topic:%s
with exception: %s", subTopic, ex.getMessage()));
+ } catch (EventDeserializationException ex) {
+ log.warn("The Message isn't json format, with
exception:{}", ex.getMessage());
+ }
+ }).subscribe();
+
+ consumerMap.putIfAbsent(consumerKey, consumer);
+
+ }
+
+ @Override
+ public void unsubscribe(String topic) {
+ try {
+ String consumerKey = topic + PulsarConstant.KEY_SEPARATOR +
properties.getProperty(Constants.CONSUMER_GROUP)
+ + PulsarConstant.KEY_SEPARATOR +
properties.getProperty(Constants.CLIENT_ADDRESS);
+ org.apache.pulsar.client.api.Consumer<byte[]> consumer =
consumerMap.get(consumerKey);
+ consumer.unsubscribe();
+ consumerMap.remove(consumerKey);
+ } catch (PulsarClientException ex) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Failed to unsubscribe the topic:%s with
exception: %s", topic, ex.getMessage()));
+ }
+ }
+
+ @Override
+ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext
context) {
+ }
+
+ @Override
+ public void registerEventListener(EventListener listener) {
+ this.eventListener = listener;
+ }
+
+ @Override
+ public boolean isStarted() {
+ return this.started.get();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !this.isStarted();
+ }
+
+ @Override
+ public void shutdown() {
+ this.started.compareAndSet(true, false);
+ try {
+
+ consumerMap.forEach((key, consumer) -> {
+ try {
+ consumer.close();
+ } catch (PulsarClientException e) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Failed to close the pulsar consumer
with exception: %s", e.getMessage()));
+ }
+ });
+ this.pulsarClient.close();
+ consumerMap.clear();
+ } catch (PulsarClientException ex) {
+ throw new StorageConnectorRuntimeException(
+ String.format("Failed to close the pulsar client with
exception: %s", ex.getMessage()));
+ }
+ }
+
+ public ClientConfiguration getClientConfiguration() {
+ return this.clientConfiguration;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/AbstractProducer.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/AbstractProducer.java
new file mode 100644
index 000000000..4137f8620
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/AbstractProducer.java
@@ -0,0 +1,55 @@
+/*
+ * 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.pulsar.producer;
+
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+
+import io.cloudevents.CloudEvent;
+
+public abstract class AbstractProducer {
+
+ protected final AtomicBoolean started = new AtomicBoolean(false);
+ final Properties properties;
+
+ AbstractProducer(final Properties properties) {
+ this.properties = properties;
+ }
+
+ public Properties properties() {
+ return this.properties;
+ }
+
+ StorageConnectorRuntimeException checkProducerException(CloudEvent
cloudEvent, Throwable e) {
+ if (cloudEvent.getData() == null) {
+ return new
StorageConnectorRuntimeException(String.format("CloudEvent message data does
not exist, %s", e.getMessage()));
+ }
+ return new StorageConnectorRuntimeException(String.format("Unknown
connector runtime exception, %s", e.getMessage()));
+ }
+
+ public boolean isStarted() {
+ return this.started.get();
+ }
+
+ public boolean isClosed() {
+ return !this.isStarted();
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/ProducerImpl.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/ProducerImpl.java
new file mode 100644
index 000000000..343634393
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/ProducerImpl.java
@@ -0,0 +1,83 @@
+/*
+ * 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.pulsar.producer;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.storage.pulsar.client.PulsarClientWrapper;
+import org.apache.eventmesh.storage.pulsar.config.ClientConfiguration;
+
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class ProducerImpl extends AbstractProducer {
+
+ private final AtomicBoolean started = new AtomicBoolean(false);
+
+ private ClientConfiguration config;
+ private PulsarClientWrapper pulsarClient;
+
+ public ProducerImpl(final Properties properties, ClientConfiguration
config) {
+ this(properties);
+ setConfig(config);
+ }
+
+ public ProducerImpl(final Properties properties) {
+ super(properties);
+ }
+
+ public void publish(CloudEvent cloudEvent, SendCallback sendCallback) {
+ this.pulsarClient.publish(cloudEvent, sendCallback);
+ }
+
+ public void init(Properties properties) {
+ new ProducerImpl(properties);
+ }
+
+ public void start() {
+ this.started.compareAndSet(false, true);
+ this.pulsarClient = new PulsarClientWrapper(config, properties);
+ }
+
+ public void shutdown() {
+ try {
+ this.started.compareAndSet(true, false);
+ this.pulsarClient.shutdown();
+ } catch (Exception ignored) {
+ // ignored
+ }
+ }
+
+ @Override
+ public boolean isStarted() {
+ return this.started.get();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !this.isStarted();
+ }
+
+ public void setConfig(ClientConfiguration config) {
+ this.config = config;
+ }
+}
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/PulsarProducerImpl.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/PulsarProducerImpl.java
new file mode 100644
index 000000000..11dfbc3f2
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/producer/PulsarProducerImpl.java
@@ -0,0 +1,99 @@
+/*
+ * 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.pulsar.producer;
+
+import org.apache.eventmesh.api.RequestReplyCallback;
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.storage.pulsar.config.ClientConfiguration;
+
+import java.util.Properties;
+
+import io.cloudevents.CloudEvent;
+
+@Config(field = "clientConfiguration")
+public class PulsarProducerImpl implements Producer {
+
+ private ProducerImpl producer;
+
+ /**
+ * Unified configuration class corresponding to pulsar-client.properties
+ */
+ private ClientConfiguration clientConfiguration;
+
+ @Override
+ public synchronized void init(Properties properties) {
+ producer = new ProducerImpl(properties, clientConfiguration);
+ }
+
+ @Override
+ public void publish(CloudEvent cloudEvent, SendCallback sendCallback)
throws Exception {
+ producer.publish(cloudEvent, sendCallback);
+ }
+
+ @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 sendOneway(CloudEvent cloudEvent) {
+ throw new StorageConnectorRuntimeException("SendOneWay is not
supported");
+ }
+
+ @Override
+ public void request(CloudEvent cloudEvent, RequestReplyCallback
rrCallback, long timeout) throws Exception {
+ throw new StorageConnectorRuntimeException("Request is not supported");
+ }
+
+ @Override
+ public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback)
throws Exception {
+ throw new StorageConnectorRuntimeException("Reply is not supported");
+ }
+
+ @Override
+ public void checkTopicExist(String topic) throws Exception {
+ throw new StorageConnectorRuntimeException("CheckTopicExist is not
supported");
+ }
+
+ @Override
+ public void setExtFields() {
+ throw new StorageConnectorRuntimeException("SetExtFields is not
supported");
+ }
+
+ public ClientConfiguration getClientConfiguration() {
+ return this.clientConfiguration;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/storage/StorageResourceServicePulsarImpl.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/storage/StorageResourceServicePulsarImpl.java
new file mode 100644
index 000000000..7b09cc685
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/storage/StorageResourceServicePulsarImpl.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.pulsar.storage;
+
+import org.apache.eventmesh.api.storage.StorageResourceService;
+
+public class StorageResourceServicePulsarImpl implements
StorageResourceService {
+
+ @Override
+ public void init() throws Exception {
+
+ }
+
+ @Override
+ public void release() throws Exception {
+
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/utils/CloudEventUtils.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/utils/CloudEventUtils.java
new file mode 100644
index 000000000..71d40d29c
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/utils/CloudEventUtils.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.pulsar.utils;
+
+import org.apache.eventmesh.api.SendResult;
+
+import io.cloudevents.CloudEvent;
+
+public class CloudEventUtils {
+
+ public static SendResult convertSendResult(CloudEvent cloudEvent) {
+ SendResult sendResult = new SendResult();
+ sendResult.setTopic(cloudEvent.getSubject());
+ sendResult.setMessageId(cloudEvent.getId());
+ return sendResult;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
new file mode 100644
index 000000000..6716ddce4
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/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.
+
+pulsar=org.apache.eventmesh.storage.pulsar.consumer.PulsarConsumerImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
new file mode 100644
index 000000000..235173dea
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/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.
+
+pulsar=org.apache.eventmesh.storage.pulsar.producer.PulsarProducerImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
new file mode 100644
index 000000000..4b210007a
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/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.
+
+pulsar=org.apache.eventmesh.storage.pulsar.storage.StorageResourceServicePulsarImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/pulsar-client.properties
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/pulsar-client.properties
new file mode 100644
index 000000000..cea69024e
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/main/resources/pulsar-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.
+#
+
+eventMesh.server.pulsar.service=127.0.0.1:6650
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/test/java/org/apache/eventmesh/storage/pulsar/config/ClientConfigurationTest.java
b/eventmesh-storage/eventmesh-storage-pulsar/src/test/java/org/apache/eventmesh/storage/pulsar/config/ClientConfigurationTest.java
new file mode 100644
index 000000000..40ffb3fdc
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/test/java/org/apache/eventmesh/storage/pulsar/config/ClientConfigurationTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.pulsar.config;
+
+import org.apache.eventmesh.api.factory.StoragePluginFactory;
+import org.apache.eventmesh.storage.pulsar.consumer.PulsarConsumerImpl;
+import org.apache.eventmesh.storage.pulsar.producer.PulsarProducerImpl;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ClientConfigurationTest {
+
+ @Test
+ public void getConfigWhenPulsarConsumerInit() {
+ PulsarConsumerImpl consumer = (PulsarConsumerImpl)
StoragePluginFactory.getMeshMQPushConsumer("pulsar");
+
+ ClientConfiguration config = consumer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ @Test
+ public void getConfigWhenPulsarProducerInit() {
+ PulsarProducerImpl producer = (PulsarProducerImpl)
StoragePluginFactory.getMeshMQProducer("pulsar");
+
+ ClientConfiguration config = producer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ private void assertConfig(ClientConfiguration config) {
+ Assert.assertEquals(config.getServiceAddr(), "127.0.0.1:6650");
+ Assert.assertEquals(config.getAuthPlugin(), "authPlugin-success!!!");
+ Assert.assertEquals(config.getAuthParams(), "authParams-success!!!");
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pulsar/src/test/resources/pulsar-client.properties
b/eventmesh-storage/eventmesh-storage-pulsar/src/test/resources/pulsar-client.properties
new file mode 100644
index 000000000..645a5edb6
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pulsar/src/test/resources/pulsar-client.properties
@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+eventMesh.server.pulsar.service=127.0.0.1:6650
+eventMesh.server.pulsar.authPlugin=authPlugin-success!!!
+eventMesh.server.pulsar.authParams=authParams-success!!!
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
index 3449270a4..826b0a8bb 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -26,6 +26,7 @@ include 'eventmesh-spi'
include 'eventmesh-storage:eventmesh-storage-api'
include 'eventmesh-storage:eventmesh-storage-standalone'
include 'eventmesh-storage:eventmesh-storage-kafka'
+include 'eventmesh-storage:eventmesh-storage-pulsar'
include 'eventmesh-storage:eventmesh-storage-pravega'
include 'eventmesh-storage:eventmesh-storage-redis'
include 'eventmesh-storage:eventmesh-storage-rocketmq'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
