This is an automated email from the ASF dual-hosted git repository.
mikexue 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 733d6ef5c [ISSUE #3849]GrpcMessageProtocolResolver has many unused
variables and repetitive method def. (#3850)
733d6ef5c is described below
commit 733d6ef5cc0002f8d278663918ab0bfa2dc4c261
Author: pandaapo <[email protected]>
AuthorDate: Sat May 6 16:47:43 2023 +0800
[ISSUE #3849]GrpcMessageProtocolResolver has many unused variables and
repetitive method def. (#3850)
* many unused variables and repetitive method def.
* modify wrong api usage.
* Revoke controversial import static.
---
.../resolver/grpc/GrpcMessageProtocolResolver.java | 163 ++++++++-------------
1 file changed, 58 insertions(+), 105 deletions(-)
diff --git
a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/grpc/GrpcMessageProtocolResolver.java
b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/grpc/GrpcMessageProtocolResolver.java
index 044364774..be1d741a3 100644
---
a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/grpc/GrpcMessageProtocolResolver.java
+++
b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/grpc/GrpcMessageProtocolResolver.java
@@ -26,13 +26,11 @@ import
org.apache.eventmesh.common.protocol.grpc.protos.SimpleMessage;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import io.cloudevents.CloudEvent;
-import io.cloudevents.SpecVersion;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.format.EventFormat;
import io.cloudevents.core.provider.EventFormatProvider;
@@ -44,22 +42,18 @@ public class GrpcMessageProtocolResolver {
String contentType =
message.getPropertiesOrDefault(ProtocolKey.CONTENT_TYPE,
Constants.CONTENT_TYPE_CLOUDEVENTS_JSON);
EventFormat eventFormat =
EventFormatProvider.getInstance().resolveFormat(contentType);
- CloudEvent event =
Objects.requireNonNull(eventFormat).deserialize(cloudEventJson.getBytes(StandardCharsets.UTF_8));
+ CloudEvent event =
Objects.requireNonNull(eventFormat).deserialize(cloudEventJson.getBytes(Constants.DEFAULT_CHARSET));
RequestHeader header = message.getHeader();
- String seqNum = getMessageItemValue(message.getSeqNum(), event,
ProtocolKey.SEQ_NUM);
- String uniqueId = getMessageItemValue(message.getUniqueId(), event,
ProtocolKey.UNIQUE_ID);
- String ttl = getMessageItemValue(message.getTtl(), event,
ProtocolKey.TTL);
- String producerGroup = getMessageItemValue(message.getProducerGroup(),
event, ProtocolKey.PRODUCERGROUP);
+ String seqNum = getEventExtensionIfAbsent(message.getSeqNum(), event,
ProtocolKey.SEQ_NUM);
+ String uniqueId = getEventExtensionIfAbsent(message.getUniqueId(),
event, ProtocolKey.UNIQUE_ID);
+ String ttl = getEventExtensionIfAbsent(message.getTtl(), event,
ProtocolKey.TTL);
+ String producerGroup =
getEventExtensionIfAbsent(message.getProducerGroup(), event,
ProtocolKey.PRODUCERGROUP);
String topic = StringUtils.defaultIfEmpty(message.getTopic(),
event.getSubject());
- CloudEventBuilder eventBuilder = builderCloudEventBuilder(header,
event);
- eventBuilder.withSubject(topic)
- .withExtension(ProtocolKey.SEQ_NUM, seqNum)
- .withExtension(ProtocolKey.UNIQUE_ID, uniqueId)
- .withExtension(ProtocolKey.PRODUCERGROUP, producerGroup)
- .withExtension(ProtocolKey.TTL, ttl);
+ CloudEventBuilder eventBuilder = builderCloudEventBuilder(header,
event, seqNum, uniqueId, producerGroup, ttl,
+ topic);
message.getPropertiesMap().forEach(eventBuilder::withExtension);
@@ -67,25 +61,21 @@ public class GrpcMessageProtocolResolver {
}
public static SimpleMessageWrapper buildSimpleMessage(CloudEvent
cloudEvent) {
-
- String env = getCloudEventExtension(cloudEvent, ProtocolKey.ENV,
"env");
- String idc = getCloudEventExtension(cloudEvent, ProtocolKey.IDC,
"idc");
-
- String ip = getCloudEventExtension(cloudEvent, ProtocolKey.IP,
"127.0.0.1");
- String pid = getCloudEventExtension(cloudEvent, ProtocolKey.PID,
"123");
- String sys = getCloudEventExtension(cloudEvent, ProtocolKey.SYS,
"sys123");
- String userName = getCloudEventExtension(cloudEvent,
ProtocolKey.USERNAME, "user");
- String passwd = getCloudEventExtension(cloudEvent, ProtocolKey.PASSWD,
"pass");
- String language = getCloudEventExtension(cloudEvent,
ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA);
-
- String protocol = getCloudEventExtension(cloudEvent,
ProtocolKey.PROTOCOL_TYPE, "protocol");
- String protocolDesc = getCloudEventExtension(cloudEvent,
ProtocolKey.PROTOCOL_DESC, "protocolDesc");
- String protocolVersion = getCloudEventExtension(cloudEvent,
ProtocolKey.PROTOCOL_VERSION, "1.0");
- String seqNum = getCloudEventExtension(cloudEvent,
ProtocolKey.SEQ_NUM, "");
- String uniqueId = getCloudEventExtension(cloudEvent,
ProtocolKey.UNIQUE_ID, "");
-
- String producerGroup = getCloudEventExtension(cloudEvent,
ProtocolKey.PRODUCERGROUP, "producerGroup");
- String ttl = getCloudEventExtension(cloudEvent, ProtocolKey.TTL,
"3000");
+ String env = getEventExtensionOrDefault(cloudEvent, ProtocolKey.ENV,
"env");
+ String idc = getEventExtensionOrDefault(cloudEvent, ProtocolKey.IDC,
"idc");
+ String ip = getEventExtensionOrDefault(cloudEvent, ProtocolKey.IP,
"127.0.0.1");
+ String pid = getEventExtensionOrDefault(cloudEvent, ProtocolKey.PID,
"123");
+ String sys = getEventExtensionOrDefault(cloudEvent, ProtocolKey.SYS,
"sys123");
+ String userName = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.USERNAME, "user");
+ String passwd = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.PASSWD, "pass");
+ String language = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA);
+ String protocol = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.PROTOCOL_TYPE, "protocol");
+ String protocolDesc = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.PROTOCOL_DESC, "protocolDesc");
+ String protocolVersion = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.PROTOCOL_VERSION, "1.0");
+ String seqNum = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.SEQ_NUM, "");
+ String uniqueId = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.UNIQUE_ID, "");
+ String producerGroup = getEventExtensionOrDefault(cloudEvent,
ProtocolKey.PRODUCERGROUP, "producerGroup");
+ String ttl = getEventExtensionOrDefault(cloudEvent, ProtocolKey.TTL,
"3000");
RequestHeader header = RequestHeader.newBuilder()
.setEnv(env).setIdc(idc)
@@ -100,7 +90,7 @@ public class GrpcMessageProtocolResolver {
SimpleMessage.Builder messageBuilder = SimpleMessage.newBuilder()
.setHeader(header)
- .setContent(new
String(Objects.requireNonNull(eventFormat).serialize(cloudEvent),
StandardCharsets.UTF_8))
+ .setContent(new
String(Objects.requireNonNull(eventFormat).serialize(cloudEvent),
Constants.DEFAULT_CHARSET))
.setProducerGroup(producerGroup)
.setSeqNum(seqNum)
.setUniqueId(uniqueId)
@@ -115,11 +105,6 @@ public class GrpcMessageProtocolResolver {
return new SimpleMessageWrapper(messageBuilder.build());
}
- private static String getCloudEventExtension(CloudEvent cloudEvent, String
protocolKey, String defaultValue) {
- Object extension = cloudEvent.getExtension(protocolKey);
- return Objects.isNull(extension) ? defaultValue : extension.toString();
- }
-
public static List<CloudEvent> buildBatchEvents(BatchMessage batchMessage)
{
List<CloudEvent> cloudEvents = new ArrayList<>();
@@ -130,46 +115,16 @@ public class GrpcMessageProtocolResolver {
String contentType =
item.getPropertiesOrDefault(ProtocolKey.CONTENT_TYPE,
Constants.CONTENT_TYPE_CLOUDEVENTS_JSON);
EventFormat eventFormat =
EventFormatProvider.getInstance().resolveFormat(contentType);
- CloudEvent event =
Objects.requireNonNull(eventFormat).deserialize(cloudEventJson.getBytes(StandardCharsets.UTF_8));
-
- String env = StringUtils.isEmpty(header.getEnv()) ?
getEventExtension(event, ProtocolKey.ENV) : header.getEnv();
- String idc = StringUtils.isEmpty(header.getIdc()) ?
getEventExtension(event, ProtocolKey.IDC) : header.getIdc();
- String ip = StringUtils.isEmpty(header.getIp()) ?
getEventExtension(event, ProtocolKey.IP) : header.getIp();
- String pid = StringUtils.isEmpty(header.getPid()) ?
getEventExtension(event, ProtocolKey.PID) : header.getPid();
- String sys = StringUtils.isEmpty(header.getSys()) ?
getEventExtension(event, ProtocolKey.SYS) : header.getSys();
-
- String language = StringUtils.isEmpty(header.getLanguage())
- ? getEventExtension(event, ProtocolKey.LANGUAGE) :
header.getLanguage();
-
- String protocolType = StringUtils.isEmpty(header.getProtocolType())
- ? getEventExtension(event, ProtocolKey.PROTOCOL_TYPE) :
header.getProtocolType();
-
- String protocolDesc = StringUtils.isEmpty(header.getProtocolDesc())
- ? getEventExtension(event, ProtocolKey.PROTOCOL_DESC) :
header.getProtocolDesc();
-
- String protocolVersion =
StringUtils.isEmpty(header.getProtocolVersion())
- ? getEventExtension(event, ProtocolKey.PROTOCOL_VERSION) :
header.getProtocolVersion();
-
- String username = StringUtils.isEmpty(header.getUsername()) ?
getEventExtension(event, ProtocolKey.USERNAME) : header.getUsername();
- String passwd = StringUtils.isEmpty(header.getPassword()) ?
getEventExtension(event, ProtocolKey.PASSWD) : header.getPassword();
-
- String seqNum = StringUtils.isEmpty(item.getSeqNum()) ?
getEventExtension(event, ProtocolKey.SEQ_NUM) : item.getSeqNum();
- String uniqueId = StringUtils.isEmpty(item.getUniqueId()) ?
getEventExtension(event, ProtocolKey.UNIQUE_ID) : item.getUniqueId();
+ CloudEvent event =
Objects.requireNonNull(eventFormat).deserialize(cloudEventJson.getBytes(Constants.DEFAULT_CHARSET));
- String topic = StringUtils.isEmpty(batchMessage.getTopic()) ?
event.getSubject() : batchMessage.getTopic();
+ String seqNum = getEventExtensionIfAbsent(item.getSeqNum(), event,
ProtocolKey.SEQ_NUM);
+ String uniqueId = getEventExtensionIfAbsent(item.getUniqueId(),
event, ProtocolKey.UNIQUE_ID);
+ String producerGroup =
getEventExtensionIfAbsent(batchMessage.getProducerGroup(), event,
ProtocolKey.PRODUCERGROUP);
+ String ttl = getEventExtensionIfAbsent(item.getTtl(), event,
ProtocolKey.TTL);
+ String topic = StringUtils.defaultIfEmpty(batchMessage.getTopic(),
event.getSubject());
- String producerGroup =
StringUtils.isEmpty(batchMessage.getProducerGroup())
- ? getEventExtension(event, ProtocolKey.PRODUCERGROUP) :
batchMessage.getProducerGroup();
-
- String ttl = StringUtils.isEmpty(item.getTtl()) ?
getEventExtension(event, ProtocolKey.TTL) : item.getTtl();
-
- CloudEventBuilder eventBuilder = builderCloudEventBuilder(header,
event);
-
- eventBuilder.withSubject(topic)
- .withExtension(ProtocolKey.SEQ_NUM, seqNum)
- .withExtension(ProtocolKey.UNIQUE_ID, uniqueId)
- .withExtension(ProtocolKey.PRODUCERGROUP, producerGroup)
- .withExtension(ProtocolKey.TTL, ttl);
+ CloudEventBuilder eventBuilder = builderCloudEventBuilder(header,
event, seqNum, uniqueId, producerGroup,
+ ttl, topic);
item.getPropertiesMap().forEach(eventBuilder::withExtension);
@@ -178,29 +133,22 @@ public class GrpcMessageProtocolResolver {
return cloudEvents;
}
- private static String getHeaderValue(String value, CloudEvent event,
String key) {
- return StringUtils.isEmpty(value) ?
Objects.requireNonNull(event.getExtension(key)).toString() : value;
- }
-
- private static String getMessageItemValue(String value, CloudEvent event,
String key) {
- return StringUtils.isEmpty(value) ?
Objects.requireNonNull(event.getExtension(key)).toString() : value;
- }
-
- private static CloudEventBuilder builderCloudEventBuilder(RequestHeader
header, CloudEvent event) {
- String env = getHeaderValue(header.getEnv(), event, ProtocolKey.ENV);
- String idc = getHeaderValue(header.getIdc(), event, ProtocolKey.IDC);
- String ip = getHeaderValue(header.getIp(), event, ProtocolKey.IP);
- String pid = getHeaderValue(header.getPid(), event, ProtocolKey.PID);
- String sys = getHeaderValue(header.getSys(), event, ProtocolKey.SYS);
- String language = getHeaderValue(header.getLanguage(), event,
ProtocolKey.LANGUAGE);
- String protocolType = getHeaderValue(header.getProtocolType(), event,
ProtocolKey.PROTOCOL_TYPE);
- String protocolDesc = getHeaderValue(header.getProtocolDesc(), event,
ProtocolKey.PROTOCOL_DESC);
- String protocolVersion = getHeaderValue(header.getProtocolVersion(),
event, ProtocolKey.PROTOCOL_VERSION);
- String username = getHeaderValue(header.getUsername(), event,
ProtocolKey.USERNAME);
- String passwd = getHeaderValue(header.getPassword(), event,
ProtocolKey.PASSWD);
-
- CloudEventBuilder eventBuilder =
StringUtils.equals(SpecVersion.V1.toString(), protocolVersion)
- ? CloudEventBuilder.v1(event) : CloudEventBuilder.v03(event);
+ private static CloudEventBuilder builderCloudEventBuilder(RequestHeader
header, CloudEvent event, String seqNum,
+ String uniqueId,
String producerGroup, String ttl,
+ String topic) {
+ String env = getEventExtensionIfAbsent(header.getEnv(), event,
ProtocolKey.ENV);
+ String idc = getEventExtensionIfAbsent(header.getIdc(), event,
ProtocolKey.IDC);
+ String ip = getEventExtensionIfAbsent(header.getIp(), event,
ProtocolKey.IP);
+ String pid = getEventExtensionIfAbsent(header.getPid(), event,
ProtocolKey.PID);
+ String sys = getEventExtensionIfAbsent(header.getSys(), event,
ProtocolKey.SYS);
+ String language = getEventExtensionIfAbsent(header.getLanguage(),
event, ProtocolKey.LANGUAGE);
+ String protocolType =
getEventExtensionIfAbsent(header.getProtocolType(), event,
ProtocolKey.PROTOCOL_TYPE);
+ String protocolDesc =
getEventExtensionIfAbsent(header.getProtocolDesc(), event,
ProtocolKey.PROTOCOL_DESC);
+ String protocolVersion =
getEventExtensionIfAbsent(header.getProtocolVersion(), event,
ProtocolKey.PROTOCOL_VERSION);
+ String username = getEventExtensionIfAbsent(header.getUsername(),
event, ProtocolKey.USERNAME);
+ String passwd = getEventExtensionIfAbsent(header.getPassword(), event,
ProtocolKey.PASSWD);
+
+ CloudEventBuilder eventBuilder = CloudEventBuilder.from(event);
return eventBuilder
.withExtension(ProtocolKey.ENV, env)
@@ -213,17 +161,22 @@ public class GrpcMessageProtocolResolver {
.withExtension(ProtocolKey.LANGUAGE, language)
.withExtension(ProtocolKey.PROTOCOL_TYPE, protocolType)
.withExtension(ProtocolKey.PROTOCOL_DESC, protocolDesc)
- .withExtension(ProtocolKey.PROTOCOL_VERSION, protocolVersion);
-
- }
+ .withExtension(ProtocolKey.PROTOCOL_VERSION, protocolVersion)
+ .withExtension(ProtocolKey.SEQ_NUM, seqNum)
+ .withExtension(ProtocolKey.UNIQUE_ID, uniqueId)
+ .withExtension(ProtocolKey.PRODUCERGROUP, producerGroup)
+ .withExtension(ProtocolKey.TTL, ttl)
+ .withSubject(topic);
- private static String getEventExtension(CloudEvent event, String
protocolKey) {
- return
Objects.requireNonNull(event.getExtension(protocolKey)).toString();
}
- private static String getEventExtension(CloudEvent event, String
protocolKey, String defaultValue) {
+ private static String getEventExtensionOrDefault(CloudEvent event, String
protocolKey, String defaultValue) {
Object extension = event.getExtension(protocolKey);
return Objects.isNull(extension) ? defaultValue : extension.toString();
}
+ private static String getEventExtensionIfAbsent(String value, CloudEvent
event, String key) {
+ return StringUtils.isEmpty(value) ?
Objects.requireNonNull(event.getExtension(key)).toString() : value;
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
