This is an automated email from the ASF dual-hosted git repository.
mytang0 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 6922664b0 #3484 Did code optimization in PulsarClientWrapper
new d4715903e Merge pull request #3600 from bsh786/bsh786/issue#3484
6922664b0 is described below
commit 6922664b043e3b5240706ad5fa34b7c5cbd88146
Author: bsh786 <[email protected]>
AuthorDate: Sun Apr 2 03:42:19 2023 +0530
#3484 Did code optimization in PulsarClientWrapper
---
.../eventmesh/storage/pulsar/client/PulsarClientWrapper.java | 11 ++++++-----
1 file changed, 6 insertions(+), 5 deletions(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/client/PulsarClientWrapper.java
b/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/client/PulsarClientWrapper.java
index 0dc2ce12e..491ce01f2 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/client/PulsarClientWrapper.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/client/PulsarClientWrapper.java
@@ -32,6 +32,7 @@ import org.apache.pulsar.client.api.PulsarClientException;
import java.util.HashMap;
import java.util.Map;
+import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
@@ -46,9 +47,9 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class PulsarClientWrapper {
- private ClientConfiguration config;
- private PulsarClient pulsarClient;
- private Map<String, Producer<byte[]>> producerMap = new HashMap<>();
+ private final ClientConfiguration config;
+ private final PulsarClient pulsarClient;
+ private final Map<String, Producer<byte[]>> producerMap = new HashMap<>();
public PulsarClientWrapper(ClientConfiguration config, Properties
properties) {
this.config = config;
@@ -96,9 +97,9 @@ public class PulsarClientWrapper {
String topic = config.getTopicPrefix() + cloudEvent.getSubject();
Producer<byte[]> producer = producerMap.computeIfAbsent(topic, k ->
createProducer(topic));
try {
- byte[] serializedCloudEvent = EventFormatProvider
+ byte[] serializedCloudEvent =
Objects.requireNonNull(EventFormatProvider
.getInstance()
- .resolveFormat(JsonFormat.CONTENT_TYPE)
+ .resolveFormat(JsonFormat.CONTENT_TYPE))
.serialize(cloudEvent);
producer.sendAsync(serializedCloudEvent).thenAccept(messageId -> {
sendCallback.onSuccess(CloudEventUtils.convertSendResult(cloudEvent));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
