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 2022/11/29 02:15:33 UTC
[incubator-eventmesh] branch master updated: Refactoring code
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/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 74457c7aa Refactoring code
new 27c2d0a23 Merge pull request #2288 from weihubeats/refactor_code
74457c7aa is described below
commit 74457c7aaf51b9c188fa21878ba44e1497f8bb57
Author: weihu <we...@163.com>
AuthorDate: Mon Nov 28 16:22:06 2022 +0800
Refactoring code
---
.../client/http/AbstractProducerHttpClient.java | 97 ++++++++++++++++
.../client/http/producer/CloudEventProducer.java | 125 +++++++--------------
.../http/producer/EventMeshMessageProducer.java | 119 +++++++-------------
.../client/http/producer/OpenMessageProducer.java | 89 ++++-----------
4 files changed, 200 insertions(+), 230 deletions(-)
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/AbstractProducerHttpClient.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/AbstractProducerHttpClient.java
new file mode 100644
index 000000000..f7280844b
--- /dev/null
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/AbstractProducerHttpClient.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.client.http;
+
+import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
+import org.apache.eventmesh.client.http.model.RequestParam;
+import org.apache.eventmesh.client.http.producer.EventMeshProtocolProducer;
+import org.apache.eventmesh.client.http.producer.RRCallback;
+import org.apache.eventmesh.client.http.producer.RRCallbackResponseHandlerAdapter;
+import org.apache.eventmesh.client.http.util.HttpUtils;
+import org.apache.eventmesh.common.exception.EventMeshException;
+import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
+import org.apache.eventmesh.common.utils.JsonUtils;
+
+import java.io.IOException;
+
+/**
+ * AbstractProducerHttpClient
+ *
+ * @param <T>
+ */
+public abstract class AbstractProducerHttpClient<T> extends AbstractHttpClient implements EventMeshProtocolProducer<T> {
+
+ public AbstractProducerHttpClient(EventMeshHttpClientConfig eventMeshHttpClientConfig)
+ throws EventMeshException {
+ super(eventMeshHttpClientConfig);
+ }
+
+ @Override
+ public void publish(T t) throws EventMeshException {
+ validateMessage(t);
+ String target = selectEventMesh();
+ try {
+ String response = HttpUtils.post(httpClient, target, builderPublishRequestParam(t));
+ EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
+ if (ret.getRetCode() != EventMeshRetCode.SUCCESS.getRetCode()) {
+ throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
+ }
+ } catch (Exception exception) {
+ throw new EventMeshException(String.format("Publish message error, target:%s", target), exception);
+ }
+ }
+
+ @Override
+ public T request(T message, long timeout) throws EventMeshException {
+ validateMessage(message);
+ String target = selectEventMesh();
+ try {
+ String response = HttpUtils.post(httpClient, target, builderRequestParam(message, timeout));
+ EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
+ if (ret.getRetCode() == EventMeshRetCode.SUCCESS.getRetCode()) {
+ return transformMessage(ret);
+ }
+ throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
+ } catch (Exception e) {
+ throw new EventMeshException(String.format("Request message error, target:%s", target), e);
+ }
+ }
+
+ @Override
+ public void request(T message, RRCallback<T> rrCallback, long timeout) throws EventMeshException {
+ validateMessage(message);
+ String target = selectEventMesh();
+ RRCallbackResponseHandlerAdapter<T> adapter = new RRCallbackResponseHandlerAdapter<>(
+ message, rrCallback, timeout);
+ try {
+ HttpUtils.post(httpClient, null, target, builderRequestParam(message, timeout), adapter);
+ } catch (IOException e) {
+ throw new EventMeshException(String.format("Request message error, target:%s", target), e);
+ }
+
+ }
+
+ public abstract RequestParam builderPublishRequestParam(T t);
+
+ public abstract RequestParam builderRequestParam(T t, long timeout);
+
+
+ public abstract void validateMessage(T t);
+
+ public abstract T transformMessage(EventMeshRetObj retObj);
+}
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/CloudEventProducer.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/CloudEventProducer.java
index 5bc421e50..7e6bf5f56 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/CloudEventProducer.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/CloudEventProducer.java
@@ -17,23 +17,20 @@
package org.apache.eventmesh.client.http.producer;
-import org.apache.eventmesh.client.http.AbstractHttpClient;
+import org.apache.eventmesh.client.http.AbstractProducerHttpClient;
import org.apache.eventmesh.client.http.EventMeshRetObj;
import org.apache.eventmesh.client.http.ProtocolConstant;
import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
import org.apache.eventmesh.client.http.model.RequestParam;
-import org.apache.eventmesh.client.http.util.HttpUtils;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
-import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
-import java.io.IOException;
import java.nio.charset.StandardCharsets;
import io.cloudevents.CloudEvent;
@@ -46,71 +43,30 @@ import com.google.common.base.Preconditions;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-class CloudEventProducer extends AbstractHttpClient implements EventMeshProtocolProducer<CloudEvent> {
+class CloudEventProducer extends AbstractProducerHttpClient<CloudEvent> {
public CloudEventProducer(EventMeshHttpClientConfig eventMeshHttpClientConfig) throws EventMeshException {
super(eventMeshHttpClientConfig);
}
@Override
- public void publish(CloudEvent cloudEvent) throws EventMeshException {
- validateCloudEvent(cloudEvent);
+ public RequestParam builderPublishRequestParam(CloudEvent cloudEvent) {
CloudEvent enhanceCloudEvent = enhanceCloudEvent(cloudEvent);
- // todo: Can put to abstract class, all protocol use the same send method? This can be a template method
- RequestParam requestParam = buildCommonPostParam(enhanceCloudEvent)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_ASYNC.getRequestCode());
- String target = selectEventMesh();
- try {
- String response = HttpUtils.post(httpClient, target, requestParam);
- EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
-
- if (ret.getRetCode() != EventMeshRetCode.SUCCESS.getRetCode()) {
- throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
- }
- } catch (Exception exception) {
- throw new EventMeshException(String.format("Publish message error, target:%s", target), exception);
- }
-
+ return buildCommonPostParam(enhanceCloudEvent)
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_ASYNC.getRequestCode());
}
@Override
- public CloudEvent request(CloudEvent cloudEvent, long timeout) throws EventMeshException {
- validateCloudEvent(cloudEvent);
+ public RequestParam builderRequestParam(CloudEvent cloudEvent, long timeout) {
CloudEvent enhanceCloudEvent = enhanceCloudEvent(cloudEvent);
- RequestParam requestParam = buildCommonPostParam(enhanceCloudEvent)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
- .setTimeout(timeout);
- String target = selectEventMesh();
-
- try {
- String response = HttpUtils.post(httpClient, target, requestParam);
- EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
- if (ret.getRetCode() == EventMeshRetCode.SUCCESS.getRetCode()) {
- return transformMessage(ret);
- }
- throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
- } catch (Exception e) {
- throw new EventMeshException(String.format("Request message error, target:%s", target), e);
- }
+ return buildCommonPostParam(enhanceCloudEvent)
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
+ .setTimeout(timeout);
}
@Override
- public void request(final CloudEvent cloudEvent, final RRCallback<CloudEvent> rrCallback, long timeout)
- throws EventMeshException {
- validateCloudEvent(cloudEvent);
- CloudEvent enhanceCloudEvent = enhanceCloudEvent(cloudEvent);
- RequestParam requestParam = buildCommonPostParam(enhanceCloudEvent)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
- .setTimeout(timeout);
- String target = selectEventMesh();
- RRCallbackResponseHandlerAdapter<CloudEvent> adapter = new RRCallbackResponseHandlerAdapter<>(
- enhanceCloudEvent, rrCallback, timeout);
- try {
- HttpUtils.post(httpClient, null, target, requestParam, adapter);
- } catch (IOException e) {
- throw new EventMeshException(String.format("Request message error, target:%s", target), e);
- }
-
+ public void validateMessage(CloudEvent cloudEvent) {
+ Preconditions.checkNotNull(cloudEvent, "CloudEvent cannot be null");
}
private void validateCloudEvent(CloudEvent cloudEvent) {
@@ -120,47 +76,48 @@ class CloudEventProducer extends AbstractHttpClient implements EventMeshProtocol
private RequestParam buildCommonPostParam(CloudEvent cloudEvent) {
validateCloudEvent(cloudEvent);
byte[] bodyByte = EventFormatProvider.getInstance().resolveFormat(cloudEvent.getDataContentType())
- .serialize(cloudEvent);
+ .serialize(cloudEvent);
String content = new String(bodyByte, StandardCharsets.UTF_8);
RequestParam requestParam = new RequestParam(HttpMethod.POST);
requestParam
- .addHeader(ProtocolKey.ClientInstanceKey.ENV, eventMeshHttpClientConfig.getEnv())
- .addHeader(ProtocolKey.ClientInstanceKey.IDC, eventMeshHttpClientConfig.getIdc())
- .addHeader(ProtocolKey.ClientInstanceKey.IP, eventMeshHttpClientConfig.getIp())
- .addHeader(ProtocolKey.ClientInstanceKey.PID, eventMeshHttpClientConfig.getPid())
- .addHeader(ProtocolKey.ClientInstanceKey.SYS, eventMeshHttpClientConfig.getSys())
- .addHeader(ProtocolKey.ClientInstanceKey.USERNAME, eventMeshHttpClientConfig.getUserName())
- .addHeader(ProtocolKey.ClientInstanceKey.PASSWD, eventMeshHttpClientConfig.getPassword())
- .addHeader(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
- .addHeader(ProtocolKey.PROTOCOL_TYPE, ProtocolConstant.CE_PROTOCOL)
- .addHeader(ProtocolKey.PROTOCOL_DESC, ProtocolConstant.PROTOCOL_DESC)
- .addHeader(ProtocolKey.PROTOCOL_VERSION, cloudEvent.getSpecVersion().toString())
-
- // todo: move producerGroup tp header
- .addBody(SendMessageRequestBody.PRODUCERGROUP, eventMeshHttpClientConfig.getProducerGroup())
- .addBody(SendMessageRequestBody.CONTENT, content);
+ .addHeader(ProtocolKey.ClientInstanceKey.ENV, eventMeshHttpClientConfig.getEnv())
+ .addHeader(ProtocolKey.ClientInstanceKey.IDC, eventMeshHttpClientConfig.getIdc())
+ .addHeader(ProtocolKey.ClientInstanceKey.IP, eventMeshHttpClientConfig.getIp())
+ .addHeader(ProtocolKey.ClientInstanceKey.PID, eventMeshHttpClientConfig.getPid())
+ .addHeader(ProtocolKey.ClientInstanceKey.SYS, eventMeshHttpClientConfig.getSys())
+ .addHeader(ProtocolKey.ClientInstanceKey.USERNAME, eventMeshHttpClientConfig.getUserName())
+ .addHeader(ProtocolKey.ClientInstanceKey.PASSWD, eventMeshHttpClientConfig.getPassword())
+ .addHeader(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
+ .addHeader(ProtocolKey.PROTOCOL_TYPE, ProtocolConstant.CE_PROTOCOL)
+ .addHeader(ProtocolKey.PROTOCOL_DESC, ProtocolConstant.PROTOCOL_DESC)
+ .addHeader(ProtocolKey.PROTOCOL_VERSION, cloudEvent.getSpecVersion().toString())
+
+ // todo: move producerGroup tp header
+ .addBody(SendMessageRequestBody.PRODUCERGROUP, eventMeshHttpClientConfig.getProducerGroup())
+ .addBody(SendMessageRequestBody.CONTENT, content);
return requestParam;
}
private CloudEvent enhanceCloudEvent(final CloudEvent cloudEvent) {
return CloudEventBuilder.from(cloudEvent)
- .withExtension(ProtocolKey.ClientInstanceKey.ENV, eventMeshHttpClientConfig.getEnv())
- .withExtension(ProtocolKey.ClientInstanceKey.IDC, eventMeshHttpClientConfig.getIdc())
- .withExtension(ProtocolKey.ClientInstanceKey.IP, eventMeshHttpClientConfig.getIp())
- .withExtension(ProtocolKey.ClientInstanceKey.PID, eventMeshHttpClientConfig.getPid())
- .withExtension(ProtocolKey.ClientInstanceKey.SYS, eventMeshHttpClientConfig.getSys())
- .withExtension(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
- .withExtension(ProtocolKey.PROTOCOL_DESC, cloudEvent.getSpecVersion().name())
- .withExtension(ProtocolKey.PROTOCOL_VERSION, cloudEvent.getSpecVersion().toString())
- .withExtension(ProtocolKey.ClientInstanceKey.BIZSEQNO, RandomStringUtils.generateNum(30))
- .withExtension(ProtocolKey.ClientInstanceKey.UNIQUEID, RandomStringUtils.generateNum(30))
- .build();
+ .withExtension(ProtocolKey.ClientInstanceKey.ENV, eventMeshHttpClientConfig.getEnv())
+ .withExtension(ProtocolKey.ClientInstanceKey.IDC, eventMeshHttpClientConfig.getIdc())
+ .withExtension(ProtocolKey.ClientInstanceKey.IP, eventMeshHttpClientConfig.getIp())
+ .withExtension(ProtocolKey.ClientInstanceKey.PID, eventMeshHttpClientConfig.getPid())
+ .withExtension(ProtocolKey.ClientInstanceKey.SYS, eventMeshHttpClientConfig.getSys())
+ .withExtension(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
+ .withExtension(ProtocolKey.PROTOCOL_DESC, cloudEvent.getSpecVersion().name())
+ .withExtension(ProtocolKey.PROTOCOL_VERSION, cloudEvent.getSpecVersion().toString())
+ .withExtension(ProtocolKey.ClientInstanceKey.BIZSEQNO, RandomStringUtils.generateNum(30))
+ .withExtension(ProtocolKey.ClientInstanceKey.UNIQUEID, RandomStringUtils.generateNum(30))
+ .build();
}
- private CloudEvent transformMessage(EventMeshRetObj retObj) {
+ @Override
+ public CloudEvent transformMessage(EventMeshRetObj retObj) {
SendMessageResponseBody.ReplyMessage replyMessage = JsonUtils.deserialize(retObj.getRetMsg(),
- SendMessageResponseBody.ReplyMessage.class);
+ SendMessageResponseBody.ReplyMessage.class);
// todo: deserialize message
return null;
}
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/EventMeshMessageProducer.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/EventMeshMessageProducer.java
index d4fcf505e..3c48b221b 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/EventMeshMessageProducer.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/EventMeshMessageProducer.java
@@ -17,25 +17,21 @@
package org.apache.eventmesh.client.http.producer;
-import org.apache.eventmesh.client.http.AbstractHttpClient;
+import org.apache.eventmesh.client.http.AbstractProducerHttpClient;
import org.apache.eventmesh.client.http.EventMeshRetObj;
import org.apache.eventmesh.client.http.ProtocolConstant;
import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
import org.apache.eventmesh.client.http.model.RequestParam;
-import org.apache.eventmesh.client.http.util.HttpUtils;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.EventMeshMessage;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
-import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.JsonUtils;
-import java.io.IOException;
-
import io.cloudevents.SpecVersion;
import io.netty.handler.codec.http.HttpMethod;
@@ -44,69 +40,31 @@ import com.google.common.base.Preconditions;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-class EventMeshMessageProducer extends AbstractHttpClient implements EventMeshProtocolProducer<EventMeshMessage> {
+class EventMeshMessageProducer extends AbstractProducerHttpClient<EventMeshMessage> {
public EventMeshMessageProducer(EventMeshHttpClientConfig eventMeshHttpClientConfig) throws EventMeshException {
super(eventMeshHttpClientConfig);
}
@Override
- public void publish(EventMeshMessage message) throws EventMeshException {
- validateEventMeshMessage(message);
- RequestParam requestParam = buildCommonPostParam(message)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_ASYNC.getRequestCode());
-
- String target = selectEventMesh();
- try {
- String response = HttpUtils.post(httpClient, target, requestParam);
- EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
-
- if (ret.getRetCode() != EventMeshRetCode.SUCCESS.getRetCode()) {
- throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
- }
- } catch (Exception exception) {
- throw new EventMeshException(String.format("Publish message error, target:%s", target), exception);
- }
+ public RequestParam builderPublishRequestParam(EventMeshMessage message) {
+ return buildCommonPostParam(message)
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_ASYNC.getRequestCode());
}
@Override
- public EventMeshMessage request(EventMeshMessage message, long timeout) throws EventMeshException {
- validateEventMeshMessage(message);
- RequestParam requestParam = buildCommonPostParam(message)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
- .setTimeout(timeout);
-
- String target = selectEventMesh();
-
- try {
- String response = HttpUtils.post(httpClient, target, requestParam);
- EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
- if (ret.getRetCode() == EventMeshRetCode.SUCCESS.getRetCode()) {
- return transformMessage(ret);
- }
- throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
- } catch (Exception e) {
- throw new EventMeshException(String.format("Request message error, target:%s", target), e);
- }
+ public RequestParam builderRequestParam(EventMeshMessage message, long timeout) {
+ return buildCommonPostParam(message)
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
+ .setTimeout(timeout);
}
@Override
- public void request(EventMeshMessage message, RRCallback<EventMeshMessage> rrCallback, long timeout)
- throws EventMeshException {
- validateEventMeshMessage(message);
-
- RequestParam requestParam = buildCommonPostParam(message)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
- .setTimeout(timeout);
+ public void validateMessage(EventMeshMessage message) {
+ Preconditions.checkNotNull(message, "eventMeshMessage invalid");
+ Preconditions.checkNotNull(message.getTopic(), "eventMeshMessage[topic] invalid");
+ Preconditions.checkNotNull(message.getContent(), "eventMeshMessage[content] invalid");
- String target = selectEventMesh();
- RRCallbackResponseHandlerAdapter<EventMeshMessage> adapter =
- new RRCallbackResponseHandlerAdapter<>(message, rrCallback, timeout);
- try {
- HttpUtils.post(httpClient, null, target, requestParam, adapter);
- } catch (IOException e) {
- throw new EventMeshException(String.format("Request message error, target:%s", target), e);
- }
}
private void validateEventMeshMessage(EventMeshMessage message) {
@@ -118,35 +76,36 @@ class EventMeshMessageProducer extends AbstractHttpClient implements EventMeshPr
private RequestParam buildCommonPostParam(EventMeshMessage message) {
RequestParam requestParam = new RequestParam(HttpMethod.POST);
requestParam
- .addHeader(ProtocolKey.ClientInstanceKey.ENV, eventMeshHttpClientConfig.getEnv())
- .addHeader(ProtocolKey.ClientInstanceKey.IDC, eventMeshHttpClientConfig.getIdc())
- .addHeader(ProtocolKey.ClientInstanceKey.IP, eventMeshHttpClientConfig.getIp())
- .addHeader(ProtocolKey.ClientInstanceKey.PID, eventMeshHttpClientConfig.getPid())
- .addHeader(ProtocolKey.ClientInstanceKey.SYS, eventMeshHttpClientConfig.getSys())
- .addHeader(ProtocolKey.ClientInstanceKey.USERNAME, eventMeshHttpClientConfig.getUserName())
- .addHeader(ProtocolKey.ClientInstanceKey.PASSWD, eventMeshHttpClientConfig.getPassword())
- .addHeader(ProtocolKey.VERSION, ProtocolVersion.V1.getVersion())
- .addHeader(ProtocolKey.PROTOCOL_TYPE, ProtocolConstant.EM_MESSAGE_PROTOCOL)
- .addHeader(ProtocolKey.PROTOCOL_DESC, ProtocolConstant.PROTOCOL_DESC)
- //default ce version is 1.0
- .addHeader(ProtocolKey.PROTOCOL_VERSION, SpecVersion.V1.toString())
- .addHeader(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
- .addBody(SendMessageRequestBody.PRODUCERGROUP, eventMeshHttpClientConfig.getProducerGroup())
- // todo: set message to content is better
- .addBody(SendMessageRequestBody.TOPIC, message.getTopic())
- .addBody(SendMessageRequestBody.CONTENT, message.getContent())
- .addBody(SendMessageRequestBody.TTL, message.getProp(Constants.EVENTMESH_MESSAGE_CONST_TTL))
- .addBody(SendMessageRequestBody.BIZSEQNO, message.getBizSeqNo())
- .addBody(SendMessageRequestBody.UNIQUEID, message.getUniqueId());
+ .addHeader(ProtocolKey.ClientInstanceKey.ENV, eventMeshHttpClientConfig.getEnv())
+ .addHeader(ProtocolKey.ClientInstanceKey.IDC, eventMeshHttpClientConfig.getIdc())
+ .addHeader(ProtocolKey.ClientInstanceKey.IP, eventMeshHttpClientConfig.getIp())
+ .addHeader(ProtocolKey.ClientInstanceKey.PID, eventMeshHttpClientConfig.getPid())
+ .addHeader(ProtocolKey.ClientInstanceKey.SYS, eventMeshHttpClientConfig.getSys())
+ .addHeader(ProtocolKey.ClientInstanceKey.USERNAME, eventMeshHttpClientConfig.getUserName())
+ .addHeader(ProtocolKey.ClientInstanceKey.PASSWD, eventMeshHttpClientConfig.getPassword())
+ .addHeader(ProtocolKey.VERSION, ProtocolVersion.V1.getVersion())
+ .addHeader(ProtocolKey.PROTOCOL_TYPE, ProtocolConstant.EM_MESSAGE_PROTOCOL)
+ .addHeader(ProtocolKey.PROTOCOL_DESC, ProtocolConstant.PROTOCOL_DESC)
+ //default ce version is 1.0
+ .addHeader(ProtocolKey.PROTOCOL_VERSION, SpecVersion.V1.toString())
+ .addHeader(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
+ .addBody(SendMessageRequestBody.PRODUCERGROUP, eventMeshHttpClientConfig.getProducerGroup())
+ // todo: set message to content is better
+ .addBody(SendMessageRequestBody.TOPIC, message.getTopic())
+ .addBody(SendMessageRequestBody.CONTENT, message.getContent())
+ .addBody(SendMessageRequestBody.TTL, message.getProp(Constants.EVENTMESH_MESSAGE_CONST_TTL))
+ .addBody(SendMessageRequestBody.BIZSEQNO, message.getBizSeqNo())
+ .addBody(SendMessageRequestBody.UNIQUEID, message.getUniqueId());
return requestParam;
}
- private EventMeshMessage transformMessage(EventMeshRetObj retObj) {
+ @Override
+ public EventMeshMessage transformMessage(EventMeshRetObj retObj) {
SendMessageResponseBody.ReplyMessage replyMessage = JsonUtils.deserialize(retObj.getRetMsg(),
- SendMessageResponseBody.ReplyMessage.class);
+ SendMessageResponseBody.ReplyMessage.class);
return EventMeshMessage.builder()
- .content(replyMessage.body)
- .prop(replyMessage.properties)
- .topic(replyMessage.topic).build();
+ .content(replyMessage.body)
+ .prop(replyMessage.properties)
+ .topic(replyMessage.topic).build();
}
}
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/OpenMessageProducer.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/OpenMessageProducer.java
index b817158cd..bfaed8d45 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/OpenMessageProducer.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/OpenMessageProducer.java
@@ -17,23 +17,19 @@
package org.apache.eventmesh.client.http.producer;
-import org.apache.eventmesh.client.http.AbstractHttpClient;
+import org.apache.eventmesh.client.http.AbstractProducerHttpClient;
import org.apache.eventmesh.client.http.EventMeshRetObj;
import org.apache.eventmesh.client.http.ProtocolConstant;
import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
import org.apache.eventmesh.client.http.model.RequestParam;
-import org.apache.eventmesh.client.http.util.HttpUtils;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
-import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.JsonUtils;
-import java.io.IOException;
-
import io.netty.handler.codec.http.HttpMethod;
import io.openmessaging.api.Message;
@@ -42,88 +38,49 @@ import com.google.common.base.Preconditions;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-class OpenMessageProducer extends AbstractHttpClient implements EventMeshProtocolProducer<Message> {
+class OpenMessageProducer extends AbstractProducerHttpClient<Message> {
public OpenMessageProducer(EventMeshHttpClientConfig eventMeshHttpClientConfig)
- throws EventMeshException {
+ throws EventMeshException {
super(eventMeshHttpClientConfig);
}
@Override
- public void publish(Message openMessage) throws EventMeshException {
- validateOpenMessage(openMessage);
- RequestParam requestParam = buildCommonPostParam(openMessage)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_ASYNC.getRequestCode());
- String target = selectEventMesh();
- try {
- String response = HttpUtils.post(httpClient, target, requestParam);
- EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
-
- if (ret.getRetCode() != EventMeshRetCode.SUCCESS.getRetCode()) {
- throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
- }
- } catch (Exception exception) {
- throw new EventMeshException(String.format("Publish message error, target:%s", target), exception);
- }
+ public RequestParam builderPublishRequestParam(Message openMessage) {
+ return buildCommonPostParam(openMessage)
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_ASYNC.getRequestCode());
}
@Override
- public Message request(Message message, long timeout) throws EventMeshException {
- validateOpenMessage(message);
- RequestParam requestParam = buildCommonPostParam(message)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
- .setTimeout(timeout);
- String target = selectEventMesh();
-
- try {
- String response = HttpUtils.post(httpClient, target, requestParam);
- EventMeshRetObj ret = JsonUtils.deserialize(response, EventMeshRetObj.class);
- if (ret.getRetCode() == EventMeshRetCode.SUCCESS.getRetCode()) {
- return transformMessage(ret);
- }
- throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
- } catch (Exception e) {
- throw new EventMeshException(String.format("Request message error, target:%s", target), e);
- }
+ public RequestParam builderRequestParam(Message message, long timeout) {
+ return buildCommonPostParam(message)
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
+ .setTimeout(timeout);
}
@Override
- public void request(Message message, RRCallback<Message> rrCallback, long timeout) throws EventMeshException {
- validateOpenMessage(message);
- RequestParam requestParam = buildCommonPostParam(message)
- .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
- .setTimeout(timeout);
- String target = selectEventMesh();
- RRCallbackResponseHandlerAdapter<Message> adapter =
- new RRCallbackResponseHandlerAdapter<>(message, rrCallback, timeout);
- try {
- HttpUtils.post(httpClient, null, target, requestParam, adapter);
- } catch (IOException e) {
- throw new EventMeshException(String.format("Request message error, target:%s", target), e);
- }
- }
-
- private void validateOpenMessage(Message openMessage) {
- Preconditions.checkNotNull(openMessage, "Message cannot be null");
+ public void validateMessage(Message message) {
+ Preconditions.checkNotNull(message, "Message cannot be null");
}
private RequestParam buildCommonPostParam(Message openMessage) {
RequestParam requestParam = new RequestParam(HttpMethod.POST);
requestParam
- .addHeader(ProtocolKey.ClientInstanceKey.USERNAME, eventMeshHttpClientConfig.getUserName())
- .addHeader(ProtocolKey.ClientInstanceKey.PASSWD, eventMeshHttpClientConfig.getPassword())
- .addHeader(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
- .addHeader(ProtocolKey.PROTOCOL_TYPE, ProtocolConstant.OP_MESSAGE_PROTOCOL)
- .addHeader(ProtocolKey.PROTOCOL_DESC, ProtocolConstant.PROTOCOL_DESC)
- // todo: add producerGroup to header, set protocol type, protocol version
- .addBody(SendMessageRequestBody.PRODUCERGROUP, eventMeshHttpClientConfig.getProducerGroup())
- .addBody(SendMessageRequestBody.CONTENT, JsonUtils.serialize(openMessage));
+ .addHeader(ProtocolKey.ClientInstanceKey.USERNAME, eventMeshHttpClientConfig.getUserName())
+ .addHeader(ProtocolKey.ClientInstanceKey.PASSWD, eventMeshHttpClientConfig.getPassword())
+ .addHeader(ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA)
+ .addHeader(ProtocolKey.PROTOCOL_TYPE, ProtocolConstant.OP_MESSAGE_PROTOCOL)
+ .addHeader(ProtocolKey.PROTOCOL_DESC, ProtocolConstant.PROTOCOL_DESC)
+ // todo: add producerGroup to header, set protocol type, protocol version
+ .addBody(SendMessageRequestBody.PRODUCERGROUP, eventMeshHttpClientConfig.getProducerGroup())
+ .addBody(SendMessageRequestBody.CONTENT, JsonUtils.serialize(openMessage));
return requestParam;
}
- private Message transformMessage(EventMeshRetObj retObj) {
+ @Override
+ public Message transformMessage(EventMeshRetObj retObj) {
SendMessageResponseBody.ReplyMessage replyMessage = JsonUtils.deserialize(retObj.getRetMsg(),
- SendMessageResponseBody.ReplyMessage.class);
+ SendMessageResponseBody.ReplyMessage.class);
// todo: deserialize message
return null;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org