This is an automated email from the ASF dual-hosted git repository.
mxsm 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 5b8616bf3 [ISSUE #4261] Some code optimizations to the
EventMeshCloudEventBuilder . (#4285)
5b8616bf3 is described below
commit 5b8616bf308217cd362a279eb3986095940e49aa
Author: silentchildh <[email protected]>
AuthorDate: Fri Jul 28 09:05:38 2023 +0800
[ISSUE #4261] Some code optimizations to the EventMeshCloudEventBuilder .
(#4285)
* fix: Some code optimizations to the EventMeshCloudEventBuilder
* [ISSUE #4261]: Fix get the corresponding IDC, not the ENV
* refactor: rename private method 'builderCloudEvent' to 'buildCloudEvent'
* refactor: rename private method
* style: to fit `CheckStyle` plugin
---
.../grpc/util/EventMeshCloudEventBuilder.java | 156 +++++++++------------
1 file changed, 65 insertions(+), 91 deletions(-)
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
index feaef016b..e16afe39e 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
@@ -68,7 +68,7 @@ public class EventMeshCloudEventBuilder {
private static final EventFormat eventProtoFormat =
EventFormatProvider.getInstance().resolveFormat(ProtobufFormat.PROTO_CONTENT_TYPE);
public static Map<String, CloudEventAttributeValue>
buildCommonCloudEventAttributes(EventMeshGrpcClientConfig clientConfig,
- EventMeshProtocolType protocolType) {
+
EventMeshProtocolType protocolType) {
final Map<String, CloudEventAttributeValue> attributeValueMap = new
HashMap<>(64);
attributeValueMap.put(ProtocolKey.ENV,
CloudEventAttributeValue.newBuilder().setCeString(clientConfig.getEnv()).build());
attributeValueMap.put(ProtocolKey.IDC,
CloudEventAttributeValue.newBuilder().setCeString(clientConfig.getIdc()).build());
@@ -85,7 +85,7 @@ public class EventMeshCloudEventBuilder {
}
public static CloudEvent buildEventSubscription(EventMeshGrpcClientConfig
clientConfig, EventMeshProtocolType protocolType, String url,
- List<SubscriptionItem> subscriptionItems) {
+ List<SubscriptionItem>
subscriptionItems) {
if (CollectionUtils.isEmpty(subscriptionItems)) {
return null;
@@ -100,13 +100,13 @@ public class EventMeshCloudEventBuilder {
attributeValueMap.put(ProtocolKey.URL,
CloudEventAttributeValue.newBuilder().setCeString(url).build());
}
return CloudEvent.newBuilder()
- .setId(RandomStringUtils.generateUUID())
- .setSource(URI.create("/").toString())
- .setSpecVersion(SpecVersion.V1.toString())
- .setType(CLOUD_EVENT_TYPE)
- .setTextData(JsonUtils.toJSONString(subscriptionItemSet))
- .putAllAttributes(attributeValueMap)
- .build();
+ .setId(RandomStringUtils.generateUUID())
+ .setSource(URI.create("/").toString())
+ .setSpecVersion(SpecVersion.V1.toString())
+ .setType(CLOUD_EVENT_TYPE)
+ .setTextData(JsonUtils.toJSONString(subscriptionItemSet))
+ .putAllAttributes(attributeValueMap)
+ .build();
}
/**
@@ -118,7 +118,7 @@ public class EventMeshCloudEventBuilder {
* @see <a
href="https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/spec.md#context-attributes";>context-attributes</a>
*/
public static <T> CloudEvent buildEventMeshCloudEvent(final T message,
final EventMeshGrpcClientConfig clientConfig,
- final EventMeshProtocolType protocolType) {
+ final
EventMeshProtocolType protocolType) {
switch (protocolType) {
case CLOUD_EVENTS: {
@@ -141,35 +141,32 @@ public class EventMeshCloudEventBuilder {
}
private static CloudEvent
switchEventMeshMessage2EventMeshCloudEvent(EventMeshMessage message,
EventMeshGrpcClientConfig clientConfig,
- EventMeshProtocolType protocolType) {
- final EventMeshMessage eventMeshMessage = message;
- final String ttl =
eventMeshMessage.getProp(Constants.EVENTMESH_MESSAGE_CONST_TTL) == null
- ? Constants.DEFAULT_EVENTMESH_MESSAGE_TTL :
eventMeshMessage.getProp(Constants.EVENTMESH_MESSAGE_CONST_TTL);
- final Map<String, String> props = eventMeshMessage.getProp() == null ?
new HashMap<>() : eventMeshMessage.getProp();
- final String seqNum = eventMeshMessage.getBizSeqNo() == null ?
RandomStringUtils.generateNum(30) : eventMeshMessage.getBizSeqNo();
- final String uniqueId = eventMeshMessage.getUniqueId() == null ?
RandomStringUtils.generateNum(30) : eventMeshMessage.getUniqueId();
- final String dataContentType =
props.computeIfAbsent(ProtocolKey.DATA_CONTENT_TYPE, (key) -> "text/plain");
+
EventMeshProtocolType protocolType) {
+ final String ttl =
message.getProp(Constants.EVENTMESH_MESSAGE_CONST_TTL) == null
+ ? Constants.DEFAULT_EVENTMESH_MESSAGE_TTL :
message.getProp(Constants.EVENTMESH_MESSAGE_CONST_TTL);
+ final Map<String, String> props = message.getProp() == null ? new
HashMap<>() : message.getProp();
+ final String seqNum = message.getBizSeqNo() == null ?
RandomStringUtils.generateNum(30) : message.getBizSeqNo();
+ final String uniqueId = message.getUniqueId() == null ?
RandomStringUtils.generateNum(30) : message.getUniqueId();
+ final String dataContentType =
props.computeIfAbsent(ProtocolKey.DATA_CONTENT_TYPE, key -> "text/plain");
final Map<String, CloudEventAttributeValue> attributeValueMap =
buildCommonCloudEventAttributes(clientConfig, protocolType);
attributeValueMap.put(ProtocolKey.TTL,
CloudEventAttributeValue.newBuilder().setCeString(ttl).build());
attributeValueMap.put(ProtocolKey.SEQ_NUM,
CloudEventAttributeValue.newBuilder().setCeString(seqNum).build());
attributeValueMap.put(ProtocolKey.UNIQUE_ID,
CloudEventAttributeValue.newBuilder().setCeString(uniqueId).build());
attributeValueMap.put(ProtocolKey.PRODUCERGROUP,
-
CloudEventAttributeValue.newBuilder().setCeString(clientConfig.getProducerGroup()).build());
- if (null != eventMeshMessage.getTopic()) {
- attributeValueMap.put(ProtocolKey.SUBJECT,
CloudEventAttributeValue.newBuilder().setCeString(eventMeshMessage.getTopic()).build());
+
CloudEventAttributeValue.newBuilder().setCeString(clientConfig.getProducerGroup()).build());
+ if (null != message.getTopic()) {
+ attributeValueMap.put(ProtocolKey.SUBJECT,
CloudEventAttributeValue.newBuilder().setCeString(message.getTopic()).build());
}
attributeValueMap.put(ProtocolKey.DATA_CONTENT_TYPE,
CloudEventAttributeValue.newBuilder().setCeString("text/plain").build());
- props.entrySet()
- .forEach(
- entry -> attributeValueMap.put(entry.getKey(),
CloudEventAttributeValue.newBuilder().setCeString(entry.getValue()).build()));
+ props.forEach((key, value) -> attributeValueMap.put(key,
CloudEventAttributeValue.newBuilder().setCeString(value).build()));
CloudEvent.Builder builder = CloudEvent.newBuilder()
- .setId(RandomStringUtils.generateUUID())
- .setSource(URI.create("/").toString())
- .setSpecVersion(SpecVersion.V1.toString())
- .setType(CLOUD_EVENT_TYPE)
- .putAllAttributes(attributeValueMap);
- final String content = eventMeshMessage.getContent();
+ .setId(RandomStringUtils.generateUUID())
+ .setSource(URI.create("/").toString())
+ .setSpecVersion(SpecVersion.V1.toString())
+ .setType(CLOUD_EVENT_TYPE)
+ .putAllAttributes(attributeValueMap);
+ final String content = message.getContent();
if (StringUtils.isNotEmpty(content)) {
if (ProtoSupport.isTextContent(dataContentType)) {
builder.setTextData(content);
@@ -189,57 +186,26 @@ public class EventMeshCloudEventBuilder {
}
private static CloudEvent
switchCloudEvent2EventMeshCloudEvent(io.cloudevents.CloudEvent message,
EventMeshGrpcClientConfig clientConfig,
- EventMeshProtocolType protocolType) {
- final io.cloudevents.CloudEvent cloudEvent = message;
- CloudEventBuilder cloudEventBuilder =
CloudEventBuilder.from(cloudEvent);
- if (null == cloudEvent.getExtension(ProtocolKey.ENV)) {
- cloudEventBuilder.withExtension(ProtocolKey.ENV,
clientConfig.getEnv());
- }
- if (null == cloudEvent.getExtension(ProtocolKey.IDC)) {
- cloudEventBuilder.withExtension(ProtocolKey.IDC,
clientConfig.getEnv());
- }
- if (null == cloudEvent.getExtension(ProtocolKey.IP)) {
- cloudEventBuilder.withExtension(ProtocolKey.IP,
Objects.requireNonNull(IPUtils.getLocalAddress()));
- }
- if (null == cloudEvent.getExtension(ProtocolKey.PID)) {
- cloudEventBuilder.withExtension(ProtocolKey.PID,
Long.toString(ThreadUtils.getPID()));
- }
- if (null == cloudEvent.getExtension(ProtocolKey.SYS)) {
- cloudEventBuilder.withExtension(ProtocolKey.SYS,
clientConfig.getSys());
- }
+
EventMeshProtocolType protocolType) {
+
+ CloudEventBuilder cloudEventBuilder = CloudEventBuilder.from(message);
+
+ buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.ENV,
clientConfig.getEnv());
+ buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.IDC,
clientConfig.getIdc());
+ buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.IP,
Objects.requireNonNull(IPUtils.getLocalAddress()));
+ buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.PID,
Long.toString(ThreadUtils.getPID()));
+ buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.SYS,
clientConfig.getSys());
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA);
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.PROTOCOL_TYPE, protocolType.protocolTypeName());
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.PROTOCOL_DESC, "grpc-cloud-event");
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.PROTOCOL_VERSION, message.getSpecVersion().toString());
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.UNIQUE_ID, RandomStringUtils.generateNum(30));
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.SEQ_NUM, RandomStringUtils.generateNum(30));
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.USERNAME, clientConfig.getUserName());
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.PASSWD, clientConfig.getPassword());
+ buildCloudEventIfAbsent(message, cloudEventBuilder,
ProtocolKey.PRODUCERGROUP, clientConfig.getProducerGroup());
+ buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.TTL,
Constants.DEFAULT_EVENTMESH_MESSAGE_TTL);
- if (null == cloudEvent.getExtension(ProtocolKey.LANGUAGE)) {
- cloudEventBuilder.withExtension(ProtocolKey.LANGUAGE,
Constants.LANGUAGE_JAVA);
- }
-
- if (null == cloudEvent.getExtension(ProtocolKey.PROTOCOL_TYPE)) {
- cloudEventBuilder.withExtension(ProtocolKey.PROTOCOL_TYPE,
protocolType.protocolTypeName());
- }
- if (null == cloudEvent.getExtension(ProtocolKey.PROTOCOL_DESC)) {
- cloudEventBuilder.withExtension(ProtocolKey.PROTOCOL_DESC,
"grpc-cloud-event");
- }
- if (null == cloudEvent.getExtension(ProtocolKey.PROTOCOL_VERSION)) {
- cloudEventBuilder.withExtension(ProtocolKey.PROTOCOL_VERSION,
cloudEvent.getSpecVersion().toString());
- }
- if (null == cloudEvent.getExtension(ProtocolKey.UNIQUE_ID)) {
- cloudEventBuilder.withExtension(ProtocolKey.UNIQUE_ID,
RandomStringUtils.generateNum(30));
- }
- if (null == cloudEvent.getExtension(ProtocolKey.SEQ_NUM)) {
- cloudEventBuilder.withExtension(ProtocolKey.SEQ_NUM,
RandomStringUtils.generateNum(30));
- }
- if (null == cloudEvent.getExtension(ProtocolKey.USERNAME)) {
- cloudEventBuilder.withExtension(ProtocolKey.USERNAME,
clientConfig.getUserName());
- }
- if (null == cloudEvent.getExtension(ProtocolKey.PASSWD)) {
- cloudEventBuilder.withExtension(ProtocolKey.PASSWD,
clientConfig.getPassword());
- }
- if (null == cloudEvent.getExtension(ProtocolKey.PRODUCERGROUP)) {
- cloudEventBuilder.withExtension(ProtocolKey.PRODUCERGROUP,
clientConfig.getProducerGroup());
- }
- if (null == cloudEvent.getExtension(ProtocolKey.TTL)) {
- final String ttl = Constants.DEFAULT_EVENTMESH_MESSAGE_TTL;
-
cloudEventBuilder.withExtension(Constants.EVENTMESH_MESSAGE_CONST_TTL, ttl);
- }
try {
return
CloudEvent.parseFrom(eventProtoFormat.serialize(cloudEventBuilder.build()));
} catch (InvalidProtocolBufferException exc) {
@@ -248,16 +214,24 @@ public class EventMeshCloudEventBuilder {
return null;
}
+ private static void buildCloudEventIfAbsent(io.cloudevents.CloudEvent
message, CloudEventBuilder cloudEventBuilder,
+ String extension, String
value) {
+ if (Objects.isNull(message.getExtension(extension))) {
+ cloudEventBuilder.withExtension(extension, value);
+ }
+ }
+
public static <T> CloudEventBatch buildEventMeshCloudEventBatch(final
List<T> messageList, final EventMeshGrpcClientConfig clientConfig,
- final EventMeshProtocolType protocolType) {
+ final
EventMeshProtocolType protocolType) {
if (CollectionUtils.isEmpty(messageList)) {
return null;
}
List<CloudEvent> cloudEventList = messageList.stream().map(item ->
buildEventMeshCloudEvent(item, clientConfig, protocolType))
- .collect(Collectors.toList());
+ .collect(Collectors.toList());
return
CloudEventBatch.newBuilder().addAllEvents(cloudEventList).build();
}
+ @SuppressWarnings("unchecked")
public static <T> T buildMessageFromEventMeshCloudEvent(final CloudEvent
cloudEvent, final EventMeshProtocolType protocolType) {
if (null == cloudEvent) {
@@ -271,9 +245,9 @@ public class EventMeshCloudEventBuilder {
if (StringUtils.isEmpty(seq) && StringUtils.isEmpty(uniqueId)) {
//The SubscriptionItem collection contains the content for the
subscription.
return (T) JsonUtils.parseTypeReferenceObject(content,
- new TypeReference<Set<HashMap<String, String>>>() {
+ new TypeReference<Set<HashMap<String, String>>>() {
- });
+ });
}
if (null == protocolType) {
return null;
@@ -299,12 +273,12 @@ public class EventMeshCloudEventBuilder {
Map<String, String> prop = new HashMap<>();
Objects.requireNonNull(cloudEvent).getAttributesMap().forEach((key,
value) -> prop.put(key, value.getCeString()));
return EventMeshMessage.builder()
- .content(cloudEvent.getTextData())
- .topic(EventMeshCloudEventUtils.getSubject(cloudEvent))
- .bizSeqNo(EventMeshCloudEventUtils.getSeqNum(cloudEvent))
- .uniqueId(EventMeshCloudEventUtils.getUniqueId(cloudEvent))
- .prop(prop)
- .build();
+ .content(cloudEvent.getTextData())
+ .topic(EventMeshCloudEventUtils.getSubject(cloudEvent))
+ .bizSeqNo(EventMeshCloudEventUtils.getSeqNum(cloudEvent))
+ .uniqueId(EventMeshCloudEventUtils.getUniqueId(cloudEvent))
+ .prop(prop)
+ .build();
}
-}
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
