Skip to content

Commit

Permalink
[ISSUE apache#3849]GrpcMessageProtocolResolver has many unused variab…
Browse files Browse the repository at this point in the history 
…les and repetitive method def. (apache#3850)

* many unused variables and repetitive method def.

* modify wrong api usage.

* Revoke controversial import static.
  • Loading branch information
@pandaapo
pandaapo authored May 6, 2023
1 parent d55288d commit 733d6ef
Showing 1 changed file with 58 additions and 105 deletions.
163 changes: 58 additions & 105 deletions .../org/apache/eventmesh/protocol/cloudevents/resolver/grpc/GrpcMessageProtocolResolver.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,13 +26,11 @@

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;
Expand All @@ -44,48 +42,40 @@ public static CloudEvent buildEvent(SimpleMessage message) {

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);

return eventBuilder.build();
}

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)
Expand All @@ -100,7 +90,7 @@ public static SimpleMessageWrapper buildSimpleMessage(CloudEvent cloudEvent) {

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)
Expand All @@ -115,11 +105,6 @@ public static SimpleMessageWrapper buildSimpleMessage(CloudEvent cloudEvent) {
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<>();

Expand All @@ -130,46 +115,16 @@ public static List<CloudEvent> buildBatchEvents(BatchMessage batchMessage) {

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);

Expand All @@ -178,29 +133,22 @@ public static List<CloudEvent> buildBatchEvents(BatchMessage batchMessage) {
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)
Expand All @@ -213,17 +161,22 @@ private static CloudEventBuilder builderCloudEventBuilder(RequestHeader header,
.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;
}

}

0 comments on commit 733d6ef

Please sign in to comment.