You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2021/11/26 13:04:54 UTC
[incubator-eventmesh] branch cloudevents updated: Java sdk update (#615)
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch cloudevents
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/cloudevents by this push:
new f7d0fbb Java sdk update (#615)
f7d0fbb is described below
commit f7d0fbb5113d8477117779455bb150c3a6985dad
Author: mike_xwm <mi...@126.com>
AuthorDate: Fri Nov 26 19:04:02 2021 +0800
Java sdk update (#615)
* update java sdk
* fix compile error
* fix sdk error
1.remove the openmessage from connector-api
2.fix the standalone connector
* 1.fix the standalone connector
2.fix the sdk
---
.../main/java/org/apache/eventmesh/common/Constants.java | 6 +++---
.../standalone/consumer/StandaloneConsumerAdaptor.java | 16 ++++++++--------
.../tcp/demo/pub/eventmeshmessage/AsyncPublish.java | 1 -
.../tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java | 12 ++++++++++--
.../org.apache.eventmesh.protocol.api.ProtocolAdaptor | 0
.../tcp/impl/cloudevent/CloudEventTCPPubClient.java | 5 +++--
.../eventmeshmessage/EventMeshMessageTCPPubClient.java | 15 ++++++++-------
.../eventmeshmessage/EventMeshMessageTCPSubClient.java | 8 ++++----
8 files changed, 36 insertions(+), 27 deletions(-)
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
index f5fa2cd..20c1b4e 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
@@ -29,11 +29,11 @@ public class Constants {
public static final String HTTPS_PROTOCOL_PREFIX = "https://";
- public static final String PROTOCOL_TYPE = "protocol_type";
+ public static final String PROTOCOL_TYPE = "protocoltype";
- public static final String PROTOCOL_VERSION = "protocol_version";
+ public static final String PROTOCOL_VERSION = "protocolversion";
- public static final String PROTOCOL_DESC = "protocol_desc";
+ public static final String PROTOCOL_DESC = "protocoldesc";
public static final int DEFAULT_HTTP_TIME_OUT = 3000;
diff --git a/eventmesh-connector-plugin/eventmesh-connector-standalone/src/main/java/org/apache/eventmesh/connector/standalone/consumer/StandaloneConsumerAdaptor.java b/eventmesh-connector-plugin/eventmesh-connector-standalone/src/main/java/org/apache/eventmesh/connector/standalone/consumer/StandaloneConsumerAdaptor.java
index 56518a7..716b4a2 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-standalone/src/main/java/org/apache/eventmesh/connector/standalone/consumer/StandaloneConsumerAdaptor.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-standalone/src/main/java/org/apache/eventmesh/connector/standalone/consumer/StandaloneConsumerAdaptor.java
@@ -40,42 +40,42 @@ public class StandaloneConsumerAdaptor implements Consumer {
@Override
public boolean isStarted() {
- return false;
+ return consumer.isStarted();
}
@Override
public boolean isClosed() {
- return false;
+ return consumer.isClosed();
}
@Override
public void start() {
-
+ consumer.start();
}
@Override
public void shutdown() {
-
+ consumer.shutdown();
}
@Override
public void init(Properties keyValue) throws Exception {
-
+ consumer = new StandaloneConsumer(keyValue);
}
@Override
public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context) {
-
+ consumer.updateOffset(cloudEvents, context);
}
@Override
public void subscribe(String topic, EventListener listener) throws Exception {
-
+ consumer.subscribe(topic, listener);
}
@Override
public void unsubscribe(String topic) {
-
+ consumer.unsubscribe(topic);
}
// @Override
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java
index a01bb25..bc39352 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java
@@ -63,7 +63,6 @@ public class AsyncPublish {
Thread.sleep(1000);
}
- client.listen();
Thread.sleep(2000);
} catch (Exception e) {
logger.warn("AsyncPublish failed", e);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java
index f27dbd1..9df4548 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.client.tcp.EventMeshTCPClient;
import org.apache.eventmesh.client.tcp.common.ReceiveMsgHook;
import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
import org.apache.eventmesh.client.tcp.impl.EventMeshTCPClientFactory;
+import org.apache.eventmesh.client.tcp.impl.eventmeshmessage.EventMeshMessageTCPClient;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
@@ -39,6 +40,8 @@ public class AsyncSubscribe implements ReceiveMsgHook<EventMeshMessage> {
public static AsyncSubscribe handler = new AsyncSubscribe();
+ private static EventMeshTCPClient client;
+
public static void main(String[] agrs) throws Exception {
Properties properties = Utils.readPropertiesFile("application.properties");
final String eventMeshIp = properties.getProperty("eventmesh.ip");
@@ -49,8 +52,8 @@ public class AsyncSubscribe implements ReceiveMsgHook<EventMeshMessage> {
.port(eventMeshTcpPort)
.userAgent(userAgent)
.build();
- try (EventMeshTCPClient<EventMeshMessage> client = EventMeshTCPClientFactory.createEventMeshTCPClient(
- eventMeshTcpClientConfig, EventMeshMessage.class)) {
+ try {
+ client = EventMeshTCPClientFactory.createEventMeshTCPClient(eventMeshTcpClientConfig, EventMeshMessage.class);
client.init();
client.heartbeat();
@@ -59,6 +62,11 @@ public class AsyncSubscribe implements ReceiveMsgHook<EventMeshMessage> {
client.listen();
+ //client.unsubscribe();
+
+ // release resource and close client
+ // client.close();
+
} catch (Exception e) {
log.warn("AsyncSubscribe failed", e);
}
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/resources/META-INF.eventmesh/org.apache.eventmesh.protocol.api.ProtocolAdaptor b/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.protocol.api.ProtocolAdaptor
similarity index 100%
rename from eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/resources/META-INF.eventmesh/org.apache.eventmesh.protocol.api.ProtocolAdaptor
rename to eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.protocol.api.ProtocolAdaptor
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/cloudevent/CloudEventTCPPubClient.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/cloudevent/CloudEventTCPPubClient.java
index d4fbec7..ca5901a 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/cloudevent/CloudEventTCPPubClient.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/cloudevent/CloudEventTCPPubClient.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.client.tcp.common.ReceiveMsgHook;
import org.apache.eventmesh.client.tcp.common.RequestContext;
import org.apache.eventmesh.client.tcp.common.TcpClient;
import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.Package;
@@ -128,7 +129,7 @@ class CloudEventTCPPubClient extends TcpClient implements EventMeshTCPPubClient<
Package msg = MessageUtils.buildPackage(cloudEvent, Command.ASYNC_MESSAGE_TO_SERVER);
log.info("SimplePubClientImpl cloud event|{}|publish|send|type={}|protocol={}|msg={}",
clientNo, msg.getHeader().getCommand(),
- msg.getHeader().getProperty(PropertyConst.PROPERTY_MESSAGE_PROTOCOL), msg);
+ msg.getHeader().getProperty(Constants.PROTOCOL_TYPE), msg);
return io(msg, timeout);
} catch (Exception ex) {
throw new EventMeshException("publish error", ex);
@@ -141,7 +142,7 @@ class CloudEventTCPPubClient extends TcpClient implements EventMeshTCPPubClient<
// todo: transform EventMeshMessage to Package
Package msg = MessageUtils.buildPackage(cloudEvent, Command.BROADCAST_MESSAGE_TO_SERVER);
log.info("{}|publish|send|type={}|protocol={}|msg={}", clientNo, msg.getHeader().getCommand(),
- msg.getHeader().getProperty(PropertyConst.PROPERTY_MESSAGE_PROTOCOL), msg);
+ msg.getHeader().getProperty(Constants.PROTOCOL_TYPE), msg);
super.send(msg);
} catch (Exception ex) {
throw new EventMeshException("Broadcast message error", ex);
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java
index a3dd6a4..18f555f 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.client.tcp.common.ReceiveMsgHook;
import org.apache.eventmesh.client.tcp.common.RequestContext;
import org.apache.eventmesh.client.tcp.common.TcpClient;
import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
@@ -72,8 +73,8 @@ class EventMeshMessageTCPPubClient extends TcpClient implements EventMeshTCPPubC
@Override
public void heartbeat() throws EventMeshException {
- if (task != null) {
- synchronized (EventMeshMessageTCPPubClient.class) {
+// if (task != null) {
+// synchronized (EventMeshMessageTCPPubClient.class) {
task = scheduler.scheduleAtFixedRate(() -> {
try {
if (!isActive()) {
@@ -86,8 +87,8 @@ class EventMeshMessageTCPPubClient extends TcpClient implements EventMeshTCPPubC
}
}, EventMeshCommon.HEARTBEAT, EventMeshCommon.HEARTBEAT, TimeUnit.MILLISECONDS);
}
- }
- }
+// }
+// }
@Override
public void reconnect() throws EventMeshException {
@@ -128,9 +129,9 @@ class EventMeshMessageTCPPubClient extends TcpClient implements EventMeshTCPPubC
try {
// todo: transform EventMeshMessage to Package
Package msg = MessageUtils.buildPackage(eventMeshMessage, Command.ASYNC_MESSAGE_TO_SERVER);
- log.info("SimplePubClientImpl cloud event|{}|publish|send|type={}|protocol={}|msg={}",
+ log.info("SimplePubClientImpl em message|{}|publish|send|type={}|protocol={}|msg={}",
clientNo, msg.getHeader().getCommand(),
- msg.getHeader().getProperty(PropertyConst.PROPERTY_MESSAGE_PROTOCOL), msg);
+ msg.getHeader().getProperty(Constants.PROTOCOL_TYPE), msg);
return io(msg, timeout);
} catch (Exception ex) {
throw new EventMeshException("publish error", ex);
@@ -143,7 +144,7 @@ class EventMeshMessageTCPPubClient extends TcpClient implements EventMeshTCPPubC
// todo: transform EventMeshMessage to Package
Package msg = MessageUtils.buildPackage(eventMeshMessage, Command.BROADCAST_MESSAGE_TO_SERVER);
log.info("{}|publish|send|type={}|protocol={}|msg={}", clientNo, msg.getHeader().getCommand(),
- msg.getHeader().getProperty(PropertyConst.PROPERTY_MESSAGE_PROTOCOL), msg);
+ msg.getHeader().getProperty(Constants.PROTOCOL_TYPE), msg);
super.send(msg);
} catch (Exception ex) {
throw new EventMeshException("Broadcast message error", ex);
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPSubClient.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPSubClient.java
index add12a5..f9da554 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPSubClient.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPSubClient.java
@@ -71,8 +71,8 @@ class EventMeshMessageTCPSubClient extends TcpClient implements EventMeshTCPSubC
@Override
public void heartbeat() throws EventMeshException {
- if (task == null) {
- synchronized (EventMeshMessageTCPSubClient.class) {
+// if (task == null) {
+// synchronized (EventMeshMessageTCPSubClient.class) {
task = scheduler.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
@@ -88,8 +88,8 @@ class EventMeshMessageTCPSubClient extends TcpClient implements EventMeshTCPSubC
}
}, EventMeshCommon.HEARTBEAT, EventMeshCommon.HEARTBEAT, TimeUnit.MILLISECONDS);
}
- }
- }
+// }
+// }
@Override
public void reconnect() throws EventMeshException {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org