You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by aa...@apache.org on 2022/08/23 06:06:58 UTC
[rocketmq-clients] branch master updated: Refactor message interceptor (#162)
This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 2bf1bf5 Refactor message interceptor (#162)
2bf1bf5 is described below
commit 2bf1bf5d443ebbb69e8ba0c7688a9be8a48e1a43
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Tue Aug 23 14:06:52 2022 +0800
Refactor message interceptor (#162)
---
.../rocketmq/client/apis/ClientConfiguration.java | 2 +-
.../rocketmq/client/apis/message/Message.java | 5 +-
.../rocketmq/client/apis/message/MessageView.java | 6 +
.../java/example/ProducerNormalMessageExample.java | 8 +-
.../rocketmq/client/java/hook/Attribute.java | 34 ++++
.../rocketmq/client/java/hook/AttributeKey.java | 56 ++++++
.../client/java/hook/CompositedMessageHandler.java | 75 ++++++++
...MessageInterceptor.java => MessageHandler.java} | 26 +--
.../client/java/hook/MessageHandlerContext.java | 28 +++
.../java/hook/MessageHandlerContextImpl.java | 82 +++++++++
.../rocketmq/client/java/impl/ClientImpl.java | 67 +++----
.../client/java/impl/consumer/ConsumeService.java | 10 +-
.../client/java/impl/consumer/ConsumeTask.java | 26 +--
.../client/java/impl/consumer/ConsumerImpl.java | 52 +++---
.../java/impl/consumer/FifoConsumeService.java | 6 +-
.../impl/consumer/MessageCacheObserverImpl.java | 57 ++++++
.../java/impl/consumer/ProcessQueueImpl.java | 31 ++--
.../java/impl/consumer/PushConsumerImpl.java | 56 ++----
.../java/impl/consumer/StandardConsumeService.java | 6 +-
.../client/java/impl/producer/ProducerImpl.java | 58 +++---
.../client/java/impl/producer/TransactionImpl.java | 5 +-
.../client/java/message/GeneralMessage.java} | 59 ++++--
.../client/java/message/GeneralMessageImpl.java | 187 +++++++++++++++++++
.../client/java/message/MessageCommon.java | 160 ----------------
.../rocketmq/client/java/message/MessageImpl.java | 13 +-
.../client/java/message/MessageViewImpl.java | 42 +++--
.../client/java/message/PublishingMessageImpl.java | 2 +-
.../rocketmq/client/java/metrics/ClientMeter.java | 10 +
.../client/java/metrics/ClientMeterProvider.java | 13 +-
.../client/java/metrics/MessageMeterHandler.java | 202 +++++++++++++++++++++
.../java/metrics/MessageMeterInterceptor.java | 192 --------------------
.../java/impl/consumer/ConsumeServiceTest.java | 4 +-
.../client/java/impl/consumer/ConsumeTaskTest.java | 10 +-
.../java/impl/consumer/FifoConsumeServiceTest.java | 16 +-
.../impl/consumer/StandardConsumeServiceTest.java | 17 +-
.../java/impl/producer/TransactionImplTest.java | 6 +-
.../java/message/GeneralMessageImplTest.java | 125 +++++++++++++
.../client/java/message/MessageViewImplTest.java | 160 ++++++++++++++++
.../apache/rocketmq/client/java/tool/TestBase.java | 3 +-
39 files changed, 1289 insertions(+), 628 deletions(-)
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java
index 8ee8dee..5b0f004 100644
--- a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java
+++ b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java
@@ -48,7 +48,7 @@ public class ClientConfiguration {
}
public Optional<SessionCredentialsProvider> getCredentialsProvider() {
- return null == sessionCredentialsProvider ? Optional.empty() : Optional.of(sessionCredentialsProvider);
+ return Optional.ofNullable(sessionCredentialsProvider);
}
public Duration getRequestTimeout() {
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/Message.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/Message.java
index b0f4d7c..2fea2b4 100644
--- a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/Message.java
+++ b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/Message.java
@@ -73,9 +73,10 @@ public interface Message {
Optional<String> getMessageGroup();
/**
- * Get the trace context,
+ * Get the parent trace context, see
+ * <a href="https://opentelemetry.io/docs/concepts/signals/traces/#trace-context">OpenTelemetry Trace context</a>.
*
- * @return trace context, which is optional, {@link Optional#empty()} means trace context is not specified.
+ * @return parent trace context, which is optional, {@link Optional#empty()} means trace context is not specified.
*/
Optional<String> getParentTraceContext();
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java
index 04d576b..761c21c 100644
--- a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java
+++ b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java
@@ -94,6 +94,12 @@ public interface MessageView {
*/
String getBornHost();
+ /**
+ * Message trace context, see
+ * <a href="https://opentelemetry.io/docs/concepts/signals/traces/#trace-context">OpenTelemetry Trace context</a>.
+ *
+ * @return message trace context.
+ */
Optional<String> getTraceContext();
/**
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/example/ProducerNormalMessageExample.java b/java/client/src/main/java/org/apache/rocketmq/client/java/example/ProducerNormalMessageExample.java
index 1b406f5..266aa01 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/example/ProducerNormalMessageExample.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/example/ProducerNormalMessageExample.java
@@ -40,17 +40,17 @@ public class ProducerNormalMessageExample {
final ClientServiceProvider provider = ClientServiceProvider.loadService();
// Credential provider is optional for client configuration.
- String accessKey = "yourAccessKey";
- String secretKey = "yourSecretKey";
+ String accessKey = "wyME6gI8ZMW4b2s4";
+ String secretKey = "E0NEQm7isSKn0R48";
SessionCredentialsProvider sessionCredentialsProvider =
new StaticSessionCredentialsProvider(accessKey, secretKey);
- String endpoints = "foobar.com:8080";
+ String endpoints = "rmq-cn-c4d2toqfu25.cn-zhangjiakou.rmq.aliyuncs.com:8080";
ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints(endpoints)
.setCredentialProvider(sessionCredentialsProvider)
.build();
- String topic = "yourNormalTopic";
+ String topic = "normalTopic";
final Producer producer = provider.newProducerBuilder()
.setClientConfiguration(clientConfiguration)
// Set the topic name(s), which is optional. It makes producer could prefetch the topic route before
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/Attribute.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/Attribute.java
new file mode 100644
index 0000000..78e9cb9
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/Attribute.java
@@ -0,0 +1,34 @@
+/*
+ * 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.rocketmq.client.java.hook;
+
+public class Attribute<T> {
+ private final T t;
+
+ private Attribute(T t) {
+ this.t = t;
+ }
+
+ public static <T> Attribute<T> create(T t) {
+ return new Attribute<>(t);
+ }
+
+ public T get() {
+ return t;
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/AttributeKey.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/AttributeKey.java
new file mode 100644
index 0000000..797249a
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/AttributeKey.java
@@ -0,0 +1,56 @@
+/*
+ * 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.rocketmq.client.java.hook;
+
+import com.google.common.base.Objects;
+
+/**
+ * Key which can be used to access {@link Attribute}. Be aware that it is not possible to have multiple keys with the
+ * same name.
+ *
+ * @param <T> the type of the {@link Attribute} which can be accessed via this {@link AttributeKey}.
+ */
+@SuppressWarnings("UnusedDeclaration") // 'T' is used only at compile time
+public class AttributeKey<T> {
+ private final String name;
+
+ private AttributeKey(String name) {
+ this.name = name;
+ }
+
+ public static <T> AttributeKey<T> create(String name) {
+ return new AttributeKey<>(name);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ AttributeKey<?> that = (AttributeKey<?>) o;
+ return Objects.equal(name, that.name);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(name);
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/CompositedMessageHandler.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/CompositedMessageHandler.java
new file mode 100644
index 0000000..7a021ab
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/CompositedMessageHandler.java
@@ -0,0 +1,75 @@
+/*
+ * 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.rocketmq.client.java.hook;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@SuppressWarnings("rawtypes")
+public class CompositedMessageHandler implements MessageHandler {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MessageHandler.class);
+ private static final AttributeKey<Map<Integer, Map<AttributeKey, Attribute>>> HANDLER_ATTRIBUTES_KEY =
+ AttributeKey.create("composited_handler_attributes");
+ private final List<MessageHandler> handlers;
+
+ public CompositedMessageHandler(List<MessageHandler> handlers) {
+ this.handlers = handlers;
+ }
+
+ @Override
+ public void doBefore(MessageHandlerContext context0, List<GeneralMessage> messages) {
+ final HashMap<Integer, Map<AttributeKey, Attribute>> attributeMap = new HashMap<>();
+ for (int index = 0; index < handlers.size(); index++) {
+ MessageHandler handler = handlers.get(index);
+ final MessageHookPoints messageHookPoints = context0.getMessageHookPoints();
+ final MessageHookPointsStatus status = context0.getStatus();
+ final MessageHandlerContextImpl context = new MessageHandlerContextImpl(messageHookPoints, status);
+ try {
+ handler.doBefore(context, messages);
+ } catch (Throwable t) {
+ LOGGER.error("Exception raised while handing messages", t);
+ }
+ final Map<AttributeKey, Attribute> attributes = context.getAttributes();
+ attributeMap.put(index, attributes);
+ }
+ context0.putAttribute(HANDLER_ATTRIBUTES_KEY, Attribute.create(attributeMap));
+ }
+
+ @Override
+ public void doAfter(MessageHandlerContext context0, List<GeneralMessage> messages) {
+ for (int index = handlers.size() - 1; index >= 0; index--) {
+ final Map<Integer, Map<AttributeKey, Attribute>> attributeMap =
+ context0.getAttribute(HANDLER_ATTRIBUTES_KEY).get();
+ final Map<AttributeKey, Attribute> attributes = attributeMap.get(index);
+ final MessageHookPoints messageHookPoints = context0.getMessageHookPoints();
+ final MessageHookPointsStatus status = context0.getStatus();
+ final MessageHandlerContextImpl context = new MessageHandlerContextImpl(messageHookPoints, status,
+ attributes);
+ MessageHandler handler = handlers.get(index);
+ try {
+ handler.doAfter(context, messages);
+ } catch (Throwable t) {
+ LOGGER.error("Exception raised while handing messages", t);
+ }
+ }
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageInterceptor.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandler.java
similarity index 51%
rename from java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageInterceptor.java
rename to java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandler.java
index ad717c0..b2c5087 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageInterceptor.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandler.java
@@ -17,30 +17,16 @@
package org.apache.rocketmq.client.java.hook;
-import java.time.Duration;
+import com.google.common.annotations.Beta;
import java.util.List;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
/**
* Message interceptor based on {@link MessageHookPoints}.
*/
-public interface MessageInterceptor {
- /**
- * Do something before {@link MessageHookPoints}.
- *
- * @param messageHookPoints message hook points.
- * @param messageCommons list of message commons.
- */
- void doBefore(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons);
+@Beta
+public interface MessageHandler {
+ void doBefore(MessageHandlerContext context, List<GeneralMessage> messages);
- /**
- * Do something after {@link MessageHookPoints}.
- *
- * @param messageHookPoints message hook points.
- * @param messageCommons list of message commons.
- * @param duration duration of the hook points.
- * @param status status of operation of the hook points.
- */
- void doAfter(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons, Duration duration,
- MessageHookPointsStatus status);
+ void doAfter(MessageHandlerContext context, List<GeneralMessage> messages);
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandlerContext.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandlerContext.java
new file mode 100644
index 0000000..e38aaec
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandlerContext.java
@@ -0,0 +1,28 @@
+/*
+ * 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.rocketmq.client.java.hook;
+
+public interface MessageHandlerContext {
+ MessageHookPoints getMessageHookPoints();
+
+ MessageHookPointsStatus getStatus();
+
+ <T> Attribute<T> getAttribute(AttributeKey<T> key);
+
+ <T> void putAttribute(AttributeKey<T> key, Attribute<T> attribute);
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandlerContextImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandlerContextImpl.java
new file mode 100644
index 0000000..486cd30
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/MessageHandlerContextImpl.java
@@ -0,0 +1,82 @@
+/*
+ * 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.rocketmq.client.java.hook;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@SuppressWarnings("rawtypes")
+public class MessageHandlerContextImpl implements MessageHandlerContext {
+ private final MessageHookPoints messageHookPoints;
+ private MessageHookPointsStatus status;
+ private final Map<AttributeKey, Attribute> attributes;
+
+ public MessageHandlerContextImpl(MessageHookPoints messageHookPoints) {
+ this.messageHookPoints = messageHookPoints;
+ this.status = MessageHookPointsStatus.UNSET;
+ this.attributes = new HashMap<>();
+ }
+
+ public MessageHandlerContextImpl(MessageHookPoints messageHookPoints, MessageHookPointsStatus status) {
+ this.messageHookPoints = messageHookPoints;
+ this.status = status;
+ this.attributes = new HashMap<>();
+ }
+
+ public MessageHandlerContextImpl(MessageHookPoints messageHookPoints, MessageHookPointsStatus status,
+ Map<AttributeKey, Attribute> attributes) {
+ this.messageHookPoints = messageHookPoints;
+ this.status = status;
+ this.attributes = attributes;
+ }
+
+ public MessageHandlerContextImpl(MessageHandlerContextImpl context, MessageHookPointsStatus status) {
+ this.messageHookPoints = context.messageHookPoints;
+ this.status = status;
+ this.attributes = context.attributes;
+ }
+
+ @Override
+ public MessageHookPoints getMessageHookPoints() {
+ return messageHookPoints;
+ }
+
+ @Override
+ public MessageHookPointsStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(MessageHookPointsStatus status) {
+ this.status = status;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> Attribute<T> getAttribute(AttributeKey<T> key) {
+ return attributes.get(key);
+ }
+
+ @Override
+ public <T> void putAttribute(AttributeKey<T> key, Attribute<T> attribute) {
+ attributes.put(key, attribute);
+ }
+
+ public Map<AttributeKey, Attribute> getAttributes() {
+ return attributes;
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
index dc20ff1..b9b6a76 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
@@ -47,7 +47,6 @@ import io.grpc.stub.StreamObserver;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -72,12 +71,13 @@ import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.java.exception.InternalErrorException;
import org.apache.rocketmq.client.java.exception.StatusChecker;
-import org.apache.rocketmq.client.java.hook.MessageHookPoints;
-import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
+import org.apache.rocketmq.client.java.hook.CompositedMessageHandler;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContext;
import org.apache.rocketmq.client.java.impl.producer.ClientSessionHandler;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
import org.apache.rocketmq.client.java.metrics.ClientMeterProvider;
+import org.apache.rocketmq.client.java.metrics.MessageMeterHandler;
import org.apache.rocketmq.client.java.metrics.Metric;
import org.apache.rocketmq.client.java.misc.ExecutorServices;
import org.apache.rocketmq.client.java.misc.ThreadFactoryImpl;
@@ -91,7 +91,7 @@ import org.slf4j.LoggerFactory;
@SuppressWarnings({"UnstableApiUsage", "NullableProblems"})
public abstract class ClientImpl extends AbstractIdleService implements Client, ClientSessionHandler,
- MessageInterceptor {
+ MessageHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(ClientImpl.class);
/**
* The telemetry timeout should not be too long, otherwise
@@ -124,9 +124,7 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
private final Map<Endpoints, ClientSessionImpl> sessionsTable;
private final ReadWriteLock sessionsLock;
- @GuardedBy("messageInterceptorsLock")
- private final List<MessageInterceptor> messageInterceptors;
- private final ReadWriteLock messageInterceptorsLock;
+ private final CompositedMessageHandler compositedMessageHandler;
public ClientImpl(ClientConfiguration clientConfiguration, Set<String> topics) {
this.clientConfiguration = checkNotNull(clientConfiguration, "clientConfiguration should not be null");
@@ -145,9 +143,6 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
this.isolated = Collections.newSetFromMap(new ConcurrentHashMap<>());
- this.messageInterceptors = new ArrayList<>();
- this.messageInterceptorsLock = new ReentrantReadWriteLock();
-
this.clientManager = new ClientManagerImpl(this);
this.clientCallbackExecutor = new ThreadPoolExecutor(
@@ -159,6 +154,10 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
new ThreadFactoryImpl("ClientCallbackWorker"));
this.clientMeterProvider = new ClientMeterProvider(this);
+
+ this.compositedMessageHandler =
+ new CompositedMessageHandler(Collections.singletonList(new MessageMeterHandler(this, clientMeterProvider)));
+
this.telemetryCommandExecutor = new ThreadPoolExecutor(
1,
1,
@@ -229,47 +228,23 @@ public abstract class ClientImpl extends AbstractIdleService implements Client,
LOGGER.info("Shutdown the rocketmq client successfully, clientId={}", clientId);
}
- public void registerMessageInterceptor(MessageInterceptor messageInterceptor) {
- messageInterceptorsLock.writeLock().lock();
- try {
- messageInterceptors.add(messageInterceptor);
- } finally {
- messageInterceptorsLock.writeLock().unlock();
- }
- }
-
@Override
- public void doBefore(MessageHookPoints hookPoint, List<MessageCommon> messageCommons) {
- messageInterceptorsLock.readLock().lock();
+ public void doBefore(MessageHandlerContext context, List<GeneralMessage> generalMessages) {
try {
- for (MessageInterceptor interceptor : messageInterceptors) {
- try {
- interceptor.doBefore(hookPoint, messageCommons);
- } catch (Throwable t) {
- LOGGER.warn("Exception raised while intercepting message, hookPoint={}, clientId={}", hookPoint,
- clientId);
- }
- }
- } finally {
- messageInterceptorsLock.readLock().unlock();
+ compositedMessageHandler.doBefore(context, generalMessages);
+ } catch (Throwable t) {
+ // Should never reach here.
+ LOGGER.error("[BUG] Exception raised while handling messages, clientId={}", clientId, t);
}
}
@Override
- public void doAfter(MessageHookPoints hookPoints, List<MessageCommon> messageCommons, Duration duration,
- MessageHookPointsStatus status) {
- messageInterceptorsLock.readLock().lock();
+ public void doAfter(MessageHandlerContext context, List<GeneralMessage> generalMessages) {
try {
- for (MessageInterceptor interceptor : messageInterceptors) {
- try {
- interceptor.doAfter(hookPoints, messageCommons, duration, status);
- } catch (Throwable t) {
- LOGGER.warn("Exception raised while intercepting message, hookPoint={}, clientId={}", hookPoints,
- clientId);
- }
- }
- } finally {
- messageInterceptorsLock.readLock().unlock();
+ compositedMessageHandler.doAfter(context, generalMessages);
+ } catch (Throwable t) {
+ // Should never reach here.
+ LOGGER.error("[BUG] Exception raised while handling messages, clientId={}", clientId, t);
}
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
index 9115b00..859a668 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
@@ -30,7 +30,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.misc.Dispatcher;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
@@ -46,18 +46,18 @@ public abstract class ConsumeService extends Dispatcher {
protected final String clientId;
private final MessageListener messageListener;
private final ThreadPoolExecutor consumptionExecutor;
- private final MessageInterceptor messageInterceptor;
+ private final MessageHandler messageHandler;
private final ScheduledExecutorService scheduler;
public ConsumeService(String clientId, ConcurrentMap<MessageQueueImpl, ProcessQueue> processQueueTable,
- MessageListener messageListener, ThreadPoolExecutor consumptionExecutor, MessageInterceptor messageInterceptor,
+ MessageListener messageListener, ThreadPoolExecutor consumptionExecutor, MessageHandler messageHandler,
ScheduledExecutorService scheduler) {
super(clientId);
this.clientId = clientId;
this.processQueueTable = processQueueTable;
this.messageListener = messageListener;
this.consumptionExecutor = consumptionExecutor;
- this.messageInterceptor = messageInterceptor;
+ this.messageHandler = messageHandler;
this.scheduler = scheduler;
}
@@ -67,7 +67,7 @@ public abstract class ConsumeService extends Dispatcher {
public ListenableFuture<ConsumeResult> consume(MessageViewImpl messageView, Duration delay) {
final ListeningExecutorService executorService = MoreExecutors.listeningDecorator(consumptionExecutor);
- final ConsumeTask task = new ConsumeTask(clientId, messageListener, messageView, messageInterceptor);
+ final ConsumeTask task = new ConsumeTask(clientId, messageListener, messageView, messageHandler);
// Consume message with no delay.
if (Duration.ZERO.compareTo(delay) >= 0) {
return executorService.submit(task);
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java
index 27fbc36..558c59f 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java
@@ -17,17 +17,17 @@
package org.apache.rocketmq.client.java.impl.consumer;
-import com.google.common.base.Stopwatch;
-import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContextImpl;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+import org.apache.rocketmq.client.java.message.GeneralMessageImpl;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,14 +38,14 @@ public class ConsumeTask implements Callable<ConsumeResult> {
private final String clientId;
private final MessageListener messageListener;
private final MessageViewImpl messageView;
- private final MessageInterceptor messageInterceptor;
+ private final MessageHandler messageHandler;
public ConsumeTask(String clientId, MessageListener messageListener, MessageViewImpl messageView,
- MessageInterceptor messageInterceptor) {
+ MessageHandler messageHandler) {
this.clientId = clientId;
this.messageListener = messageListener;
this.messageView = messageView;
- this.messageInterceptor = messageInterceptor;
+ this.messageHandler = messageHandler;
}
/**
@@ -56,20 +56,20 @@ public class ConsumeTask implements Callable<ConsumeResult> {
@Override
public ConsumeResult call() {
ConsumeResult consumeResult;
- final List<MessageCommon> messageCommons = Collections.singletonList(messageView.getMessageCommon());
- messageInterceptor.doBefore(MessageHookPoints.CONSUME, messageCommons);
- final Stopwatch stopwatch = Stopwatch.createStarted();
+ final List<GeneralMessage> generalMessages = Collections.singletonList(new GeneralMessageImpl(messageView));
+ MessageHandlerContextImpl context = new MessageHandlerContextImpl(MessageHookPoints.CONSUME);
+ messageHandler.doBefore(context, generalMessages);
try {
consumeResult = messageListener.consume(messageView);
} catch (Throwable t) {
- LOGGER.error("Message listener raised an exception while consuming messages, client id={}", clientId, t);
+ LOGGER.error("Message listener raised an exception while consuming messages, clientId={}", clientId, t);
// If exception was thrown during the period of message consumption, mark it as failure.
consumeResult = ConsumeResult.FAILURE;
}
- final Duration duration = stopwatch.elapsed();
MessageHookPointsStatus status = ConsumeResult.SUCCESS.equals(consumeResult) ? MessageHookPointsStatus.OK :
MessageHookPointsStatus.ERROR;
- messageInterceptor.doAfter(MessageHookPoints.CONSUME, messageCommons, duration, status);
+ context = new MessageHandlerContextImpl(context, status);
+ messageHandler.doAfter(context, generalMessages);
// Make sure that the return value is the subset of messageViews.
return consumeResult;
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java
index 9c60856..1a2bf16 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumerImpl.java
@@ -31,13 +31,13 @@ import apache.rocketmq.v2.ReceiveMessageRequest;
import apache.rocketmq.v2.ReceiveMessageResponse;
import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.Status;
-import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.Timestamp;
import com.google.protobuf.util.Durations;
+import com.google.protobuf.util.Timestamps;
import io.grpc.Metadata;
import java.time.Duration;
import java.util.ArrayList;
@@ -49,11 +49,13 @@ import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.java.exception.StatusChecker;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContextImpl;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.impl.ClientImpl;
import org.apache.rocketmq.client.java.impl.ClientManager;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+import org.apache.rocketmq.client.java.message.GeneralMessageImpl;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
@@ -88,7 +90,7 @@ abstract class ConsumerImpl extends ClientImpl {
Status status = Status.newBuilder().setCode(Code.INTERNAL_SERVER_ERROR)
.setMessage("status was not set by server")
.build();
- Timestamp deliveryTimestampFromRemote = null;
+ Long transportDeliveryTimestamp = null;
List<Message> messageList = new ArrayList<>();
for (ReceiveMessageResponse response : responses) {
switch (response.getContentCase()) {
@@ -99,15 +101,16 @@ abstract class ConsumerImpl extends ClientImpl {
messageList.add(response.getMessage());
break;
case DELIVERY_TIMESTAMP:
- deliveryTimestampFromRemote = response.getDeliveryTimestamp();
+ final Timestamp deliveryTimestamp = response.getDeliveryTimestamp();
+ transportDeliveryTimestamp = Timestamps.toMillis(deliveryTimestamp);
break;
default:
LOGGER.warn("[Bug] Not recognized content for receive message response, mq={}, " +
- "clientId={}, resp={}", mq, clientId, response);
+ "clientId={}, response={}", mq, clientId, response);
}
}
for (Message message : messageList) {
- final MessageViewImpl view = MessageViewImpl.fromProtobuf(message, mq, deliveryTimestampFromRemote);
+ final MessageViewImpl view = MessageViewImpl.fromProtobuf(message, mq, transportDeliveryTimestamp);
messages.add(view);
}
StatusChecker.check(status, future);
@@ -142,10 +145,9 @@ abstract class ConsumerImpl extends ClientImpl {
protected RpcFuture<AckMessageRequest, AckMessageResponse> ackMessage(MessageViewImpl messageView) {
final Endpoints endpoints = messageView.getEndpoints();
RpcFuture<AckMessageRequest, AckMessageResponse> future;
-
- final Stopwatch stopwatch = Stopwatch.createStarted();
- final List<MessageCommon> messageCommons = Collections.singletonList(messageView.getMessageCommon());
- doBefore(MessageHookPoints.ACK, messageCommons);
+ final List<GeneralMessage> generalMessages = Collections.singletonList(new GeneralMessageImpl(messageView));
+ final MessageHandlerContextImpl context = new MessageHandlerContextImpl(MessageHookPoints.ACK);
+ doBefore(context, generalMessages);
try {
final AckMessageRequest request = wrapAckMessageRequest(messageView);
final Metadata metadata = sign();
@@ -159,16 +161,17 @@ abstract class ConsumerImpl extends ClientImpl {
public void onSuccess(AckMessageResponse response) {
final Status status = response.getStatus();
final Code code = status.getCode();
- final Duration duration = stopwatch.elapsed();
MessageHookPointsStatus messageHookPointsStatus = Code.OK.equals(code) ?
MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
- doAfter(MessageHookPoints.ACK, messageCommons, duration, messageHookPointsStatus);
+ MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context, messageHookPointsStatus);
+ doAfter(context0, generalMessages);
}
@Override
public void onFailure(Throwable t) {
- final Duration duration = stopwatch.elapsed();
- doAfter(MessageHookPoints.ACK, messageCommons, duration, MessageHookPointsStatus.ERROR);
+ MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context,
+ MessageHookPointsStatus.ERROR);
+ doAfter(context0, generalMessages);
}
}, MoreExecutors.directExecutor());
return future;
@@ -178,10 +181,10 @@ abstract class ConsumerImpl extends ClientImpl {
MessageViewImpl messageView, Duration invisibleDuration) {
final Endpoints endpoints = messageView.getEndpoints();
RpcFuture<ChangeInvisibleDurationRequest, ChangeInvisibleDurationResponse> future;
-
- final Stopwatch stopwatch = Stopwatch.createStarted();
- final List<MessageCommon> messageCommons = Collections.singletonList(messageView.getMessageCommon());
- doBefore(MessageHookPoints.CHANGE_INVISIBLE_DURATION, messageCommons);
+ final List<GeneralMessage> generalMessages = Collections.singletonList(new GeneralMessageImpl(messageView));
+ final MessageHandlerContextImpl context =
+ new MessageHandlerContextImpl(MessageHookPoints.CHANGE_INVISIBLE_DURATION);
+ doBefore(context, generalMessages);
try {
final Metadata metadata = sign();
final ChangeInvisibleDurationRequest request = wrapChangeInvisibleDuration(messageView, invisibleDuration);
@@ -196,22 +199,23 @@ abstract class ConsumerImpl extends ClientImpl {
public void onSuccess(ChangeInvisibleDurationResponse response) {
final Status status = response.getStatus();
final Code code = status.getCode();
- final Duration duration = stopwatch.elapsed();
MessageHookPointsStatus messageHookPointsStatus = Code.OK.equals(code) ?
MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
if (!Code.OK.equals(code)) {
LOGGER.error("Failed to change message invisible duration, messageId={}, endpoints={}, code={}, " +
"status message=[{}], clientId={}", messageId, endpoints, code, status.getMessage(), clientId);
}
- doAfter(MessageHookPoints.CHANGE_INVISIBLE_DURATION, messageCommons, duration, messageHookPointsStatus);
+ MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context, messageHookPointsStatus);
+ doAfter(context0, generalMessages);
}
@Override
public void onFailure(Throwable t) {
- final Duration duration = stopwatch.elapsed();
- doAfter(MessageHookPoints.CHANGE_INVISIBLE_DURATION, messageCommons, duration,
- MessageHookPointsStatus.ERROR);
- LOGGER.error("Failed to change message invisible duration, messageId={}, endpoints={}, clientId={}",
+ MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context,
+ MessageHookPointsStatus.ERROR);
+ doAfter(context0, generalMessages);
+ LOGGER.error("Exception raised while changing message invisible duration, messageId={}, endpoints={}, "
+ + "clientId={}",
messageId, endpoints, clientId, t);
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
index 6dd8c9a..1b9bfba 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
@@ -29,7 +29,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
import org.slf4j.Logger;
@@ -39,9 +39,9 @@ class FifoConsumeService extends ConsumeService {
private static final Logger LOGGER = LoggerFactory.getLogger(FifoConsumeService.class);
public FifoConsumeService(String clientId, ConcurrentMap<MessageQueueImpl, ProcessQueue> processQueueTable,
- MessageListener messageListener, ThreadPoolExecutor consumptionExecutor, MessageInterceptor messageInterceptor,
+ MessageListener messageListener, ThreadPoolExecutor consumptionExecutor, MessageHandler messageHandler,
ScheduledExecutorService scheduler) {
- super(clientId, processQueueTable, messageListener, consumptionExecutor, messageInterceptor, scheduler);
+ super(clientId, processQueueTable, messageListener, consumptionExecutor, messageHandler, scheduler);
}
@Override
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/MessageCacheObserverImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/MessageCacheObserverImpl.java
new file mode 100644
index 0000000..b5802da
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/MessageCacheObserverImpl.java
@@ -0,0 +1,57 @@
+/*
+ * 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.rocketmq.client.java.impl.consumer;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.rocketmq.client.java.metrics.MessageCacheObserver;
+import org.apache.rocketmq.client.java.route.MessageQueueImpl;
+
+public class MessageCacheObserverImpl implements MessageCacheObserver {
+ private final ConcurrentMap<MessageQueueImpl, ProcessQueue> processQueueTable;
+
+ public MessageCacheObserverImpl(ConcurrentMap<MessageQueueImpl, ProcessQueue> processQueueTable) {
+ this.processQueueTable = processQueueTable;
+ }
+
+ @Override
+ public Map<String, Long> getCachedMessageCount() {
+ Map<String, Long> cachedMessageCountMap = new HashMap<>();
+ for (ProcessQueue pq : processQueueTable.values()) {
+ final String topic = pq.getMessageQueue().getTopic();
+ long count = cachedMessageCountMap.containsKey(topic) ? cachedMessageCountMap.get(topic) : 0;
+ count += pq.getInflightMessageCount();
+ count += pq.getPendingMessageCount();
+ cachedMessageCountMap.put(topic, count);
+ }
+ return cachedMessageCountMap;
+ }
+
+ @Override
+ public Map<String, Long> getCachedMessageBytes() {
+ Map<String, Long> cachedMessageBytesMap = new HashMap<>();
+ for (ProcessQueue pq : processQueueTable.values()) {
+ final String topic = pq.getMessageQueue().getTopic();
+ long bytes = cachedMessageBytesMap.containsKey(topic) ? cachedMessageBytesMap.get(topic) : 0;
+ bytes += pq.getCachedMessageBytes();
+ cachedMessageBytesMap.put(topic, bytes);
+ }
+ return cachedMessageBytesMap;
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java
index 9c106a2..b0b2ea3 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java
@@ -26,7 +26,6 @@ import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueRequest;
import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueResponse;
import apache.rocketmq.v2.ReceiveMessageRequest;
import apache.rocketmq.v2.Status;
-import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@@ -44,15 +43,19 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.message.MessageId;
+import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.client.java.exception.BadRequestException;
import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContextImpl;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+import org.apache.rocketmq.client.java.message.GeneralMessageImpl;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.retry.RetryPolicy;
import org.apache.rocketmq.client.java.route.Endpoints;
@@ -253,19 +256,24 @@ class ProcessQueueImpl implements ProcessQueue {
final int batchSize = this.getReceptionBatchSize();
final ReceiveMessageRequest request = consumer.wrapReceiveMessageRequest(batchSize, mq, filterExpression);
activityNanoTime = System.nanoTime();
+
// Intercept before message reception.
- consumer.doBefore(MessageHookPoints.RECEIVE, Collections.emptyList());
- final Stopwatch stopwatch = Stopwatch.createStarted();
+ final MessageHandlerContextImpl context = new MessageHandlerContextImpl(MessageHookPoints.RECEIVE);
+ consumer.doBefore(context, Collections.emptyList());
+
final ListenableFuture<ReceiveMessageResult> future = consumer.receiveMessage(request, mq,
consumer.getPushConsumerSettings().getLongPollingTimeout());
Futures.addCallback(future, new FutureCallback<ReceiveMessageResult>() {
@Override
public void onSuccess(ReceiveMessageResult result) {
// Intercept after message reception.
- final Duration duration = stopwatch.elapsed();
- final List<MessageCommon> commons = result.getMessageViewImpls().stream()
- .map(MessageViewImpl::getMessageCommon).collect(Collectors.toList());
- consumer.doAfter(MessageHookPoints.RECEIVE, commons, duration, MessageHookPointsStatus.OK);
+ final List<GeneralMessage> generalMessages = result.getMessageViewImpls().stream()
+ .map((Function<MessageView, GeneralMessage>) GeneralMessageImpl::new)
+ .collect(Collectors.toList());
+ final MessageHandlerContextImpl context0 =
+ new MessageHandlerContextImpl(context, MessageHookPointsStatus.OK);
+ consumer.doAfter(context0, generalMessages);
+
try {
onReceiveMessageResult(result);
} catch (Throwable t) {
@@ -279,9 +287,10 @@ class ProcessQueueImpl implements ProcessQueue {
@Override
public void onFailure(Throwable t) {
// Intercept after message reception.
- final Duration duration = stopwatch.elapsed();
- consumer.doAfter(MessageHookPoints.RECEIVE, Collections.emptyList(), duration,
- MessageHookPointsStatus.ERROR);
+ final MessageHandlerContextImpl context0 =
+ new MessageHandlerContextImpl(context, MessageHookPointsStatus.ERROR);
+ consumer.doAfter(context0, Collections.emptyList());
+
LOGGER.error("Exception raised during message reception, mq={}, endpoints={}, clientId={}", mq,
endpoints, consumer.clientId(), t);
onReceiveMessageException(t);
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
index b289aa7..d4c398d 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
@@ -28,7 +28,6 @@ import apache.rocketmq.v2.TelemetryCommand;
import apache.rocketmq.v2.VerifyMessageCommand;
import apache.rocketmq.v2.VerifyMessageResult;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@@ -59,10 +58,13 @@ import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.java.exception.StatusChecker;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContext;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContextImpl;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.impl.Settings;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+import org.apache.rocketmq.client.java.message.GeneralMessageImpl;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.message.protocol.Resource;
import org.apache.rocketmq.client.java.metrics.MessageCacheObserver;
@@ -85,7 +87,7 @@ import org.slf4j.LoggerFactory;
* @see PushConsumer
*/
@SuppressWarnings({"UnstableApiUsage", "NullableProblems"})
-class PushConsumerImpl extends ConsumerImpl implements PushConsumer, MessageCacheObserver {
+class PushConsumerImpl extends ConsumerImpl implements PushConsumer {
private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerImpl.class);
final AtomicLong consumptionOkQuantity;
@@ -140,6 +142,9 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer, MessageCach
this.consumptionErrorQuantity = new AtomicLong(0);
this.processQueueTable = new ConcurrentHashMap<>();
+ MessageCacheObserver messageCacheObserver = new MessageCacheObserverImpl(processQueueTable);
+ this.clientMeterProvider.setMessageCacheObserver(messageCacheObserver);
+
this.consumptionExecutor = new ThreadPoolExecutor(
consumptionThreadCount,
consumptionThreadCount,
@@ -154,7 +159,6 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer, MessageCach
try {
LOGGER.info("Begin to start the rocketmq push consumer, clientId={}", clientId);
super.startUp();
- clientMeterProvider.setMessageCacheObserver(this);
final ScheduledExecutorService scheduler = this.getClientManager().getScheduler();
this.consumeService = createConsumeService();
this.consumeService.startAsync().awaitRunning();
@@ -508,9 +512,9 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer, MessageCach
public RpcFuture<ForwardMessageToDeadLetterQueueRequest, ForwardMessageToDeadLetterQueueResponse>
forwardMessageToDeadLetterQueue(final MessageViewImpl messageView) {
// Intercept before forwarding message to DLQ.
- final Stopwatch stopwatch = Stopwatch.createStarted();
- final List<MessageCommon> messageCommons = Collections.singletonList(messageView.getMessageCommon());
- doBefore(MessageHookPoints.FORWARD_TO_DLQ, messageCommons);
+ final List<GeneralMessage> generalMessages = Collections.singletonList(new GeneralMessageImpl(messageView));
+ final MessageHandlerContextImpl context = new MessageHandlerContextImpl(MessageHookPoints.FORWARD_TO_DLQ);
+ doBefore(context, generalMessages);
final Endpoints endpoints = messageView.getEndpoints();
RpcFuture<ForwardMessageToDeadLetterQueueRequest, ForwardMessageToDeadLetterQueueResponse> future;
@@ -526,18 +530,19 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer, MessageCach
Futures.addCallback(future, new FutureCallback<ForwardMessageToDeadLetterQueueResponse>() {
@Override
public void onSuccess(ForwardMessageToDeadLetterQueueResponse response) {
- final Duration duration = stopwatch.elapsed();
- MessageHookPointsStatus messageHookPointsStatus = Code.OK.equals(response.getStatus().getCode()) ?
- MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
// Intercept after forwarding message to DLQ.
- doAfter(MessageHookPoints.FORWARD_TO_DLQ, messageCommons, duration, messageHookPointsStatus);
+ MessageHookPointsStatus status = Code.OK.equals(response.getStatus().getCode()) ?
+ MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
+ final MessageHandlerContext context0 = new MessageHandlerContextImpl(context, status);
+ doAfter(context0, generalMessages);
}
@Override
public void onFailure(Throwable t) {
// Intercept after forwarding message to DLQ.
- final Duration duration = stopwatch.elapsed();
- doAfter(MessageHookPoints.FORWARD_TO_DLQ, messageCommons, duration, MessageHookPointsStatus.ERROR);
+ final MessageHandlerContext context0 = new MessageHandlerContextImpl(context,
+ MessageHookPointsStatus.ERROR);
+ doAfter(context0, generalMessages);
}
}, MoreExecutors.directExecutor());
return future;
@@ -564,29 +569,4 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer, MessageCach
public ThreadPoolExecutor getConsumptionExecutor() {
return consumptionExecutor;
}
-
- @Override
- public Map<String, Long> getCachedMessageCount() {
- Map<String, Long> cachedMessageCountMap = new HashMap<>();
- for (ProcessQueue pq : processQueueTable.values()) {
- final String topic = pq.getMessageQueue().getTopic();
- long count = cachedMessageCountMap.containsKey(topic) ? cachedMessageCountMap.get(topic) : 0;
- count += pq.getInflightMessageCount();
- count += pq.getPendingMessageCount();
- cachedMessageCountMap.put(topic, count);
- }
- return cachedMessageCountMap;
- }
-
- @Override
- public Map<String, Long> getCachedMessageBytes() {
- Map<String, Long> cachedMessageBytesMap = new HashMap<>();
- for (ProcessQueue pq : processQueueTable.values()) {
- final String topic = pq.getMessageQueue().getTopic();
- long bytes = cachedMessageBytesMap.containsKey(topic) ? cachedMessageBytesMap.get(topic) : 0;
- bytes += pq.getCachedMessageBytes();
- cachedMessageBytesMap.put(topic, bytes);
- }
- return cachedMessageBytesMap;
- }
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
index 452ef8f..8ace72f 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
@@ -30,7 +30,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
import org.slf4j.Logger;
@@ -41,9 +41,9 @@ public class StandardConsumeService extends ConsumeService {
private static final Logger LOGGER = LoggerFactory.getLogger(StandardConsumeService.class);
public StandardConsumeService(String clientId, ConcurrentMap<MessageQueueImpl, ProcessQueue> processQueueTable,
- MessageListener messageListener, ThreadPoolExecutor consumptionExecutor, MessageInterceptor messageInterceptor,
+ MessageListener messageListener, ThreadPoolExecutor consumptionExecutor, MessageHandler messageHandler,
ScheduledExecutorService scheduler) {
- super(clientId, processQueueTable, messageListener, consumptionExecutor, messageInterceptor, scheduler);
+ super(clientId, processQueueTable, messageListener, consumptionExecutor, messageHandler, scheduler);
}
@Override
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerImpl.java
index e323a59..58d2c23 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerImpl.java
@@ -28,7 +28,6 @@ import apache.rocketmq.v2.RecoverOrphanedTransactionCommand;
import apache.rocketmq.v2.SendMessageRequest;
import apache.rocketmq.v2.SendMessageResponse;
import apache.rocketmq.v2.Status;
-import com.google.common.base.Stopwatch;
import com.google.common.math.IntMath;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
@@ -48,6 +47,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import java.util.stream.Collectors;
import net.javacrumbs.futureconverter.java8guava.FutureConverter;
import org.apache.rocketmq.client.apis.ClientConfiguration;
@@ -61,11 +61,13 @@ import org.apache.rocketmq.client.apis.producer.TransactionChecker;
import org.apache.rocketmq.client.apis.producer.TransactionResolution;
import org.apache.rocketmq.client.java.exception.InternalErrorException;
import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContextImpl;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.impl.ClientImpl;
import org.apache.rocketmq.client.java.impl.Settings;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+import org.apache.rocketmq.client.java.message.GeneralMessageImpl;
import org.apache.rocketmq.client.java.message.MessageType;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
@@ -159,7 +161,8 @@ class ProducerImpl extends ClientImpl implements Producer {
if (null == resolution || TransactionResolution.UNKNOWN.equals(resolution)) {
return;
}
- endTransaction(endpoints, messageView.getMessageCommon(), messageView.getMessageId(),
+ final GeneralMessage generalMessage = new GeneralMessageImpl(messageView);
+ endTransaction(endpoints, generalMessage, messageView.getMessageId(),
transactionId, resolution);
} catch (Throwable t) {
LOGGER.error("Exception raised while ending the transaction, messageId={}, transactionId={}, "
@@ -252,7 +255,7 @@ class ProducerImpl extends ClientImpl implements Producer {
this.stopAsync().awaitTerminated();
}
- public void endTransaction(Endpoints endpoints, MessageCommon messageCommon, MessageId messageId,
+ public void endTransaction(Endpoints endpoints, GeneralMessage generalMessage, MessageId messageId,
String transactionId, final TransactionResolution resolution) throws ClientException {
Metadata metadata;
try {
@@ -262,7 +265,7 @@ class ProducerImpl extends ClientImpl implements Producer {
}
final EndTransactionRequest.Builder builder = EndTransactionRequest.newBuilder()
.setMessageId(messageId.toString()).setTransactionId(transactionId)
- .setTopic(apache.rocketmq.v2.Resource.newBuilder().setName(messageCommon.getTopic()).build());
+ .setTopic(apache.rocketmq.v2.Resource.newBuilder().setName(generalMessage.getTopic()).build());
switch (resolution) {
case COMMIT:
builder.setResolution(apache.rocketmq.v2.TransactionResolution.COMMIT);
@@ -273,30 +276,31 @@ class ProducerImpl extends ClientImpl implements Producer {
}
final Duration requestTimeout = clientConfiguration.getRequestTimeout();
final EndTransactionRequest request = builder.build();
-
- final Stopwatch stopwatch = Stopwatch.createStarted();
- final List<MessageCommon> messageCommons = Collections.singletonList(messageCommon);
+ final List<GeneralMessage> generalMessages = Collections.singletonList(generalMessage);
MessageHookPoints messageHookPoints = TransactionResolution.COMMIT.equals(resolution) ?
MessageHookPoints.COMMIT_TRANSACTION : MessageHookPoints.ROLLBACK_TRANSACTION;
- doBefore(messageHookPoints, messageCommons);
+ final MessageHandlerContextImpl context = new MessageHandlerContextImpl(messageHookPoints);
+ doBefore(context, generalMessages);
final RpcFuture<EndTransactionRequest, EndTransactionResponse> future =
this.getClientManager().endTransaction(endpoints, metadata, request, requestTimeout);
Futures.addCallback(future, new FutureCallback<EndTransactionResponse>() {
@Override
public void onSuccess(EndTransactionResponse response) {
- final Duration duration = stopwatch.elapsed();
final Status status = response.getStatus();
final Code code = status.getCode();
MessageHookPointsStatus messageHookPointsStatus = Code.OK.equals(code) ? MessageHookPointsStatus.OK :
MessageHookPointsStatus.ERROR;
- doAfter(messageHookPoints, messageCommons, duration, messageHookPointsStatus);
+ final MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context,
+ messageHookPointsStatus);
+ doAfter(context0, generalMessages);
}
@Override
public void onFailure(Throwable t) {
- final Duration duration = stopwatch.elapsed();
- doAfter(messageHookPoints, messageCommons, duration, MessageHookPointsStatus.ERROR);
+ final MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context,
+ MessageHookPointsStatus.ERROR);
+ doAfter(context0, generalMessages);
}
}, MoreExecutors.directExecutor());
final EndTransactionResponse response = handleClientFuture(future);
@@ -454,26 +458,35 @@ class ProducerImpl extends ClientImpl implements Producer {
final Endpoints endpoints = mq.getBroker().getEndpoints();
final ListenableFuture<List<SendReceiptImpl>> future = send0(metadata, endpoints, messages, mq);
final int maxAttempts = this.getRetryPolicy().getMaxAttempts();
+
// Intercept before message publishing.
- final Stopwatch stopwatch = Stopwatch.createStarted();
- final List<MessageCommon> messageCommons =
- messages.stream().map(PublishingMessageImpl::getMessageCommon).collect(Collectors.toList());
- doBefore(MessageHookPoints.SEND, messageCommons);
+ final List<GeneralMessage> generalMessages = messages.stream().map((Function<PublishingMessageImpl,
+ GeneralMessage>) GeneralMessageImpl::new).collect(Collectors.toList());
+ final MessageHandlerContextImpl context = new MessageHandlerContextImpl(MessageHookPoints.SEND);
+ doBefore(context, generalMessages);
Futures.addCallback(future, new FutureCallback<List<SendReceiptImpl>>() {
@Override
public void onSuccess(List<SendReceiptImpl> sendReceipts) {
- // Intercept after message publishing.
- final Duration duration = stopwatch.elapsed();
- doAfter(MessageHookPoints.SEND, messageCommons, duration, MessageHookPointsStatus.OK);
// Should never reach here.
if (sendReceipts.size() != messages.size()) {
final InternalErrorException e = new InternalErrorException("[Bug] due to an"
+ " unknown reason from remote, received send receipt's quantity " + sendReceipts.size()
+ " is not equal to sent message's quantity " + messages.size());
future0.setException(e);
+
+ // Intercept after message publishing.
+ final MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context,
+ MessageHookPointsStatus.ERROR);
+ doAfter(context0, generalMessages);
+
return;
}
+ // Intercept after message publishing.
+ final MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context,
+ MessageHookPointsStatus.OK);
+ doAfter(context0, generalMessages);
+
// No need more attempts.
future0.set(sendReceipts);
// Resend message(s) successfully.
@@ -493,8 +506,9 @@ class ProducerImpl extends ClientImpl implements Producer {
@Override
public void onFailure(Throwable t) {
// Intercept after message publishing.
- final Duration duration = stopwatch.elapsed();
- doAfter(MessageHookPoints.SEND, messageCommons, duration, MessageHookPointsStatus.ERROR);
+ final MessageHandlerContextImpl context0 = new MessageHandlerContextImpl(context,
+ MessageHookPointsStatus.ERROR);
+ doAfter(context0, generalMessages);
// Collect messageId(s) for logging.
List<MessageId> messageIds = new ArrayList<>();
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/TransactionImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/TransactionImpl.java
index 986e8b4..51ca92c 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/TransactionImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/TransactionImpl.java
@@ -30,6 +30,7 @@ import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Transaction;
import org.apache.rocketmq.client.apis.producer.TransactionResolution;
+import org.apache.rocketmq.client.java.message.GeneralMessageImpl;
import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
class TransactionImpl implements Transaction {
@@ -93,7 +94,7 @@ class TransactionImpl implements Transaction {
for (Map.Entry<PublishingMessageImpl, SendReceiptImpl> entry : messageSendReceiptMap.entrySet()) {
final PublishingMessageImpl publishingMessage = entry.getKey();
final SendReceiptImpl sendReceipt = entry.getValue();
- producerImpl.endTransaction(sendReceipt.getEndpoints(), publishingMessage.getMessageCommon(),
+ producerImpl.endTransaction(sendReceipt.getEndpoints(), new GeneralMessageImpl(publishingMessage),
sendReceipt.getMessageId(), sendReceipt.getTransactionId(), TransactionResolution.COMMIT);
}
}
@@ -106,7 +107,7 @@ class TransactionImpl implements Transaction {
for (Map.Entry<PublishingMessageImpl, SendReceiptImpl> entry : messageSendReceiptMap.entrySet()) {
final PublishingMessageImpl publishingMessage = entry.getKey();
final SendReceiptImpl sendReceipt = entry.getValue();
- producerImpl.endTransaction(sendReceipt.getEndpoints(), publishingMessage.getMessageCommon(),
+ producerImpl.endTransaction(sendReceipt.getEndpoints(), new GeneralMessageImpl(publishingMessage),
sendReceipt.getMessageId(), sendReceipt.getTransactionId(), TransactionResolution.ROLLBACK);
}
}
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessage.java
similarity index 62%
copy from java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java
copy to java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessage.java
index 04d576b..575235d 100644
--- a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessage.java
@@ -15,24 +15,26 @@
* limitations under the License.
*/
-package org.apache.rocketmq.client.apis.message;
+package org.apache.rocketmq.client.java.message;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
+import org.apache.rocketmq.client.apis.message.Message;
+import org.apache.rocketmq.client.apis.message.MessageId;
+import org.apache.rocketmq.client.apis.message.MessageView;
/**
- * {@link MessageView} provides a read-only view for the message, that's why setters do not exist here. In addition,
- * it only makes sense when {@link Message} is sent successfully.
+ * General message for RocketMQ, which combines {@link Message} and {@link MessageView}.
*/
-public interface MessageView {
+public interface GeneralMessage {
/**
* Get the unique id of the message.
*
* @return unique id.
*/
- MessageId getMessageId();
+ Optional<MessageId> getMessageId();
/**
* Get the topic of the message, which is the first classifier for the message.
@@ -80,7 +82,23 @@ public interface MessageView {
Optional<String> getMessageGroup();
/**
- * Get the expected delivery timestamp, which makes sense only when the topic type is delay.
+ * Get the parent trace context, see
+ * <a href="https://opentelemetry.io/docs/concepts/signals/traces/#trace-context">OpenTelemetry Trace context</a>.
+ *
+ * @return parent trace context, which is optional, {@link Optional#empty()} means trace context is not specified.
+ */
+ Optional<String> getParentTraceContext();
+
+ /**
+ * Message trace context, see
+ * <a href="https://opentelemetry.io/docs/concepts/signals/traces/#trace-context">OpenTelemetry Trace context</a>.
+ *
+ * @return message trace context.
+ */
+ Optional<String> getTraceContext();
+
+ /**
+ * Get the expected delivery timestamp, which makes sense only when topic type is delay.
*
* @return message expected delivery timestamp, which is optional, {@link Optional#empty()} means delivery
* timestamp is not specified.
@@ -90,11 +108,9 @@ public interface MessageView {
/**
* Get the born host of the message.
*
- * @return born host of the message.
+ * @return born host of the message, which is optional, {@link Optional#empty()} means born host is not specified.
*/
- String getBornHost();
-
- Optional<String> getTraceContext();
+ Optional<String> getBornHost();
/**
* Get the born timestamp of the message.
@@ -102,14 +118,31 @@ public interface MessageView {
* <p>Born time means the timestamp that the message is prepared to send rather than the timestamp the
* {@link Message} was built.
*
- * @return born timestamp of the message.
+ * @return born timestamp of the message, which is optional, {@link Optional#empty()} means born timestamp is not
+ * specified.
*/
- long getBornTimestamp();
+ Optional<Long> getBornTimestamp();
/**
* Get the delivery attempt for the message.
*
* @return delivery attempt.
*/
- int getDeliveryAttempt();
+ Optional<Integer> getDeliveryAttempt();
+
+ /**
+ * Get decode timestamp, which makes sense only if the message is delivered from remote.
+ *
+ * @return message decode timestamp, which is optional, {@link Optional#empty()} means decode timestamp is not
+ * specified.
+ */
+ Optional<Long> getDecodeTimestamp();
+
+ /**
+ * Get transport delivery timestamp, which makes sense only if the message is delivered from remote.
+ *
+ * @return message transport delivery timestamp, which is optional, {@link Optional#empty()} means transport
+ * delivery timestamp is not specified.
+ */
+ Optional<Long> getTransportDeliveryTimestamp();
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessageImpl.java
new file mode 100644
index 0000000..4d5da19
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessageImpl.java
@@ -0,0 +1,187 @@
+/*
+ * 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.rocketmq.client.java.message;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.rocketmq.client.apis.message.Message;
+import org.apache.rocketmq.client.apis.message.MessageId;
+import org.apache.rocketmq.client.apis.message.MessageView;
+
+public class GeneralMessageImpl implements GeneralMessage {
+ private final String topic;
+ private final MessageId messageId;
+ private final byte[] body;
+ private final Map<String, String> properties;
+ private final String tag;
+ private final Collection<String> keys;
+ private final String messageGroup;
+ private final Long deliveryTimestamp;
+ private final String bornHost;
+ private final String parentTraceContext;
+ private final String traceContext;
+ private final Long bornTimestamp;
+ private final Integer deliveryAttempt;
+ private final Long decodeTimestamp;
+ private final Long transportDeliveryTimestamp;
+
+ public GeneralMessageImpl(Message message) {
+ this.topic = message.getTopic();
+ this.messageId = null;
+ if (message instanceof MessageImpl) {
+ MessageImpl impl = (MessageImpl) message;
+ this.body = impl.body;
+ } else {
+ // Should never reach here.
+ final ByteBuffer byteBuffer = message.getBody();
+ this.body = new byte[byteBuffer.remaining()];
+ byteBuffer.get(body);
+ }
+ this.properties = message.getProperties();
+ this.tag = message.getTag().orElse(null);
+ this.keys = message.getKeys();
+ this.messageGroup = message.getMessageGroup().orElse(null);
+ this.deliveryTimestamp = message.getDeliveryTimestamp().orElse(null);
+ this.bornHost = null;
+ this.parentTraceContext = message.getParentTraceContext().orElse(null);
+ this.traceContext = null;
+ this.bornTimestamp = null;
+ this.deliveryAttempt = null;
+ this.decodeTimestamp = null;
+ this.transportDeliveryTimestamp = null;
+ }
+
+ public GeneralMessageImpl(MessageView message) {
+ this.topic = message.getTopic();
+ this.messageId = message.getMessageId();
+ byte[] messageBody;
+ String messageTraceContext;
+ Long messageDecodeTimestamp;
+ Long messageTransportDeliveryTimestamp;
+ if (message instanceof MessageViewImpl) {
+ MessageViewImpl impl = (MessageViewImpl) message;
+ messageBody = impl.body;
+ messageTraceContext = impl.getTraceContext().orElse(null);
+ messageDecodeTimestamp = impl.getDecodeTimestamp();
+ messageTransportDeliveryTimestamp = impl.getTransportDeliveryTimestamp().orElse(null);
+ } else {
+ // Should never reach here.
+ final ByteBuffer byteBuffer = message.getBody();
+ messageBody = new byte[byteBuffer.remaining()];
+ byteBuffer.get(messageBody);
+ messageTraceContext = null;
+ // Could not get accurate decode timestamp.
+ messageDecodeTimestamp = null;
+ // Could not get accurate transport delivery timestamp.
+ messageTransportDeliveryTimestamp = null;
+ }
+ this.body = messageBody;
+ this.properties = message.getProperties();
+ this.tag = message.getTag().orElse(null);
+ this.keys = message.getKeys();
+ this.messageGroup = message.getMessageGroup().orElse(null);
+ this.deliveryTimestamp = message.getDeliveryTimestamp().orElse(null);
+ this.bornHost = message.getBornHost();
+ this.parentTraceContext = null;
+ this.traceContext = messageTraceContext;
+ this.bornTimestamp = message.getBornTimestamp();
+ this.deliveryAttempt = message.getDeliveryAttempt();
+ this.decodeTimestamp = messageDecodeTimestamp;
+ this.transportDeliveryTimestamp = messageTransportDeliveryTimestamp;
+ }
+
+
+ @Override
+ public Optional<MessageId> getMessageId() {
+ return Optional.ofNullable(messageId);
+ }
+
+ @Override
+ public String getTopic() {
+ return topic;
+ }
+
+ @Override
+ public ByteBuffer getBody() {
+ return ByteBuffer.wrap(body).asReadOnlyBuffer();
+ }
+
+ @Override
+ public Map<String, String> getProperties() {
+ return new HashMap<>(properties);
+ }
+
+ @Override
+ public Optional<String> getTag() {
+ return Optional.ofNullable(tag);
+ }
+
+ @Override
+ public Collection<String> getKeys() {
+ return new ArrayList<>(keys);
+ }
+
+ @Override
+ public Optional<String> getMessageGroup() {
+ return Optional.ofNullable(messageGroup);
+ }
+
+ @Override
+ public Optional<String> getParentTraceContext() {
+ return Optional.ofNullable(parentTraceContext);
+ }
+
+ @Override
+ public Optional<String> getTraceContext() {
+ return Optional.ofNullable(traceContext);
+ }
+
+ @Override
+ public Optional<Long> getDeliveryTimestamp() {
+ return Optional.ofNullable(deliveryTimestamp);
+ }
+
+ @Override
+ public Optional<String> getBornHost() {
+ return Optional.ofNullable(bornHost);
+ }
+
+ @Override
+ public Optional<Long> getBornTimestamp() {
+ return Optional.ofNullable(bornTimestamp);
+ }
+
+ @Override
+ public Optional<Integer> getDeliveryAttempt() {
+ return Optional.ofNullable(deliveryAttempt);
+ }
+
+ @Override
+ public Optional<Long> getDecodeTimestamp() {
+ return Optional.ofNullable(decodeTimestamp);
+ }
+
+ @Override
+ public Optional<Long> getTransportDeliveryTimestamp() {
+ return Optional.ofNullable(transportDeliveryTimestamp);
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageCommon.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageCommon.java
deleted file mode 100644
index a700436..0000000
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageCommon.java
+++ /dev/null
@@ -1,160 +0,0 @@
-/*
- * 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.rocketmq.client.java.message;
-
-import com.google.common.base.Stopwatch;
-import com.google.protobuf.Timestamp;
-import java.nio.ByteBuffer;
-import java.time.Duration;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import javax.annotation.Nullable;
-import org.apache.rocketmq.client.apis.message.Message;
-import org.apache.rocketmq.client.apis.message.MessageId;
-import org.apache.rocketmq.client.apis.message.MessageView;
-
-/**
- * Abstract common structure of {@link Message} and {@link MessageView}.
- */
-public class MessageCommon {
- @Nullable
- private final MessageId messageId;
- private final String topic;
- private final byte[] body;
- private final Map<String, String> properties;
- @Nullable
- private final String tag;
- private final Collection<String> keys;
- @Nullable
- private final String messageGroup;
- @Nullable
- private final Long deliveryTimestamp;
- @Nullable
- private final String bornHost;
- @Nullable
- private final String parentTraceContext;
- @Nullable
- private final String traceContext;
- @Nullable
- private final Long bornTimestamp;
- @Nullable
- private final Integer deliveryAttempt;
- @Nullable
- private final Stopwatch decodeStopwatch;
- @Nullable
- private final Timestamp deliveryTimestampFromRemote;
-
- public MessageCommon(String topic, byte[] body, String tag, String messageGroup, Long deliveryTimestamp,
- String parentTraceContext, Collection<String> keys, Map<String, String> properties) {
- this(null, topic, body, properties, tag, keys, messageGroup, deliveryTimestamp, null, parentTraceContext,
- null, null, null, null, null);
- }
-
- public MessageCommon(MessageId messageId, String topic, byte[] body, String tag, String messageGroup,
- Long deliveryTimestamp, Collection<String> keys, Map<String, String> properties, String bornHost,
- String traceContext, long bornTimestamp, int deliveryAttempt, Stopwatch decodeStopwatch,
- Timestamp deliveryTimestampFromRemote) {
- this(messageId, topic, body, properties, tag, keys, messageGroup, deliveryTimestamp, bornHost, null,
- traceContext, bornTimestamp, deliveryAttempt, decodeStopwatch, deliveryTimestampFromRemote);
- }
-
- private MessageCommon(@Nullable MessageId messageId, String topic, byte[] body,
- Map<String, String> properties, @Nullable String tag, Collection<String> keys,
- @Nullable String messageGroup, @Nullable Long deliveryTimestamp, @Nullable String bornHost,
- @Nullable String parentTraceContext, @Nullable String traceContext, @Nullable Long bornTimestamp,
- @Nullable Integer deliveryAttempt, @Nullable Stopwatch decodeStopwatch,
- @Nullable Timestamp deliveryTimestampFromRemote) {
- this.messageId = messageId;
- this.topic = topic;
- this.body = body;
- this.properties = properties;
- this.tag = tag;
- this.keys = keys;
- this.messageGroup = messageGroup;
- this.deliveryTimestamp = deliveryTimestamp;
- this.bornHost = bornHost;
- this.parentTraceContext = parentTraceContext;
- this.traceContext = traceContext;
- this.bornTimestamp = bornTimestamp;
- this.deliveryAttempt = deliveryAttempt;
- this.decodeStopwatch = decodeStopwatch;
- this.deliveryTimestampFromRemote = deliveryTimestampFromRemote;
- }
-
- public Optional<MessageId> getMessageId() {
- return null == messageId ? Optional.empty() : Optional.of(messageId);
- }
-
- public String getTopic() {
- return topic;
- }
-
- public ByteBuffer getBody() {
- return ByteBuffer.wrap(body).asReadOnlyBuffer();
- }
-
- public Map<String, String> getProperties() {
- return new HashMap<>(properties);
- }
-
- public Optional<String> getTag() {
- return null == tag ? Optional.empty() : Optional.of(tag);
- }
-
- public Collection<String> getKeys() {
- return keys;
- }
-
- public Optional<String> getMessageGroup() {
- return null == messageGroup ? Optional.empty() : Optional.of(messageGroup);
- }
-
- public Optional<Long> getDeliveryTimestamp() {
- return null == deliveryTimestamp ? Optional.empty() : Optional.of(deliveryTimestamp);
- }
-
- public Optional<String> getBornHost() {
- return null == bornHost ? Optional.empty() : Optional.of(bornHost);
- }
-
- public Optional<String> getParentTraceContext() {
- return null == parentTraceContext ? Optional.empty() : Optional.of(parentTraceContext);
- }
-
- public Optional<String> getTraceContext() {
- return null == traceContext ? Optional.empty() : Optional.of(traceContext);
- }
-
- public Optional<Long> getBornTimestamp() {
- return null == bornTimestamp ? Optional.empty() : Optional.of(bornTimestamp);
- }
-
- public Optional<Integer> getDeliveryAttempt() {
- return null == deliveryAttempt ? Optional.empty() : Optional.of(deliveryAttempt);
- }
-
- public Optional<Duration> getDurationAfterDecoding() {
- return null == decodeStopwatch ? Optional.empty() : Optional.of(decodeStopwatch.elapsed());
- }
-
- public Optional<Timestamp> getDeliveryTimestampFromRemote() {
- return null == deliveryTimestampFromRemote ? Optional.empty() : Optional.of(deliveryTimestampFromRemote);
- }
-}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageImpl.java
index 093dbad..c4f0db5 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageImpl.java
@@ -86,11 +86,6 @@ public class MessageImpl implements Message {
this.properties = message.getProperties();
}
- public MessageCommon getMessageCommon() {
- return new MessageCommon(topic, body, tag, messageGroup, deliveryTimestamp, parentTraceContext, keys,
- properties);
- }
-
/**
* @see Message#getTopic()
*/
@@ -120,7 +115,7 @@ public class MessageImpl implements Message {
*/
@Override
public Optional<String> getTag() {
- return null == tag ? Optional.empty() : Optional.of(tag);
+ return Optional.ofNullable(tag);
}
/**
@@ -136,7 +131,7 @@ public class MessageImpl implements Message {
*/
@Override
public Optional<Long> getDeliveryTimestamp() {
- return null == deliveryTimestamp ? Optional.empty() : Optional.of(deliveryTimestamp);
+ return Optional.ofNullable(deliveryTimestamp);
}
/**
@@ -144,7 +139,7 @@ public class MessageImpl implements Message {
*/
@Override
public Optional<String> getMessageGroup() {
- return null == messageGroup ? Optional.empty() : Optional.of(messageGroup);
+ return Optional.ofNullable(messageGroup);
}
/**
@@ -152,7 +147,7 @@ public class MessageImpl implements Message {
*/
@Override
public Optional<String> getParentTraceContext() {
- return null == parentTraceContext ? Optional.empty() : Optional.of(parentTraceContext);
+ return Optional.ofNullable(parentTraceContext);
}
@Override
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageViewImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageViewImpl.java
index 2215c61..6ac264a 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageViewImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageViewImpl.java
@@ -25,13 +25,12 @@ import apache.rocketmq.v2.Encoding;
import apache.rocketmq.v2.Message;
import apache.rocketmq.v2.SystemProperties;
import com.google.common.base.MoreObjects;
-import com.google.common.base.Stopwatch;
import com.google.protobuf.ProtocolStringList;
-import com.google.protobuf.Timestamp;
import com.google.protobuf.util.Timestamps;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
@@ -50,9 +49,9 @@ import org.slf4j.LoggerFactory;
public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageView {
private static final Logger LOGGER = LoggerFactory.getLogger(MessageViewImpl.class);
+ final byte[] body;
private final MessageId messageId;
private final String topic;
- private final byte[] body;
private final String tag;
private final String messageGroup;
private final Long deliveryTimestamp;
@@ -67,15 +66,15 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
private final String traceContext;
private final long offset;
private final boolean corrupted;
- private final Stopwatch decodeStopwatch;
- private final Timestamp deliveryTimestampFromRemote;
+ private final long decodeTimestamp;
+ private final Long transportDeliveryTimestamp;
private MessageViewImpl next;
public MessageViewImpl(MessageId messageId, String topic, byte[] body, String tag, String messageGroup,
Long deliveryTimestamp, Collection<String> keys, Map<String, String> properties,
String bornHost, long bornTimestamp, int deliveryAttempt, MessageQueueImpl messageQueue,
String receiptHandle, String traceContext, long offset, boolean corrupted,
- Timestamp deliveryTimestampFromRemote) {
+ Long transportDeliveryTimestamp) {
this.messageId = checkNotNull(messageId, "messageId should not be null");
this.topic = checkNotNull(topic, "topic should not be null");
this.body = checkNotNull(body, "body should not be null");
@@ -93,16 +92,11 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
this.traceContext = traceContext;
this.offset = offset;
this.corrupted = corrupted;
- this.decodeStopwatch = Stopwatch.createStarted();
- this.deliveryTimestampFromRemote = deliveryTimestampFromRemote;
+ this.decodeTimestamp = System.currentTimeMillis();
+ this.transportDeliveryTimestamp = transportDeliveryTimestamp;
this.next = null;
}
- public MessageCommon getMessageCommon() {
- return new MessageCommon(messageId, topic, body, tag, messageGroup, deliveryTimestamp, keys, properties,
- bornHost, traceContext, bornTimestamp, deliveryAttempt, decodeStopwatch, deliveryTimestampFromRemote);
- }
-
/**
* @see MessageView#getMessageId()
*/
@@ -140,7 +134,7 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
*/
@Override
public Optional<String> getTag() {
- return null == tag ? Optional.empty() : Optional.of(tag);
+ return Optional.ofNullable(tag);
}
/**
@@ -148,7 +142,7 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
*/
@Override
public Collection<String> getKeys() {
- return keys;
+ return new ArrayList<>(keys);
}
/**
@@ -156,7 +150,7 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
*/
@Override
public Optional<String> getMessageGroup() {
- return null == messageGroup ? Optional.empty() : Optional.of(messageGroup);
+ return Optional.ofNullable(messageGroup);
}
/**
@@ -164,7 +158,7 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
*/
@Override
public Optional<Long> getDeliveryTimestamp() {
- return null == deliveryTimestamp ? Optional.empty() : Optional.of(deliveryTimestamp);
+ return Optional.ofNullable(deliveryTimestamp);
}
/**
@@ -192,7 +186,7 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
}
public Optional<String> getTraceContext() {
- return null == traceContext ? Optional.empty() : Optional.of(traceContext);
+ return Optional.ofNullable(traceContext);
}
public int incrementAndGetDeliveryAttempt() {
@@ -225,6 +219,14 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
return corrupted;
}
+ public long getDecodeTimestamp() {
+ return decodeTimestamp;
+ }
+
+ public Optional<Long> getTransportDeliveryTimestamp() {
+ return Optional.ofNullable(transportDeliveryTimestamp);
+ }
+
public void setNext(MessageViewImpl messageView) {
this.next = messageView;
}
@@ -248,7 +250,7 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
}
public static MessageViewImpl fromProtobuf(Message message, MessageQueueImpl mq,
- Timestamp deliveryTimestampFromRemote) {
+ Long transportDeliveryTimestamp) {
final SystemProperties systemProperties = message.getSystemProperties();
final String topic = message.getTopic().getName();
final MessageId messageId = MessageIdCodec.getInstance().decode(systemProperties.getMessageId());
@@ -324,7 +326,7 @@ public class MessageViewImpl implements LinkedElement<MessageViewImpl>, MessageV
String traceContext = systemProperties.hasTraceContext() ? systemProperties.getTraceContext() : null;
return new MessageViewImpl(messageId, topic, body, tag, messageGroup, deliveryTimestamp, keys, properties,
bornHost, bornTimestamp, deliveryAttempt, mq, receiptHandle, traceContext, offset, corrupted,
- deliveryTimestampFromRemote);
+ transportDeliveryTimestamp);
}
@Override
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java
index 306ff5c..c1e2a85 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java
@@ -89,7 +89,7 @@ public class PublishingMessageImpl extends MessageImpl {
}
public Optional<String> getTraceContext() {
- return null == traceContext ? Optional.empty() : Optional.of(traceContext);
+ return Optional.ofNullable(traceContext);
}
/**
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeter.java b/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeter.java
index 8acfe15..1afbd0a 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeter.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeter.java
@@ -20,6 +20,7 @@ package org.apache.rocketmq.client.java.metrics;
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.common.base.MoreObjects;
+import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.metrics.DoubleHistogram;
import io.opentelemetry.api.metrics.Meter;
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
@@ -77,6 +78,15 @@ public class ClientMeter {
return null == histogram ? Optional.empty() : Optional.of(histogram);
}
+ public void record(HistogramEnum histogramEnum, Attributes attributes, double value) {
+ final DoubleHistogram histogram = histogramMap.computeIfAbsent(histogramEnum.getName(), name -> enabled ?
+ meter.histogramBuilder(histogramEnum.getName()).build() : null);
+ if (null == histogram) {
+ return;
+ }
+ histogram.record(value, attributes);
+ }
+
public void shutdown() {
if (!enabled) {
return;
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeterProvider.java b/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeterProvider.java
index 4abddfd..5af1aa1 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeterProvider.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientMeterProvider.java
@@ -23,7 +23,6 @@ import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.opentelemetry.api.common.Attributes;
-import io.opentelemetry.api.metrics.DoubleHistogram;
import io.opentelemetry.api.metrics.Meter;
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
@@ -37,7 +36,6 @@ import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.java.impl.ClientImpl;
import org.apache.rocketmq.client.java.route.Endpoints;
@@ -59,7 +57,6 @@ public class ClientMeterProvider {
public ClientMeterProvider(ClientImpl client) {
this.client = client;
- this.client.registerMessageInterceptor(new MessageMeterInterceptor(this));
this.clientMeter = ClientMeter.disabledInstance(client.clientId());
this.messageCacheObserver = null;
}
@@ -68,10 +65,11 @@ public class ClientMeterProvider {
this.messageCacheObserver = messageCacheObserver;
}
- Optional<DoubleHistogram> getHistogramByEnum(HistogramEnum histogramEnum) {
- return clientMeter.getHistogramByEnum(histogramEnum);
+ public void record(HistogramEnum histogramEnum, Attributes attributes, double value) {
+ clientMeter.record(histogramEnum, attributes, value);
}
+ @SuppressWarnings("deprecation")
public synchronized void reset(Metric metric) {
final String clientId = client.clientId();
try {
@@ -171,11 +169,8 @@ public class ClientMeterProvider {
}
}
+ @SuppressWarnings("BooleanMethodIsAlwaysInverted")
public boolean isEnabled() {
return clientMeter.isEnabled();
}
-
- public ClientImpl getClient() {
- return client;
- }
}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/MessageMeterHandler.java b/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/MessageMeterHandler.java
new file mode 100644
index 0000000..4c2132d
--- /dev/null
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/MessageMeterHandler.java
@@ -0,0 +1,202 @@
+/*
+ * 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.rocketmq.client.java.metrics;
+
+import com.google.common.base.Stopwatch;
+import io.opentelemetry.api.common.Attributes;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import org.apache.rocketmq.client.apis.consumer.PushConsumer;
+import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
+import org.apache.rocketmq.client.java.hook.Attribute;
+import org.apache.rocketmq.client.java.hook.AttributeKey;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContext;
+import org.apache.rocketmq.client.java.hook.MessageHookPoints;
+import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
+import org.apache.rocketmq.client.java.impl.ClientImpl;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MessageMeterHandler implements MessageHandler {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MessageMeterHandler.class);
+
+ private static final AttributeKey<Stopwatch> MESSAGE_SEND_STOPWATCH_KEY = AttributeKey.create(
+ "message_send_stopwatch");
+ private static final AttributeKey<Stopwatch> MESSAGE_CONSUME_STOPWATCH_KEY = AttributeKey.create(
+ "message_consume_stopwatch");
+
+ private final ClientImpl client;
+ private final ClientMeterProvider meterProvider;
+
+ public MessageMeterHandler(ClientImpl client, ClientMeterProvider meterProvider) {
+ this.client = client;
+ this.meterProvider = meterProvider;
+ }
+
+ private void doBeforeSendMessage(MessageHandlerContext context) {
+ // Record the time before sending message.
+ context.putAttribute(MESSAGE_SEND_STOPWATCH_KEY, Attribute.create(Stopwatch.createStarted()));
+ }
+
+ private void doAfterSendMessage(MessageHandlerContext context, List<GeneralMessage> messages) {
+ final Attribute<Stopwatch> stopwatchAttr = context.getAttribute(MESSAGE_SEND_STOPWATCH_KEY);
+ if (null == stopwatchAttr) {
+ // Should never reach here.
+ return;
+ }
+ for (GeneralMessage message : messages) {
+ final Duration duration = stopwatchAttr.get().elapsed();
+ InvocationStatus status = MessageHookPointsStatus.OK.equals(context.getStatus()) ?
+ InvocationStatus.SUCCESS : InvocationStatus.FAILURE;
+ Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, message.getTopic())
+ .put(MetricLabels.CLIENT_ID, client.clientId())
+ .put(MetricLabels.INVOCATION_STATUS, status.getName()).build();
+ meterProvider.record(HistogramEnum.SEND_SUCCESS_COST_TIME, attributes, duration.toMillis());
+ }
+ }
+
+ private void doAfterReceiveMessage(List<GeneralMessage> messages) {
+ if (messages.isEmpty()) {
+ return;
+ }
+ String consumerGroup = null;
+ if (client instanceof PushConsumer) {
+ consumerGroup = ((PushConsumer) client).getConsumerGroup();
+ }
+ if (client instanceof SimpleConsumer) {
+ consumerGroup = ((SimpleConsumer) client).getConsumerGroup();
+ }
+ if (null == consumerGroup) {
+ LOGGER.error("[Bug] consumerGroup is not recognized, clientId={}", client.clientId());
+ return;
+ }
+ final GeneralMessage message = messages.iterator().next();
+ final Optional<Long> optionalTransportDeliveryTimestamp = message.getTransportDeliveryTimestamp();
+ if (!optionalTransportDeliveryTimestamp.isPresent()) {
+ return;
+ }
+ final long transportDeliveryTimestamp = optionalTransportDeliveryTimestamp.get();
+ final long currentTimeMillis = System.currentTimeMillis();
+ final long latency = currentTimeMillis - transportDeliveryTimestamp;
+ if (0 > latency) {
+ LOGGER.debug("latency is negative, latency={}ms, currentTimeMillis={}, transportDeliveryTimestamp={}",
+ latency, currentTimeMillis, transportDeliveryTimestamp);
+ return;
+ }
+ final Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, message.getTopic())
+ .put(MetricLabels.CONSUMER_GROUP, consumerGroup).put(MetricLabels.CLIENT_ID, client.clientId()).build();
+ meterProvider.record(HistogramEnum.DELIVERY_LATENCY, attributes, latency);
+ }
+
+ private void doBeforeConsumeMessage(MessageHandlerContext context, List<GeneralMessage> messages) {
+ String consumerGroup = null;
+ if (client instanceof PushConsumer) {
+ consumerGroup = ((PushConsumer) client).getConsumerGroup();
+ }
+ if (null == consumerGroup) {
+ LOGGER.error("[Bug] consumerGroup is not recognized, clientId={}", client.clientId());
+ return;
+ }
+ final GeneralMessage message = messages.iterator().next();
+ final Optional<Long> optionalDecodeTimestamp = message.getDecodeTimestamp();
+ if (!optionalDecodeTimestamp.isPresent()) {
+ return;
+ }
+ final long decodeTimestamp = optionalDecodeTimestamp.get();
+ Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, message.getTopic())
+ .put(MetricLabels.CONSUMER_GROUP, consumerGroup)
+ .put(MetricLabels.CLIENT_ID, client.clientId()).build();
+ final long latency = System.currentTimeMillis() - decodeTimestamp;
+ meterProvider.record(HistogramEnum.AWAIT_TIME, attributes, latency);
+ // Record the time before consuming message.
+ context.putAttribute(MESSAGE_CONSUME_STOPWATCH_KEY, Attribute.create(Stopwatch.createStarted()));
+ }
+
+ private void doAfterConsumeMessage(MessageHandlerContext context, List<GeneralMessage> messages) {
+ if (!(client instanceof PushConsumer)) {
+ // Should never reach here.
+ LOGGER.error("[Bug] current client is not push consumer, clientId={}", client.clientId());
+ return;
+ }
+ final Attribute<Stopwatch> stopwatchAttr = context.getAttribute(MESSAGE_CONSUME_STOPWATCH_KEY);
+ if (null == stopwatchAttr) {
+ // Should never reach here.
+ return;
+ }
+ PushConsumer pushConsumer = (PushConsumer) client;
+ final MessageHookPointsStatus status = context.getStatus();
+ for (GeneralMessage message : messages) {
+ InvocationStatus invocationStatus = MessageHookPointsStatus.OK.equals(status) ? InvocationStatus.SUCCESS :
+ InvocationStatus.FAILURE;
+ Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, message.getTopic())
+ .put(MetricLabels.CONSUMER_GROUP, pushConsumer.getConsumerGroup())
+ .put(MetricLabels.CLIENT_ID, client.clientId())
+ .put(MetricLabels.INVOCATION_STATUS, invocationStatus.getName())
+ .build();
+ final Duration duration = stopwatchAttr.get().elapsed();
+ meterProvider.record(HistogramEnum.PROCESS_TIME, attributes, duration.toMillis());
+ }
+ }
+
+ @Override
+ public void doBefore(MessageHandlerContext context, List<GeneralMessage> messages) {
+ if (!meterProvider.isEnabled()) {
+ return;
+ }
+ final MessageHookPoints hookPoints = context.getMessageHookPoints();
+ switch (hookPoints) {
+ case SEND: {
+ doBeforeSendMessage(context);
+ break;
+ }
+ case CONSUME: {
+ doBeforeConsumeMessage(context, messages);
+ break;
+ }
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void doAfter(MessageHandlerContext context, List<GeneralMessage> messages) {
+ if (!meterProvider.isEnabled()) {
+ return;
+ }
+ final MessageHookPoints hookPoints = context.getMessageHookPoints();
+ switch (hookPoints) {
+ case SEND: {
+ doAfterSendMessage(context, messages);
+ break;
+ }
+ case RECEIVE: {
+ doAfterReceiveMessage(messages);
+ break;
+ }
+ case CONSUME: {
+ doAfterConsumeMessage(context, messages);
+ break;
+ }
+ default:
+ break;
+ }
+ }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/MessageMeterInterceptor.java b/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/MessageMeterInterceptor.java
deleted file mode 100644
index f9c6a70..0000000
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/metrics/MessageMeterInterceptor.java
+++ /dev/null
@@ -1,192 +0,0 @@
-/*
- * 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.rocketmq.client.java.metrics;
-
-import com.google.protobuf.Timestamp;
-import com.google.protobuf.util.Timestamps;
-import io.opentelemetry.api.common.Attributes;
-import io.opentelemetry.api.metrics.DoubleHistogram;
-import java.time.Duration;
-import java.util.List;
-import java.util.Optional;
-import org.apache.rocketmq.client.apis.consumer.PushConsumer;
-import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
-import org.apache.rocketmq.client.java.hook.MessageHookPoints;
-import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
-import org.apache.rocketmq.client.java.impl.ClientImpl;
-import org.apache.rocketmq.client.java.message.MessageCommon;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class MessageMeterInterceptor implements MessageInterceptor {
- private static final Logger LOGGER = LoggerFactory.getLogger(MessageMeterInterceptor.class);
-
- private final ClientMeterProvider clientMeterProvider;
-
- public MessageMeterInterceptor(ClientMeterProvider clientMeterProvider) {
- this.clientMeterProvider = clientMeterProvider;
- }
-
- private void doAfterSendMessage(List<MessageCommon> messageCommons, Duration duration,
- MessageHookPointsStatus status) {
- final Optional<DoubleHistogram> optionalHistogram =
- clientMeterProvider.getHistogramByEnum(HistogramEnum.SEND_SUCCESS_COST_TIME);
- if (!optionalHistogram.isPresent()) {
- return;
- }
- final DoubleHistogram histogram = optionalHistogram.get();
- for (MessageCommon messageCommon : messageCommons) {
- InvocationStatus invocationStatus = MessageHookPointsStatus.OK.equals(status) ? InvocationStatus.SUCCESS :
- InvocationStatus.FAILURE;
- Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, messageCommon.getTopic())
- .put(MetricLabels.CLIENT_ID, clientMeterProvider.getClient().clientId())
- .put(MetricLabels.INVOCATION_STATUS, invocationStatus.getName()).build();
- histogram.record(duration.toMillis(), attributes);
- }
- }
-
- private void doAfterReceive(List<MessageCommon> messageCommons) {
- if (messageCommons.isEmpty()) {
- return;
- }
- final ClientImpl client = clientMeterProvider.getClient();
- String consumerGroup = null;
- if (client instanceof PushConsumer) {
- consumerGroup = ((PushConsumer) client).getConsumerGroup();
- }
- if (client instanceof SimpleConsumer) {
- consumerGroup = ((SimpleConsumer) client).getConsumerGroup();
- }
- if (null == consumerGroup) {
- LOGGER.error("[Bug] consumerGroup is not recognized, clientId={}", client.clientId());
- return;
- }
- final MessageCommon messageCommon = messageCommons.iterator().next();
- final Optional<Timestamp> optionalDeliveryTimestampFromRemote = messageCommon.getDeliveryTimestampFromRemote();
- if (!optionalDeliveryTimestampFromRemote.isPresent()) {
- return;
- }
- final Timestamp deliveryTimestampFromRemote = optionalDeliveryTimestampFromRemote.get();
- final long currentTimeMillis = System.currentTimeMillis();
- final long deliveryTimestampFromRemoteMillis = Timestamps.toMillis(deliveryTimestampFromRemote);
- final long latency = currentTimeMillis - deliveryTimestampFromRemoteMillis;
- if (0 > latency) {
- LOGGER.debug("[Bug] latency is negative, latency={}ms, currentTimeMillis={}, "
- + "deliveryTimestampFromRemoteMillis={}", latency, currentTimeMillis,
- deliveryTimestampFromRemoteMillis);
- return;
- }
- final Optional<DoubleHistogram> optionalHistogram =
- clientMeterProvider.getHistogramByEnum(HistogramEnum.DELIVERY_LATENCY);
- if (!optionalHistogram.isPresent()) {
- return;
- }
- final DoubleHistogram histogram = optionalHistogram.get();
- final Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, messageCommon.getTopic())
- .put(MetricLabels.CONSUMER_GROUP, consumerGroup)
- .put(MetricLabels.CLIENT_ID, client.clientId()).build();
- histogram.record(latency, attributes);
- }
-
- private void doBeforeConsumeMessage(List<MessageCommon> messageCommons) {
- final ClientImpl client = clientMeterProvider.getClient();
- String consumerGroup = null;
- if (client instanceof PushConsumer) {
- consumerGroup = ((PushConsumer) client).getConsumerGroup();
- }
- if (null == consumerGroup) {
- LOGGER.error("[Bug] consumerGroup is not recognized, clientId={}", client.clientId());
- return;
- }
- final MessageCommon messageCommon = messageCommons.iterator().next();
- final Optional<Duration> optionalDurationAfterDecoding = messageCommon.getDurationAfterDecoding();
- if (!optionalDurationAfterDecoding.isPresent()) {
- return;
- }
- final Duration durationAfterDecoding = optionalDurationAfterDecoding.get();
- Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, messageCommon.getTopic())
- .put(MetricLabels.CONSUMER_GROUP, consumerGroup)
- .put(MetricLabels.CLIENT_ID, client.clientId()).build();
- final Optional<DoubleHistogram> optionalHistogram =
- clientMeterProvider.getHistogramByEnum(HistogramEnum.AWAIT_TIME);
- if (!optionalHistogram.isPresent()) {
- return;
- }
- final DoubleHistogram histogram = optionalHistogram.get();
- histogram.record(durationAfterDecoding.toMillis(), attributes);
- }
-
- private void doAfterProcessMessage(List<MessageCommon> messageCommons, Duration duration,
- MessageHookPointsStatus status) {
- final ClientImpl client = clientMeterProvider.getClient();
- if (!(client instanceof PushConsumer)) {
- // Should never reach here.
- LOGGER.error("[Bug] current client is not push consumer, clientId={}", client.clientId());
- return;
- }
- PushConsumer pushConsumer = (PushConsumer) client;
- for (MessageCommon messageCommon : messageCommons) {
- InvocationStatus invocationStatus = MessageHookPointsStatus.OK.equals(status) ? InvocationStatus.SUCCESS :
- InvocationStatus.FAILURE;
- Attributes attributes = Attributes.builder().put(MetricLabels.TOPIC, messageCommon.getTopic())
- .put(MetricLabels.CONSUMER_GROUP, pushConsumer.getConsumerGroup())
- .put(MetricLabels.CLIENT_ID, clientMeterProvider.getClient().clientId())
- .put(MetricLabels.INVOCATION_STATUS, invocationStatus.getName())
- .build();
- final Optional<DoubleHistogram> optionalHistogram =
- clientMeterProvider.getHistogramByEnum(HistogramEnum.PROCESS_TIME);
- if (!optionalHistogram.isPresent()) {
- return;
- }
- final DoubleHistogram histogram = optionalHistogram.get();
- histogram.record(duration.toMillis(), attributes);
- }
- }
-
- @Override
- public void doBefore(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons) {
- if (!clientMeterProvider.isEnabled()) {
- return;
- }
- if (MessageHookPoints.CONSUME.equals(messageHookPoints)) {
- doBeforeConsumeMessage(messageCommons);
- }
- }
-
- @Override
- public void doAfter(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons, Duration duration,
- MessageHookPointsStatus status) {
- if (!clientMeterProvider.isEnabled()) {
- return;
- }
- switch (messageHookPoints) {
- case SEND:
- doAfterSendMessage(messageCommons, duration, status);
- break;
- case RECEIVE:
- doAfterReceive(messageCommons);
- break;
- case CONSUME:
- doAfterProcessMessage(messageCommons, duration, status);
- break;
- default:
- break;
- }
- }
-}
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java
index 8dd7be5..dee9e10 100644
--- a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java
@@ -32,7 +32,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.misc.ThreadFactoryImpl;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
@@ -43,7 +43,7 @@ import org.mockito.Mockito;
public class ConsumeServiceTest extends TestBase {
private final String clientId = "clientId";
private final ConcurrentMap<MessageQueueImpl, ProcessQueue> table = new ConcurrentHashMap<>();
- private final MessageInterceptor interceptor = Mockito.mock(MessageInterceptor.class);
+ private final MessageHandler interceptor = Mockito.mock(MessageHandler.class);
private final ThreadPoolExecutor consumptionExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(), new ThreadFactoryImpl("TestMessageConsumption"));
private final ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(1, new ThreadFactoryImpl(
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java
index 3cc63f0..2264a80 100644
--- a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java
@@ -21,7 +21,7 @@ import static org.junit.Assert.assertEquals;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.tool.TestBase;
import org.junit.Test;
@@ -35,8 +35,8 @@ public class ConsumeTaskTest extends TestBase {
final MessageViewImpl messageView = fakeMessageViewImpl();
final MessageListener messageListener = Mockito.mock(MessageListener.class);
Mockito.when(messageListener.consume(messageView)).thenReturn(ConsumeResult.SUCCESS);
- final MessageInterceptor messageInterceptor = Mockito.mock(MessageInterceptor.class);
- final ConsumeTask consumeTask = new ConsumeTask(clientId, messageListener, messageView, messageInterceptor);
+ final MessageHandler messageHandler = Mockito.mock(MessageHandler.class);
+ final ConsumeTask consumeTask = new ConsumeTask(clientId, messageListener, messageView, messageHandler);
final ConsumeResult consumeResult = consumeTask.call();
assertEquals(ConsumeResult.SUCCESS, consumeResult);
}
@@ -47,8 +47,8 @@ public class ConsumeTaskTest extends TestBase {
final MessageViewImpl messageView = fakeMessageViewImpl();
final MessageListener messageListener = Mockito.mock(MessageListener.class);
Mockito.when(messageListener.consume(messageView)).thenThrow(new RuntimeException());
- final MessageInterceptor messageInterceptor = Mockito.mock(MessageInterceptor.class);
- final ConsumeTask consumeTask = new ConsumeTask(clientId, messageListener, messageView, messageInterceptor);
+ final MessageHandler messageHandler = Mockito.mock(MessageHandler.class);
+ final ConsumeTask consumeTask = new ConsumeTask(clientId, messageListener, messageView, messageHandler);
final ConsumeResult consumeResult = consumeTask.call();
assertEquals(ConsumeResult.FAILURE, consumeResult);
}
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeServiceTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeServiceTest.java
index 19eb29b..de513a7 100644
--- a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeServiceTest.java
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeServiceTest.java
@@ -23,17 +23,15 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
-import org.apache.rocketmq.client.java.hook.MessageHookPoints;
-import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContext;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
import org.apache.rocketmq.client.java.tool.TestBase;
@@ -67,14 +65,13 @@ public class FifoConsumeServiceTest extends TestBase {
when(processQueue1.tryTakeFifoMessages()).thenReturn(messageViewList1.iterator());
MessageListener listener = messageView -> ConsumeResult.SUCCESS;
- MessageInterceptor interceptor = new MessageInterceptor() {
+ MessageHandler interceptor = new MessageHandler() {
@Override
- public void doBefore(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons) {
+ public void doBefore(MessageHandlerContext context, List<GeneralMessage> messages) {
}
@Override
- public void doAfter(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons,
- Duration duration, MessageHookPointsStatus status) {
+ public void doAfter(MessageHandlerContext context, List<GeneralMessage> messages) {
}
};
final FifoConsumeService service = new FifoConsumeService(FAKE_CLIENT_ID, processQueueTable, listener,
@@ -85,6 +82,5 @@ public class FifoConsumeServiceTest extends TestBase {
verify(processQueue1, times(1)).tryTakeFifoMessages();
verify(processQueue0, times(2)).eraseFifoMessage(any(MessageViewImpl.class), any(ConsumeResult.class));
verify(processQueue1, times(1)).eraseFifoMessage(any(MessageViewImpl.class), any(ConsumeResult.class));
-
}
}
\ No newline at end of file
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeServiceTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeServiceTest.java
index 45d9c91..f2ea49d 100644
--- a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeServiceTest.java
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeServiceTest.java
@@ -23,17 +23,15 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
-import org.apache.rocketmq.client.java.hook.MessageHookPoints;
-import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
-import org.apache.rocketmq.client.java.hook.MessageInterceptor;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.hook.MessageHandler;
+import org.apache.rocketmq.client.java.hook.MessageHandlerContext;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
import org.apache.rocketmq.client.java.tool.TestBase;
@@ -60,14 +58,15 @@ public class StandardConsumeServiceTest extends TestBase {
when(processQueue1.tryTakeMessage()).thenReturn(Optional.of(messageView1));
MessageListener listener = messageView -> ConsumeResult.SUCCESS;
- MessageInterceptor interceptor = new MessageInterceptor() {
+ MessageHandler interceptor = new MessageHandler() {
@Override
- public void doBefore(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons) {
+ public void doBefore(MessageHandlerContext context, List<GeneralMessage> messages) {
+
}
@Override
- public void doAfter(MessageHookPoints messageHookPoints, List<MessageCommon> messageCommons,
- Duration duration, MessageHookPointsStatus status) {
+ public void doAfter(MessageHandlerContext context, List<GeneralMessage> messages) {
+
}
};
final StandardConsumeService service = new StandardConsumeService(FAKE_CLIENT_ID, processQueueTable, listener,
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/TransactionImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/TransactionImplTest.java
index d7d547f..6cca321 100644
--- a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/TransactionImplTest.java
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/TransactionImplTest.java
@@ -29,7 +29,7 @@ import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.producer.TransactionResolution;
-import org.apache.rocketmq.client.java.message.MessageCommon;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
import org.apache.rocketmq.client.java.message.MessageType;
import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
import org.apache.rocketmq.client.java.route.Endpoints;
@@ -114,7 +114,7 @@ public class TransactionImplTest extends TestBase {
final PublishingMessageImpl publishingMessage = transaction.tryAddMessage(message);
final SendReceiptImpl sendReceipt = fakeSendReceiptImpl(fakeMessageQueueImpl(FAKE_TOPIC_0));
transaction.tryAddReceipt(publishingMessage, sendReceipt);
- Mockito.doNothing().when(producer).endTransaction(any(Endpoints.class), any(MessageCommon.class),
+ Mockito.doNothing().when(producer).endTransaction(any(Endpoints.class), any(GeneralMessage.class),
any(MessageId.class), anyString(), any(TransactionResolution.class));
transaction.commit();
}
@@ -129,7 +129,7 @@ public class TransactionImplTest extends TestBase {
final PublishingMessageImpl publishingMessage = transaction.tryAddMessage(message);
final SendReceiptImpl sendReceipt = fakeSendReceiptImpl(fakeMessageQueueImpl(FAKE_TOPIC_0));
transaction.tryAddReceipt(publishingMessage, sendReceipt);
- Mockito.doNothing().when(producer).endTransaction(any(Endpoints.class), any(MessageCommon.class),
+ Mockito.doNothing().when(producer).endTransaction(any(Endpoints.class), any(GeneralMessage.class),
any(MessageId.class), anyString(), any(TransactionResolution.class));
transaction.rollback();
}
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/message/GeneralMessageImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/message/GeneralMessageImplTest.java
new file mode 100644
index 0000000..8b1bf89
--- /dev/null
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/message/GeneralMessageImplTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.rocketmq.client.java.message;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.client.apis.message.MessageId;
+import org.apache.rocketmq.client.java.route.MessageQueueImpl;
+import org.apache.rocketmq.client.java.tool.TestBase;
+import org.junit.Test;
+
+public class GeneralMessageImplTest extends TestBase {
+ @Test
+ public void testMessage() {
+ String topic = "testTopic";
+ byte[] body = "foobar".getBytes(StandardCharsets.UTF_8);
+ String tag = "tagA";
+ List<String> keys = new ArrayList<>();
+ keys.add("keyA");
+ String messageGroup = "messageGroup0";
+ String parentTraceContext = "parentTraceContext0";
+ long deliveryTimestamp = System.currentTimeMillis();
+ Map<String, String> properties = new HashMap<>();
+ properties.put("propertyA", "valueA");
+
+ final MessageImpl message = new MessageImpl(topic, body, tag, keys, messageGroup, parentTraceContext,
+ deliveryTimestamp, properties);
+ final GeneralMessageImpl generalMessage = new GeneralMessageImpl(message);
+ assertFalse(generalMessage.getMessageId().isPresent());
+ assertEquals(topic, generalMessage.getTopic());
+ assertEquals(ByteBuffer.wrap(body), generalMessage.getBody());
+ assertEquals(properties, generalMessage.getProperties());
+ assertTrue(generalMessage.getTag().isPresent());
+ assertEquals(tag, generalMessage.getTag().get());
+ assertEquals(keys, generalMessage.getKeys());
+ assertTrue(generalMessage.getMessageGroup().isPresent());
+ assertEquals(messageGroup, generalMessage.getMessageGroup().get());
+ assertTrue(generalMessage.getParentTraceContext().isPresent());
+ assertEquals(parentTraceContext, generalMessage.getParentTraceContext().get());
+ assertFalse(generalMessage.getTraceContext().isPresent());
+ assertTrue(generalMessage.getDeliveryTimestamp().isPresent());
+ assertEquals(deliveryTimestamp, (long) generalMessage.getDeliveryTimestamp().get());
+ assertFalse(generalMessage.getBornHost().isPresent());
+ assertFalse(generalMessage.getBornTimestamp().isPresent());
+ assertFalse(generalMessage.getDeliveryAttempt().isPresent());
+ assertFalse(generalMessage.getDecodeTimestamp().isPresent());
+ assertFalse(generalMessage.getTransportDeliveryTimestamp().isPresent());
+ }
+
+ @Test
+ public void testMessageView() {
+ MessageId messageId = MessageIdCodec.getInstance().nextMessageId();
+ String topic = "testTopic";
+ byte[] body = "foobar".getBytes(StandardCharsets.UTF_8);
+ String tag = "tagA";
+ String messageGroup = "messageGroup0";
+ long deliveryTimestamp = System.currentTimeMillis();
+ List<String> keys = new ArrayList<>();
+ keys.add("keyA");
+ Map<String, String> properties = new HashMap<>();
+ properties.put("propertyA", "valueA");
+ String bornHost = "bornHost0";
+ long bornTimestamp = System.currentTimeMillis();
+ int deliveryAttempt = 1;
+ final MessageQueueImpl mq = fakeMessageQueueImpl(topic);
+ String receiptHandle = "receiptHandle0";
+ String traceContext = "traceContext0";
+ long offset = 8;
+ boolean corrupted = false;
+ long transportDeliveryTimestamp = System.currentTimeMillis();
+
+ final MessageViewImpl messageView = new MessageViewImpl(messageId, topic, body, tag, messageGroup,
+ deliveryTimestamp, keys, properties, bornHost, bornTimestamp, deliveryAttempt, mq, receiptHandle,
+ traceContext, offset, corrupted, transportDeliveryTimestamp);
+ final GeneralMessageImpl generalMessage = new GeneralMessageImpl(messageView);
+ assertTrue(generalMessage.getMessageId().isPresent());
+ assertEquals(messageId, generalMessage.getMessageId().get());
+ assertEquals(topic, generalMessage.getTopic());
+ assertEquals(ByteBuffer.wrap(body), generalMessage.getBody());
+ assertEquals(properties, generalMessage.getProperties());
+ assertTrue(generalMessage.getTag().isPresent());
+ assertEquals(tag, generalMessage.getTag().get());
+ assertEquals(keys, generalMessage.getKeys());
+ assertTrue(generalMessage.getMessageGroup().isPresent());
+ assertEquals(messageGroup, generalMessage.getMessageGroup().get());
+ assertFalse(generalMessage.getParentTraceContext().isPresent());
+ assertTrue(generalMessage.getTraceContext().isPresent());
+ assertEquals(traceContext, generalMessage.getTraceContext().get());
+ assertTrue(generalMessage.getDeliveryTimestamp().isPresent());
+ assertEquals(deliveryTimestamp, (long) generalMessage.getDeliveryTimestamp().get());
+ assertTrue(generalMessage.getBornHost().isPresent());
+ assertEquals(bornHost, generalMessage.getBornHost().get());
+ assertTrue(generalMessage.getBornTimestamp().isPresent());
+ assertEquals(bornTimestamp, (long) generalMessage.getBornTimestamp().get());
+ assertTrue(generalMessage.getDeliveryAttempt().isPresent());
+ assertEquals(deliveryAttempt, (int) generalMessage.getDeliveryAttempt().get());
+ assertTrue(generalMessage.getDecodeTimestamp().isPresent());
+ assertEquals(messageView.getDecodeTimestamp(), (long) generalMessage.getDecodeTimestamp().get());
+ assertTrue(generalMessage.getTransportDeliveryTimestamp().isPresent());
+ assertEquals(transportDeliveryTimestamp, (long) generalMessage.getTransportDeliveryTimestamp().get());
+ }
+}
\ No newline at end of file
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/message/MessageViewImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/message/MessageViewImplTest.java
new file mode 100644
index 0000000..92036ec
--- /dev/null
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/message/MessageViewImplTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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.rocketmq.client.java.message;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import apache.rocketmq.v2.Digest;
+import apache.rocketmq.v2.DigestType;
+import apache.rocketmq.v2.Message;
+import apache.rocketmq.v2.MessageType;
+import apache.rocketmq.v2.Resource;
+import apache.rocketmq.v2.SystemProperties;
+import com.google.protobuf.ByteString;
+import java.nio.charset.StandardCharsets;
+import org.apache.rocketmq.client.java.tool.TestBase;
+import org.junit.Test;
+
+public class MessageViewImplTest extends TestBase {
+
+ @Test
+ public void testFromProtobufWithCrc32() {
+ final Digest digest = Digest.newBuilder().setType(DigestType.CRC32).setChecksum("9EF61F95").build();
+ SystemProperties systemProperties = SystemProperties.newBuilder().setMessageType(MessageType.NORMAL)
+ .setMessageId(MessageIdCodec.getInstance().nextMessageId().toString())
+ .setBornHost(FAKE_HOST_0)
+ .setBodyDigest(digest)
+ .build();
+ String topic = FAKE_TOPIC_0;
+ Resource resource = Resource.newBuilder().setName(topic).build();
+ final ByteString body = ByteString.copyFrom("foobar", StandardCharsets.UTF_8);
+ final Message message = Message.newBuilder().setSystemProperties(systemProperties)
+ .setTopic(resource).setBody(body)
+ .setSystemProperties(systemProperties).build();
+ MessageViewImpl messageView = MessageViewImpl.fromProtobuf(message);
+ assertEquals(body.asReadOnlyByteBuffer(), messageView.getBody());
+ assertEquals(topic, messageView.getTopic());
+ assertFalse(messageView.isCorrupted());
+ }
+
+ @Test
+ public void testFromProtobufWithWrongCrc32() {
+ final Digest digest = Digest.newBuilder().setType(DigestType.CRC32).setChecksum("9EF61F96").build();
+ SystemProperties systemProperties = SystemProperties.newBuilder().setMessageType(MessageType.NORMAL)
+ .setMessageId(MessageIdCodec.getInstance().nextMessageId().toString())
+ .setBornHost(FAKE_HOST_0)
+ .setBodyDigest(digest)
+ .build();
+ String topic = FAKE_TOPIC_0;
+ Resource resource = Resource.newBuilder().setName(topic).build();
+ final ByteString body = ByteString.copyFrom("foobar", StandardCharsets.UTF_8);
+ final Message message = Message.newBuilder().setSystemProperties(systemProperties)
+ .setTopic(resource).setBody(body)
+ .setSystemProperties(systemProperties).build();
+ MessageViewImpl messageView = MessageViewImpl.fromProtobuf(message);
+ assertEquals(body.asReadOnlyByteBuffer(), messageView.getBody());
+ assertEquals(topic, messageView.getTopic());
+ assertTrue(messageView.isCorrupted());
+ }
+
+ @Test
+ public void testFromProtobufWithMd5() {
+ final Digest digest = Digest.newBuilder().setType(DigestType.MD5).setChecksum(
+ "3858F62230AC3C915F300C664312C63F").build();
+ SystemProperties systemProperties = SystemProperties.newBuilder().setMessageType(MessageType.NORMAL)
+ .setMessageId(MessageIdCodec.getInstance().nextMessageId().toString())
+ .setBornHost(FAKE_HOST_0)
+ .setBodyDigest(digest)
+ .build();
+ String topic = FAKE_TOPIC_0;
+ Resource resource = Resource.newBuilder().setName(topic).build();
+ final ByteString body = ByteString.copyFrom("foobar", StandardCharsets.UTF_8);
+ final Message message = Message.newBuilder().setSystemProperties(systemProperties)
+ .setTopic(resource).setBody(body)
+ .setSystemProperties(systemProperties).build();
+ MessageViewImpl messageView = MessageViewImpl.fromProtobuf(message);
+ assertEquals(body.asReadOnlyByteBuffer(), messageView.getBody());
+ assertEquals(topic, messageView.getTopic());
+ assertFalse(messageView.isCorrupted());
+ }
+
+ @Test
+ public void testFromProtobufWithWrongMd5() {
+ final Digest digest = Digest.newBuilder().setType(DigestType.MD5).setChecksum(
+ "3858F62230AC3C915F300C664312C63G").build();
+ SystemProperties systemProperties = SystemProperties.newBuilder().setMessageType(MessageType.NORMAL)
+ .setMessageId(MessageIdCodec.getInstance().nextMessageId().toString())
+ .setBornHost(FAKE_HOST_0)
+ .setBodyDigest(digest)
+ .build();
+ String topic = FAKE_TOPIC_0;
+ Resource resource = Resource.newBuilder().setName(topic).build();
+ final ByteString body = ByteString.copyFrom("foobar", StandardCharsets.UTF_8);
+ final Message message = Message.newBuilder().setSystemProperties(systemProperties)
+ .setTopic(resource).setBody(body)
+ .setSystemProperties(systemProperties).build();
+ MessageViewImpl messageView = MessageViewImpl.fromProtobuf(message);
+ assertEquals(body.asReadOnlyByteBuffer(), messageView.getBody());
+ assertEquals(topic, messageView.getTopic());
+ assertTrue(messageView.isCorrupted());
+ }
+
+ @Test
+ public void testFromProtobufWithSha1() {
+ final Digest digest = Digest.newBuilder().setType(DigestType.SHA1).setChecksum(
+ "8843D7F92416211DE9EBB963FF4CE28125932878").build();
+ SystemProperties systemProperties = SystemProperties.newBuilder().setMessageType(MessageType.NORMAL)
+ .setMessageId(MessageIdCodec.getInstance().nextMessageId().toString())
+ .setBornHost(FAKE_HOST_0)
+ .setBodyDigest(digest)
+ .build();
+ String topic = FAKE_TOPIC_0;
+ Resource resource = Resource.newBuilder().setName(topic).build();
+ final ByteString body = ByteString.copyFrom("foobar", StandardCharsets.UTF_8);
+ final Message message = Message.newBuilder().setSystemProperties(systemProperties)
+ .setTopic(resource).setBody(body)
+ .setSystemProperties(systemProperties).build();
+ MessageViewImpl messageView = MessageViewImpl.fromProtobuf(message);
+ assertEquals(body.asReadOnlyByteBuffer(), messageView.getBody());
+ assertEquals(topic, messageView.getTopic());
+ assertFalse(messageView.isCorrupted());
+ }
+
+ @Test
+ public void testFromProtobufWithWrongSha1() {
+ final Digest digest = Digest.newBuilder().setType(DigestType.SHA1).setChecksum(
+ "8843D7F92416211DE9EBB963FF4CE28125932879").build();
+ SystemProperties systemProperties = SystemProperties.newBuilder().setMessageType(MessageType.NORMAL)
+ .setMessageId(MessageIdCodec.getInstance().nextMessageId().toString())
+ .setBornHost(FAKE_HOST_0)
+ .setBodyDigest(digest)
+ .build();
+ String topic = FAKE_TOPIC_0;
+ Resource resource = Resource.newBuilder().setName(topic).build();
+ final ByteString body = ByteString.copyFrom("foobar", StandardCharsets.UTF_8);
+ final Message message = Message.newBuilder().setSystemProperties(systemProperties)
+ .setTopic(resource).setBody(body)
+ .setSystemProperties(systemProperties).build();
+ MessageViewImpl messageView = MessageViewImpl.fromProtobuf(message);
+ assertEquals(body.asReadOnlyByteBuffer(), messageView.getBody());
+ assertEquals(topic, messageView.getTopic());
+ assertTrue(messageView.isCorrupted());
+ }
+}
\ No newline at end of file
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/tool/TestBase.java b/java/client/src/test/java/org/apache/rocketmq/client/java/tool/TestBase.java
index 9a6a1b0..e0da4a1 100644
--- a/java/client/src/test/java/org/apache/rocketmq/client/java/tool/TestBase.java
+++ b/java/client/src/test/java/org/apache/rocketmq/client/java/tool/TestBase.java
@@ -181,7 +181,8 @@ public class TestBase {
Map<String, String> properties = new HashMap<>();
List<String> keys = new ArrayList<>();
return new MessageViewImpl(messageId, FAKE_TOPIC_0, body, null, null, null,
- keys, properties, FAKE_HOST_0, 1, 1, mq, FAKE_RECEIPT_HANDLE_0, null, 1, corrupted, null);
+ keys, properties, FAKE_HOST_0, 1, 1, mq, FAKE_RECEIPT_HANDLE_0, null, 1, corrupted,
+ System.currentTimeMillis());
}
protected MessageQueueImpl fakeMessageQueueImpl(String topic) {