This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-spi-extensions.git
The following commit(s) were added to refs/heads/master by this push:
new b368f1c Support rocketmq protocol (#138)
b368f1c is described below
commit b368f1c0eee0242bf11be9718e1c70b336294b60
Author: githublaohu <[email protected]>
AuthorDate: Tue Oct 11 16:59:13 2022 +0800
Support rocketmq protocol (#138)
---
dubbo-api-docs/pom.xml | 2 +-
.../dubbo-registry-dns/pom.xml | 5 +
.../dubbo-registry-nameservice/pom.xml | 61 +++++
.../registry/nameservice/NameServiceRegistry.java | 241 +++++++++++++++++++
.../nameservice/NameServiceRegistryFactory.java | 30 +++
.../dubbo/registry/nameservice/ServiceName.java | 180 ++++++++++++++
.../org.apache.dubbo.registry.RegistryFactory | 1 +
dubbo-registry-extensions/pom.xml | 7 +-
.../dubbo-rpc-rocketmq}/pom.xml | 40 ++--
.../apache/dubbo/rpc/rocketmq/RocketMQChannel.java | 183 ++++++++++++++
.../dubbo/rpc/rocketmq/RocketMQExporter.java | 76 ++++++
.../apache/dubbo/rpc/rocketmq/RocketMQInvoker.java | 232 ++++++++++++++++++
.../dubbo/rpc/rocketmq/RocketMQProtocol.java | 266 +++++++++++++++++++++
.../rpc/rocketmq/RocketMQProtocolConstant.java | 38 +++
.../dubbo/rpc/rocketmq/RocketMQProtocolServer.java | 164 +++++++++++++
.../rocketmq/codec/DecodeableRpcInvocation.java | 226 +++++++++++++++++
.../rpc/rocketmq/codec/DecodeableRpcResult.java | 179 ++++++++++++++
.../dubbo/rpc/rocketmq/codec/RocketMQCodec.java | 230 ++++++++++++++++++
.../rpc/rocketmq/codec/RocketMQCodecSupport.java | 55 +++++
.../rpc/rocketmq/codec/RocketMQCountCodec.java | 45 ++++
.../dubbo/internal/org.apache.dubbo.rpc.Protocol | 1 +
dubbo-rpc-extensions/pom.xml | 1 +
22 files changed, 2239 insertions(+), 24 deletions(-)
diff --git a/dubbo-api-docs/pom.xml b/dubbo-api-docs/pom.xml
index f1d1ac7..9756e8b 100644
--- a/dubbo-api-docs/pom.xml
+++ b/dubbo-api-docs/pom.xml
@@ -163,7 +163,7 @@
<build>
<plugins>
- <plugin>
+ <plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<version>${maven-checkstyle-plugin-version}</version>
diff --git a/dubbo-registry-extensions/dubbo-registry-dns/pom.xml
b/dubbo-registry-extensions/dubbo-registry-dns/pom.xml
index 6240bea..f06665b 100644
--- a/dubbo-registry-extensions/dubbo-registry-dns/pom.xml
+++ b/dubbo-registry-extensions/dubbo-registry-dns/pom.xml
@@ -43,6 +43,11 @@
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.alibaba</groupId>
+ <artifactId>fastjson</artifactId>
+ </dependency>
+
</dependencies>
</project>
diff --git a/dubbo-registry-extensions/dubbo-registry-nameservice/pom.xml
b/dubbo-registry-extensions/dubbo-registry-nameservice/pom.xml
new file mode 100644
index 0000000..3785fa7
--- /dev/null
+++ b/dubbo-registry-extensions/dubbo-registry-nameservice/pom.xml
@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
+<project
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";
+ xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";>
+ <parent>
+ <artifactId>dubbo-registry-extensions</artifactId>
+ <groupId>org.apache.dubbo.extensions</groupId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>dubbo-registry-nameservice</artifactId>
+ <name>dubbo-registry-nameservice</name>
+ <properties>
+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>3.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.dubbo</groupId>
+ <artifactId>dubbo-registry-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.dubbo</groupId>
+ <artifactId>dubbo-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <version>4.9.3</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-tools</artifactId>
+ <version>4.9.3</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/NameServiceRegistry.java
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/NameServiceRegistry.java
new file mode 100644
index 0000000..961da40
--- /dev/null
+++
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/NameServiceRegistry.java
@@ -0,0 +1,241 @@
+/*
+ * 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.dubbo.registry.nameservice;
+
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.URLBuilder;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.constants.RegistryConstants;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.registry.NotifyListener;
+import org.apache.dubbo.registry.support.FailbackRegistry;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.PermName;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.protocol.body.GroupList;
+import org.apache.rocketmq.common.protocol.body.TopicList;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.QueueData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+
+public class NameServiceRegistry extends FailbackRegistry {
+
+ private static final Logger logger =
LoggerFactory.getLogger(NameServiceRegistry.class);
+
+ private ScheduledExecutorService scheduledExecutorService;
+
+ private Map<URL, RegistryInfoWrapper> consumerRegistryInfoWrapperMap = new
ConcurrentHashMap<>();
+
+ private MQAdminExt mqAdminExt;
+
+ private boolean isNotRoute = true;
+
+ private ClusterInfo clusterInfo;
+
+ private TopicList topicList;
+
+ private long timeoutMillis;
+
+ private String instanceName;
+
+ public NameServiceRegistry(URL url) {
+ super(url);
+ this.isNotRoute = url.getParameter("route", true);
+ if (this.isNotRoute) {
+ return;
+ }
+ this.timeoutMillis = url.getParameter("timeoutMillis", 3000);
+ this.instanceName = url.getParameter("instanceName",
"nameservic-registry");
+ DefaultMQAdminExt clientConfig = new DefaultMQAdminExt();
+ clientConfig.setNamesrvAddr(url.getAddress());
+ clientConfig.setInstanceName(instanceName);
+ mqAdminExt = new DefaultMQAdminExtImpl(clientConfig,
this.timeoutMillis);
+ try {
+ mqAdminExt.start();
+ this.initBeasInfo();
+ } catch (Exception e) {
+ String exeptionInfo = String.format("initBeasInfo pullRoute
exception , cause %s ", e.getMessage());
+ logger.error(exeptionInfo, e);
+ throw new RuntimeException(exeptionInfo, e);
+ }
+ this.scheduledExecutorService =
Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ return new Thread(r, "dubbo-registry-nameservice");
+ }
+ });
+ scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ NameServiceRegistry.this.initBeasInfo();
+
+ if (consumerRegistryInfoWrapperMap.isEmpty()) {
+ return;
+ }
+ for (Entry<URL, RegistryInfoWrapper> e :
consumerRegistryInfoWrapperMap.entrySet()) {
+ List<URL> urls = new ArrayList<URL>();
+
NameServiceRegistry.this.pullRoute(e.getValue().serviceName, e.getKey(), urls);
+ e.getValue().listener.notify(urls);
+ }
+ } catch (Exception e) {
+ String exeptionInfo = String.format("ScheduledTask
pullRoute exception , cause %s ", e.getMessage());
+ logger.error(exeptionInfo, e);
+ }
+ }
+ }, 1000 * 10, 3000 * 10, TimeUnit.MILLISECONDS);
+ }
+
+ private void initBeasInfo() throws Exception {
+ this.clusterInfo = this.mqAdminExt.examineBrokerClusterInfo();
+ this.topicList = this.mqAdminExt.fetchAllTopicList();
+ }
+
+ private URL createProviderURL(ServiceName serviceName, URL url, int queue)
{
+ URLBuilder builder =
URLBuilder.from(url).setProtocol("rocketmq").setAddress(this.getUrl().getAddress());
+ builder.addParameter(CommonConstants.INTERFACE_KEY,
serviceName.getServiceInterface());
+ builder.addParameter(CommonConstants.PATH_KEY,
serviceName.getServiceInterface());
+ builder.addParameter("bean.name", "ServiceBean:" +
serviceName.getServiceInterface());
+ builder.addParameter(CommonConstants.SIDE_KEY,
CommonConstants.PROVIDER);
+ builder.addParameter(RegistryConstants.CATEGORY_KEY, "providers");
+ builder.addParameter(CommonConstants.PROTOCOL_KEY, "rocketmq");
+ builder.addParameter("queueId", queue + "");
+ builder.addParameter("topic", serviceName.getValue());
+ return builder.build();
+ }
+
+ private ServiceName createServiceName(URL url) {
+ return new ServiceName(url);
+ }
+
+ private void createTopic(ServiceName serviceName) {
+ if (!this.topicList.getTopicList().contains(serviceName.getValue())) {
+ try {
+ TopicConfig topicConfig = new
TopicConfig(serviceName.getValue());
+ topicConfig.setReadQueueNums(2);
+ topicConfig.setWriteQueueNums(2);
+ for (Entry<String, BrokerData> entry :
clusterInfo.getBrokerAddrTable().entrySet()) {
+
this.mqAdminExt.createAndUpdateTopicConfig(entry.getValue().selectBrokerAddr(),
topicConfig);
+ }
+ } catch (Exception e) {
+ String exceptionInfo = String.format("create topic fial, topic
name is %s , cause %s", serviceName.getValue(), e.getMessage());
+ logger.error(exceptionInfo, e);
+ throw new RuntimeException(exceptionInfo, e);
+ }
+ }
+ }
+
+ @Override
+ public boolean isAvailable() {
+ return true;
+ }
+
+ @Override
+ public void doRegister(URL url) {
+ ServiceName serviceName = this.createServiceName(url);
+ this.createTopic(serviceName);
+ }
+
+ @Override
+ public void doUnregister(URL url) {
+ }
+
+ @Override
+ public void doSubscribe(URL url, NotifyListener listener) {
+ if (Objects.equals(url.getCategory(),
+
org.apache.dubbo.common.constants.RegistryConstants.CONFIGURATORS_CATEGORY)) {
+ return;
+ }
+ ServiceName serviceName = this.createServiceName(url);
+ try {
+ GroupList groupList =
this.mqAdminExt.queryTopicConsumeByWho(serviceName.getValue());
+ if (Objects.isNull(groupList) ||
groupList.getGroupList().isEmpty()) {
+ return;
+ }
+ } catch (InterruptedException | MQBrokerException | RemotingException
| MQClientException e) {
+ String exceptionInfo = String.format("query topic consume fial,
topic name is %s , url is %s , cause %s", serviceName.getValue(), url,
e.getMessage());
+ logger.error(exceptionInfo, e);
+ throw new RuntimeException(exceptionInfo, e);
+ }
+ List<URL> urls = new ArrayList<URL>();
+ if (this.isNotRoute) {
+ URL providerURL = this.createProviderURL(serviceName, url, -1);
+ urls.add(providerURL);
+ } else {
+ RegistryInfoWrapper registryInfoWrapper = new
RegistryInfoWrapper();
+ registryInfoWrapper.listener = listener;
+ registryInfoWrapper.serviceName = serviceName;
+ consumerRegistryInfoWrapperMap.put(url, registryInfoWrapper);
+ this.pullRoute(serviceName, url, urls);
+ }
+ listener.notify(urls);
+ }
+
+ void pullRoute(ServiceName serviceName, URL url, List<URL> urls) {
+ try {
+ String topic = serviceName.getValue();
+ TopicRouteData topicRouteData =
this.mqAdminExt.examineTopicRouteInfo(topic);
+ for (QueueData queueData : topicRouteData.getQueueDatas()) {
+ if (!PermName.isReadable(queueData.getPerm())) {
+ continue;
+ }
+ for (int i = 0; i < queueData.getReadQueueNums(); i++) {
+ URL newUrl = this.createProviderURL(serviceName, url, i);
+ urls.add(newUrl.addParameter("brokerName",
queueData.getBrokerName()));
+ }
+ }
+ } catch (Exception e) {
+ String exceptionInfo = String.format("query topic route fial,
topic name is %s , url is %s , cause %s", serviceName.getValue(), url,
e.getMessage());
+ logger.error(exceptionInfo, e);
+ throw new RuntimeException(exceptionInfo, e);
+ }
+ }
+
+ @Override
+ public void doUnsubscribe(URL url, NotifyListener listener) {
+ this.consumerRegistryInfoWrapperMap.remove(url);
+ }
+
+ private class RegistryInfoWrapper {
+
+ private NotifyListener listener;
+
+ private ServiceName serviceName;
+ }
+}
diff --git
a/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/NameServiceRegistryFactory.java
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/NameServiceRegistryFactory.java
new file mode 100644
index 0000000..28c27d1
--- /dev/null
+++
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/NameServiceRegistryFactory.java
@@ -0,0 +1,30 @@
+/*
+ * 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.dubbo.registry.nameservice;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.registry.Registry;
+import org.apache.dubbo.registry.support.AbstractRegistryFactory;
+
+public class NameServiceRegistryFactory extends AbstractRegistryFactory {
+
+ @Override
+ protected Registry createRegistry(URL url) {
+ return new NameServiceRegistry(url);
+ }
+
+}
diff --git
a/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/ServiceName.java
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/ServiceName.java
new file mode 100644
index 0000000..be21f05
--- /dev/null
+++
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/java/org/apache/dubbo/registry/nameservice/ServiceName.java
@@ -0,0 +1,180 @@
+/*
+ * 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.dubbo.registry.nameservice;
+
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+import static org.apache.dubbo.common.constants.RegistryConstants.CATEGORY_KEY;
+import static
org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_CATEGORY;
+import static org.apache.dubbo.common.utils.StringUtils.isBlank;
+
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.zip.CRC32;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.StringUtils;
+
+public class ServiceName {
+
+ public static final String DEFAULT_PARAM_VALUE = "";
+
+ public static final String NAME_SEPARATOR = "_";
+
+ public static final String VALUE_SEPARATOR = ",";
+
+ public static final String WILDCARD = "*";
+
+ private String category;
+
+ private String serviceInterface;
+
+ private String version;
+
+ private String group;
+
+ private String value;
+
+ private String groupModel;
+
+ public ServiceName() {
+ }
+
+ public ServiceName(URL url) {
+ this.serviceInterface = url.getParameter(INTERFACE_KEY);
+ this.category = isConcrete(serviceInterface) ? DEFAULT_CATEGORY :
url.getParameter(CATEGORY_KEY);
+ this.version = url.getParameter(VERSION_KEY, DEFAULT_PARAM_VALUE);
+ this.group = url.getParameter(GROUP_KEY, DEFAULT_PARAM_VALUE);
+ this.groupModel = url.getParameter("groupModel");
+ this.value = toValue();
+ }
+
+ public boolean isConcrete() {
+ return isConcrete(serviceInterface) && isConcrete(version) &&
isConcrete(group);
+ }
+
+ public boolean isCompatible(ServiceName serviceName) {
+
+ // The argument must be the concrete NacosServiceName
+ if (!serviceName.isConcrete()) {
+ return false;
+ }
+
+ // Not match comparison
+ if (!StringUtils.isEquals(this.category, serviceName.category)
+ && !matchRange(this.category, serviceName.category)) {
+ return false;
+ }
+
+ if (!StringUtils.isEquals(this.serviceInterface,
serviceName.serviceInterface)) {
+ return false;
+ }
+
+ // wildcard condition
+ if (isWildcard(this.version)) {
+ return true;
+ }
+
+ if (isWildcard(this.group)) {
+ return true;
+ }
+
+ // range condition
+ if (!StringUtils.isEquals(this.version, serviceName.version)
+ && !matchRange(this.version, serviceName.version)) {
+ return false;
+ }
+
+ if (!StringUtils.isEquals(this.group, serviceName.group) &&
+ !matchRange(this.group, serviceName.group)) {
+ return false;
+ }
+
+ return true;
+ }
+
+ private boolean matchRange(String range, String value) {
+ if (isBlank(range)) {
+ return true;
+ }
+ if (!isRange(range)) {
+ return false;
+ }
+ String[] values = range.split(VALUE_SEPARATOR);
+ return Arrays.asList(values).contains(value);
+ }
+
+ private boolean isConcrete(String value) {
+ return !isWildcard(value) && !isRange(value);
+ }
+
+ private boolean isWildcard(String value) {
+ return WILDCARD.equals(value);
+ }
+
+ private boolean isRange(String value) {
+ return value != null && value.contains(VALUE_SEPARATOR) &&
value.split(VALUE_SEPARATOR).length > 1;
+ }
+
+ private String toValue() {
+ String value = null;
+ if (Objects.equals(this.groupModel, "topic")) {
+ value = category +
+ NAME_SEPARATOR + serviceInterface +
+ NAME_SEPARATOR + version +
+ NAME_SEPARATOR + group;
+ } else {
+ value = category + NAME_SEPARATOR + serviceInterface;
+ }
+ CRC32 crc32 = new CRC32();
+ crc32.update(value.getBytes());
+ value = value.replace(".", "-") + NAME_SEPARATOR +
Long.toString(crc32.getValue());
+ return value;
+ }
+
+
+ public String getValue() {
+ return value;
+ }
+
+ public String getServiceInterface() {
+ return this.serviceInterface;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof ServiceName)) {
+ return false;
+ }
+ ServiceName that = (ServiceName) o;
+ return Objects.equals(getValue(), that.getValue());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(getValue());
+ }
+
+ @Override
+ public String toString() {
+ return getValue();
+ }
+}
diff --git
a/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory
new file mode 100644
index 0000000..dc31ed4
--- /dev/null
+++
b/dubbo-registry-extensions/dubbo-registry-nameservice/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory
@@ -0,0 +1 @@
+nameservice=org.apache.dubbo.registry.nameservice.NameServiceRegistryFactory
\ No newline at end of file
diff --git a/dubbo-registry-extensions/pom.xml
b/dubbo-registry-extensions/pom.xml
index 6d8b87f..bd6e82c 100644
--- a/dubbo-registry-extensions/pom.xml
+++ b/dubbo-registry-extensions/pom.xml
@@ -6,9 +6,7 @@
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.
@@ -18,7 +16,7 @@
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
- <parent>
+ <parent>
<groupId>org.apache.dubbo.extensions</groupId>
<artifactId>extensions-parent</artifactId>
<version>${revision}</version>
@@ -37,5 +35,6 @@
<module>dubbo-registry-etcd3</module>
<module>dubbo-registry-redis</module>
<module>dubbo-registry-sofa</module>
- </modules>
+ <module>dubbo-registry-nameservice</module>
+ </modules>
</project>
diff --git a/dubbo-registry-extensions/dubbo-registry-dns/pom.xml
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/pom.xml
similarity index 58%
copy from dubbo-registry-extensions/dubbo-registry-dns/pom.xml
copy to dubbo-rpc-extensions/dubbo-rpc-rocketmq/pom.xml
index 6240bea..25651ee 100644
--- a/dubbo-registry-extensions/dubbo-registry-dns/pom.xml
+++ b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/pom.xml
@@ -1,4 +1,3 @@
-<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
@@ -15,34 +14,37 @@
See the License for the specific language governing permissions and
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0";
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+<project
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";
+ xmlns="http://maven.apache.org/POM/4.0.0";>
+ <modelVersion>4.0.0</modelVersion>
<parent>
- <artifactId>dubbo-registry-extensions</artifactId>
<groupId>org.apache.dubbo.extensions</groupId>
+ <artifactId>dubbo-rpc-extensions</artifactId>
<version>${revision}</version>
<relativePath>../pom.xml</relativePath>
</parent>
- <modelVersion>4.0.0</modelVersion>
-
- <artifactId>dubbo-registry-dns</artifactId>
- <packaging>jar</packaging>
- <name>${project.artifactId}</name>
- <version>1.0.1-SNAPSHOT</version>
- <description>The DNS registry module of Dubbo project</description>
-
+ <artifactId>dubbo-rpc-rocketmq</artifactId>
+ <name>dubbo-rpc-rocketmq</name>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
<dependencies>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>3.8.1</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
- <artifactId>dubbo</artifactId>
- <version>3.0.10</version>
+ <artifactId>dubbo-rpc-api</artifactId>
</dependency>
-
<dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-all</artifactId>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <version>4.9.2</version>
</dependency>
-
</dependencies>
</project>
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQChannel.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQChannel.java
new file mode 100644
index 0000000..c325933
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQChannel.java
@@ -0,0 +1,183 @@
+/*
+ * 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.dubbo.rpc.rocketmq;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.remoting.Channel;
+import org.apache.dubbo.remoting.ChannelHandler;
+import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.buffer.ChannelBuffer;
+import org.apache.dubbo.remoting.buffer.DynamicChannelBuffer;
+import org.apache.dubbo.rpc.rocketmq.codec.RocketMQCountCodec;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.utils.MessageUtil;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+
+public class RocketMQChannel implements Channel {
+
+ protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+ private final Map<String, Object> attributes = new
ConcurrentHashMap<String, Object>();
+
+ private RocketMQCountCodec rocketmqCountCodec;
+
+ private DefaultMQProducer defaultMQProducer;
+
+ private MessageExt messageExt;
+
+ private String urlString;
+
+ private URL url;
+
+ private InetSocketAddress remoteAddress;
+
+ public RocketMQChannel() {
+
+ }
+
+ public void setRocketMQCountCodec(RocketMQCountCodec rocketmqCountCodec) {
+ this.rocketmqCountCodec = rocketmqCountCodec;
+ }
+
+ public void setDefaultMQProducer(DefaultMQProducer defaultMQProducer) {
+ this.defaultMQProducer = defaultMQProducer;
+ }
+
+ public void setMessageExt(MessageExt messageExt) {
+ this.messageExt = messageExt;
+ }
+
+
+ public void setUrlString(String urlString) {
+ this.urlString = urlString;
+ }
+
+ @Override
+ public URL getUrl() {
+ return url;
+ }
+
+ public void setUrl(URL url) {
+ this.url = url;
+ }
+
+ @Override
+ public ChannelHandler getChannelHandler() {
+ return null;
+ }
+
+ @Override
+ public InetSocketAddress getLocalAddress() {
+ return RocketMQProtocolConstant.LOCAL_ADDRESS;
+ }
+
+ @Override
+ public void send(Object message) throws RemotingException {
+ this.send(message, false);
+ }
+
+ @Override
+ public void send(Object message, boolean sent) throws RemotingException {
+ ChannelBuffer buffer = new DynamicChannelBuffer(2048);
+ try {
+ rocketmqCountCodec.encode(this, buffer, message);
+ } catch (Exception e) {
+ logger.error(e);
+ }
+ try {
+ Message newMessage = MessageUtil.createReplyMessage(messageExt,
buffer.array());
+ newMessage.putUserProperty(RocketMQProtocolConstant.SEND_ADDRESS,
RocketMQProtocolConstant.LOCAL_ADDRESS.getHostString());
+ newMessage.putUserProperty(RocketMQProtocolConstant.URL_STRING,
urlString);
+ SendResult sendResult = defaultMQProducer.send(newMessage, 3000);
+ if (logger.isDebugEnabled()) {
+ logger.debug(String.format("send result is : %s", sendResult));
+ }
+ } catch (Exception e) {
+ logger.error(e);
+ }
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public void close(int timeout) {
+
+ }
+
+ @Override
+ public void startClose() {
+
+ }
+
+ @Override
+ public boolean isClosed() {
+ return false;
+ }
+
+ @Override
+ public InetSocketAddress getRemoteAddress() {
+ return remoteAddress;
+ }
+
+ public void setRemoteAddress(InetSocketAddress remoteAddress) {
+ this.remoteAddress = remoteAddress;
+ }
+
+ @Override
+ public boolean isConnected() {
+ return true;
+ }
+
+ @Override
+ public boolean hasAttribute(String key) {
+ return attributes.containsKey(key);
+ }
+
+ @Override
+ public Object getAttribute(String key) {
+ return attributes.get(key);
+ }
+
+ @Override
+ public void setAttribute(String key, Object value) {
+ if (value == null) {
+ attributes.remove(key);
+ } else {
+ attributes.put(key, value);
+ }
+ }
+
+ @Override
+ public void removeAttribute(String key) {
+ attributes.remove(key);
+ }
+
+
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQExporter.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQExporter.java
new file mode 100644
index 0000000..26b33b8
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQExporter.java
@@ -0,0 +1,76 @@
+/*
+ * 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.dubbo.rpc.rocketmq;
+
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+import static
org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_CATEGORY;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.zip.CRC32;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.Exporter;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.protocol.AbstractExporter;
+
+public class RocketMQExporter<T> extends AbstractExporter<T> {
+
+ private static final String DEFAULT_PARAM_VALUE = "";
+
+ private static final String NAME_SEPARATOR = "_";
+
+ private final String key;
+
+ private final Map<String, Exporter<?>> exporterMap;
+
+ public RocketMQExporter(Invoker<T> invoker, URL url, Map<String,
Exporter<?>> exporterMap) {
+ super(invoker);
+ this.key = toValue(url);
+ this.exporterMap = exporterMap;
+ this.exporterMap.put(key, this);
+ }
+
+ public void afterUnExport() {
+ exporterMap.remove(key, this);
+ }
+
+ public String getKey() {
+ return this.key;
+ }
+
+ private String toValue(URL url) {
+ String serviceInterface = url.getParameter(INTERFACE_KEY);
+ String version = url.getParameter(VERSION_KEY, DEFAULT_PARAM_VALUE);
+ String group = url.getParameter(GROUP_KEY, DEFAULT_PARAM_VALUE);
+
+ String value = null;
+ if (Objects.equals(url.getParameter("groupModel"), "topic")) {
+ value = DEFAULT_CATEGORY + NAME_SEPARATOR + serviceInterface +
NAME_SEPARATOR + version + NAME_SEPARATOR + group;
+ } else {
+ value = DEFAULT_CATEGORY + NAME_SEPARATOR + serviceInterface;
+ }
+ CRC32 crc32 = new CRC32();
+ crc32.update(value.getBytes());
+ value = value.replace(".", "-") + NAME_SEPARATOR +
Long.toString(crc32.getValue());
+ return value;
+ }
+
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQInvoker.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQInvoker.java
new file mode 100644
index 0000000..be76df4
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQInvoker.java
@@ -0,0 +1,232 @@
+/*
+ * 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.dubbo.rpc.rocketmq;
+
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.remoting.Channel;
+import org.apache.dubbo.remoting.buffer.DynamicChannelBuffer;
+import org.apache.dubbo.remoting.buffer.HeapChannelBuffer;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.exchange.Response;
+import org.apache.dubbo.remoting.exchange.support.DefaultFuture;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.AsyncRpcResult;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.TimeoutCountDown;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.protocol.AbstractInvoker;
+import org.apache.dubbo.rpc.rocketmq.codec.RocketMQCountCodec;
+import org.apache.dubbo.rpc.support.RpcUtils;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.RequestCallback;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+
+public class RocketMQInvoker<T> extends AbstractInvoker<T> {
+
+ private final ReentrantLock destroyLock = new ReentrantLock();
+ private final String version;
+ private RocketMQCountCodec rocketMQCountCodec = new
RocketMQCountCodec(FrameworkModel.defaultModel());
+ private DefaultMQProducer defaultMQProducer;
+ private String group;
+
+ private MessageQueue messageQueue;
+
+ private Channel channel = new RocketMQChannel();
+
+ private String topic;
+
+ private String groupModel;
+
+ private Integer timeout;
+
+ public RocketMQInvoker(Class<T> type, URL url, RocketMQProtocolServer
rocketMQProtocolServer) {
+ super(type, url);
+ this.version = url.getParameter(CommonConstants.VERSION_KEY);
+ this.group = url.getParameter(CommonConstants.GROUP_KEY);
+ this.groupModel = url.getParameter("groupModel");
+ this.defaultMQProducer = rocketMQProtocolServer.getDefaultMQProducer();
+ this.topic = url.getParameter("topic");
+ this.timeout = url.getParameter(CommonConstants.TIMEOUT_KEY,
CommonConstants.DEFAULT_TIMEOUT);
+ Integer queueId = url.getParameter("queueId", Integer.class, -1);
+ if (queueId != -1) {
+ messageQueue = new MessageQueue();
+ messageQueue.setBrokerName(url.getParameter("brokerName"));
+ messageQueue.setTopic(this.topic);
+ messageQueue.setQueueId(queueId);
+ }
+ }
+
+
+ @SuppressWarnings("deprecation")
+ @Override
+ protected Result doInvoke(Invocation invocation) throws Throwable {
+ RpcInvocation inv = (RpcInvocation) invocation;
+ final String methodName = RpcUtils.getMethodName(invocation);
+ inv.setAttachment(CommonConstants.PATH_KEY, getUrl().getPath());
+ inv.setAttachment(CommonConstants.VERSION_KEY, version);
+ try {
+
+ RocketMQChannel channel = new RocketMQChannel();
+
+ channel.setUrl(getUrl());
+
+
RpcContext.getContext().setLocalAddress(RocketMQProtocolConstant.LOCAL_ADDRESS);
+
+ boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
+ int timeout = calculateTimeout(invocation, methodName);
+ invocation.put(CommonConstants.TIMEOUT_KEY, timeout);
+
+ Request request = new Request();
+ request.setData(inv);
+ DynamicChannelBuffer buffer = new DynamicChannelBuffer(2048);
+
+ rocketMQCountCodec.encode(channel, buffer, request);
+
+ Message message = new Message(topic, null, buffer.array());
+ //message.putUserProperty(MessageConst.PROPERTY_MESSAGE_TYPE,
"MixAll.REPLY_MESSAGE_FLAG");
+
+ if (!Objects.equals(this.groupModel, "topic")) {
+ message.putUserProperty(CommonConstants.GENERIC_KEY,
this.group);
+ message.putUserProperty(CommonConstants.VERSION_KEY,
this.version);
+ }
+ message.putUserProperty(RocketMQProtocolConstant.SEND_ADDRESS,
NetUtils.getLocalHost());
+ Long messageTimeout = System.currentTimeMillis() + timeout;
+ message.putUserProperty(CommonConstants.TIMEOUT_KEY,
messageTimeout.toString());
+ message.putUserProperty(RocketMQProtocolConstant.URL_STRING,
getUrl().toString());
+ if (isOneway) {
+ if (Objects.isNull(messageQueue)) {
+ defaultMQProducer.sendOneway(message);
+ } else {
+ defaultMQProducer.sendOneway(message, messageQueue);
+ }
+ return AsyncRpcResult.newDefaultAsyncResult(invocation);
+ } else {
+ CompletableFuture<AppResponse> appResponseFuture =
+ DefaultFuture.newFuture(channel, request, timeout,
this.getCallbackExecutor(getUrl(), inv))
+ .thenApply(obj -> (AppResponse) obj);
+ DubboRequestCallback dubboRequestCallback = new
DubboRequestCallback();
+ AsyncRpcResult result = new AsyncRpcResult(appResponseFuture,
inv);
+ if (Objects.isNull(messageQueue)) {
+ defaultMQProducer.request(message, dubboRequestCallback,
timeout);
+ } else {
+ defaultMQProducer.request(message, messageQueue,
dubboRequestCallback, timeout);
+ }
+ return result;
+ }
+ } catch (RemotingTooMuchRequestException e) {
+ String exceptionInfo = "Invoke remote method timeout. method: "
+ + invocation.getMethodName() + ", provider: " + getUrl() + ",
cause: " + e.getMessage();
+ logger.error(exceptionInfo, e);
+ throw new RpcException(RpcException.TIMEOUT_EXCEPTION,
exceptionInfo, e);
+ } catch (Exception e) {
+ String exceptionInfo = "Failed to invoke remote method: "
+ + invocation.getMethodName() + ", provider: " + getUrl() + ",
cause: " + e.getMessage();
+ logger.error(exceptionInfo, e);
+ throw new RpcException(RpcException.NETWORK_EXCEPTION,
exceptionInfo, e);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private int calculateTimeout(Invocation invocation, String methodName) {
+ Object countdown =
RpcContext.getContext().get(CommonConstants.TIME_COUNTDOWN_KEY);
+ int timeout = 1000;
+ if (countdown == null) {
+ timeout = (int) RpcUtils.getTimeout(getUrl(), methodName,
RpcContext.getContext(), this.timeout);
+ if
(getUrl().getParameter(CommonConstants.ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
+
invocation.setObjectAttachment(CommonConstants.TIMEOUT_ATTACHMENT_KEY,
timeout); // pass timeout to remote server
+ }
+ } else {
+ TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countdown;
+ timeout = (int)
timeoutCountDown.timeRemaining(TimeUnit.MILLISECONDS);
+
invocation.setObjectAttachment(CommonConstants.TIMEOUT_ATTACHMENT_KEY,
timeout);// pass timeout to remote server
+ }
+ return timeout;
+ }
+
+ @Override
+ public boolean isAvailable() {
+ if (!super.isAvailable()) {
+ return false;
+ }
+ return true;
+ }
+
+ public void destroy() {
+ if (super.isDestroyed()) {
+ return;
+ }
+ try {
+ destroyLock.lock();
+ if (super.isDestroyed()) {
+ return;
+ }
+ defaultMQProducer.shutdown();
+ } finally {
+ destroyLock.unlock();
+ }
+ }
+
+ class DubboRequestCallback implements RequestCallback {
+ @SuppressWarnings("deprecation")
+ @Override
+ public void onSuccess(Message message) {
+ try {
+
RpcContext.getContext().setRemoteAddress(message.getUserProperty(RocketMQProtocolConstant.SEND_ADDRESS),
9876);
+
+ String urlString =
message.getUserProperty(RocketMQProtocolConstant.URL_STRING);
+ URL url = URL.valueOf(urlString);
+
+ RocketMQChannel channel = new RocketMQChannel();
+
channel.setRemoteAddress(RpcContext.getContext().getRemoteAddress());
+ channel.setUrl(url);
+
+ HeapChannelBuffer heapChannelBuffer = new
HeapChannelBuffer(message.getBody());
+ Object object = (Object) rocketMQCountCodec.decode(channel,
heapChannelBuffer);
+ Response response = (Response) object;
+ DefaultFuture.received(channel, response);
+ } catch (Exception e) {
+ this.onException(e);
+ }
+ }
+
+ @Override
+ public void onException(Throwable e) {
+ Response response = new Response();
+ response.setErrorMessage(e.getMessage());
+ response.setStatus(Response.SERVICE_ERROR);
+ DefaultFuture.received(channel, response);
+ logger.error(e.getMessage(), e);
+ }
+ }
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocol.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocol.java
new file mode 100644
index 0000000..8c86c21
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocol.java
@@ -0,0 +1,266 @@
+/*
+ * 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.dubbo.rpc.rocketmq;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.remoting.buffer.ChannelBuffer;
+import org.apache.dubbo.remoting.buffer.DynamicChannelBuffer;
+import org.apache.dubbo.remoting.buffer.HeapChannelBuffer;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.exchange.Response;
+import org.apache.dubbo.rpc.Exporter;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Protocol;
+import org.apache.dubbo.rpc.ProtocolServer;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.ScopeModel;
+import org.apache.dubbo.rpc.protocol.AbstractProtocol;
+import org.apache.dubbo.rpc.rocketmq.codec.RocketMQCountCodec;
+
+import org.apache.rocketmq.client.common.ClientErrorCode;
+import org.apache.rocketmq.client.consumer.MessageSelector;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.utils.MessageUtil;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+
+public class RocketMQProtocol extends AbstractProtocol {
+
+ public static final String NAME = "rocketmq";
+
+ public static final int DEFAULT_PORT = 20880;
+
+
+ public RocketMQProtocol() {
+ }
+
+
+ public static RocketMQProtocol getDubboProtocol(ScopeModel scopeModel) {
+ return (RocketMQProtocol)
scopeModel.getExtensionLoader(Protocol.class).getExtension(RocketMQProtocol.NAME,
false);
+ }
+
+ /**
+ * <host:port,Exchanger>
+ */
+
+ @Override
+ public int getDefaultPort() {
+ return 9876;
+ }
+
+ @Override
+ public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
+ URL url = invoker.getUrl();
+ RocketMQExporter<T> exporter = new RocketMQExporter<T>(invoker, url,
exporterMap);
+
+ String topic = exporter.getKey();
+ RocketMQProtocolServer rocketMQProtocolServer;
+ try {
+ rocketMQProtocolServer = this.openServer(url,
CommonConstants.PROVIDER);
+ } catch (Exception e) {
+ String exeptionInfo = String.format("create rocketmq client fail,
url is %s , topic is %s, cause is %s", url, topic, e.getMessage());
+ logger.error(exeptionInfo, e);
+ throw new RpcException(exeptionInfo, e);
+ }
+ try {
+ String groupModel = url.getParameter("groupModel");
+ if (Objects.nonNull(groupModel) && Objects.equals(groupModel,
"select")) {
+ if
(Objects.isNull(url.getParameter(CommonConstants.GROUP_KEY)) &&
+
Objects.isNull((url.getParameter(CommonConstants.GROUP_KEY)))) {
+ // error
+ }
+ StringBuffer stringBuffer = new StringBuffer();
+ boolean isGroup = false;
+ if
(Objects.nonNull(url.getParameter(CommonConstants.GROUP_KEY))) {
+
stringBuffer.append(CommonConstants.GROUP_KEY).append("=").append(url.getParameter(CommonConstants.GROUP_KEY));
+ isGroup = true;
+ }
+ if
(Objects.nonNull(url.getParameter(CommonConstants.VERSION_KEY))) {
+ if (isGroup) {
+ stringBuffer.append(" and ");
+ }
+
stringBuffer.append(CommonConstants.VERSION_KEY).append("=").append(url.getParameter(CommonConstants.VERSION_KEY));
+ }
+ MessageSelector messageSelector =
MessageSelector.bySql(stringBuffer.toString());
+
rocketMQProtocolServer.getDefaultMQPushConsumer().subscribe(topic,
messageSelector);
+ } else {
+
rocketMQProtocolServer.getDefaultMQPushConsumer().subscribe(topic,
CommonConstants.ANY_VALUE);
+ }
+ return exporter;
+ } catch (Exception e) {
+ String exeptionInfo = String.format("topic subscirbe fail, topic
is %s, cause is %s", topic, e.getMessage());
+ logger.error(exeptionInfo, e);
+ throw new RpcException(exeptionInfo, e);
+ }
+ }
+
+ private RocketMQProtocolServer openServer(URL url, String model) {
+ // find server.
+ String key = url.getAddress();
+ ProtocolServer server = serverMap.get(key);
+ if (server == null) {
+ synchronized (this) {
+ server = serverMap.get(key);
+ if (server == null) {
+ serverMap.put(key, createServer(url, key, model));
+ }
+ server = serverMap.get(key);
+
+ RocketMQProtocolServer rocketMQProtocolServer =
(RocketMQProtocolServer) server;
+ return rocketMQProtocolServer;
+ }
+ } else {
+ return (RocketMQProtocolServer) server;
+ }
+ }
+
+ private ProtocolServer createServer(URL url, String key, String model) {
+ RocketMQProtocolServer rocketMQProtocolServer = new
RocketMQProtocolServer();
+ rocketMQProtocolServer.setModel(model);
+ DubboMessageListenerConcurrently dubboMessageListenerConcurrently =
new DubboMessageListenerConcurrently();
+
rocketMQProtocolServer.setMessageListenerConcurrently(dubboMessageListenerConcurrently);
+ rocketMQProtocolServer.reset(url);
+ dubboMessageListenerConcurrently.defaultMQProducer =
rocketMQProtocolServer.getDefaultMQProducer();
+ return rocketMQProtocolServer;
+ }
+
+ @Override
+ protected <T> Invoker<T> protocolBindingRefer(Class<T> type, URL url)
throws RpcException {
+ try {
+ RocketMQProtocolServer rocketMQProtocolServer =
this.openServer(url, CommonConstants.CONSUMER);
+ RocketMQInvoker<T> rocketMQInvoker = new RocketMQInvoker<>(type,
url, rocketMQProtocolServer);
+ return rocketMQInvoker;
+ } catch (Exception e) {
+ String exceptionInfo = String.format("protocol binding refer fail,
url is %s , cause is %s ", url, e.getMessage());
+ logger.error(exceptionInfo, e);
+ throw new RpcException(exceptionInfo, e);
+ }
+ }
+
+ private class DubboMessageListenerConcurrently implements
MessageListenerConcurrently {
+
+ private RocketMQCountCodec rocketmqCountCodec = new
RocketMQCountCodec(FrameworkModel.defaultModel());
+
+ private DefaultMQProducer defaultMQProducer;
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
+
+ for (MessageExt messageExt : msgs) {
+ String timeoutString =
messageExt.getUserProperty(CommonConstants.TIMEOUT_KEY);
+ Long timeout = Long.valueOf(timeoutString);
+
+
RpcContext.getContext().setRemoteAddress(messageExt.getUserProperty(RocketMQProtocolConstant.SEND_ADDRESS),
9876);
+ String urlString =
messageExt.getUserProperty(RocketMQProtocolConstant.URL_STRING);
+ URL url = URL.valueOf(urlString);
+
+ RocketMQChannel channel = new RocketMQChannel();
+
channel.setRemoteAddress(RpcContext.getContext().getRemoteAddress());
+ channel.setUrl(url);
+ channel.setUrlString(urlString);
+ channel.setMessageExt(messageExt);
+ channel.setDefaultMQProducer(defaultMQProducer);
+ channel.setRocketMQCountCodec(rocketmqCountCodec);
+
+
+ Response response = new Response();
+ try {
+ if (logger.isDebugEnabled()) {
+ logger.debug(String.format("reply message ext is :
%s", messageExt));
+ }
+ if
(Objects.isNull(messageExt.getProperty(MessageConst.PROPERTY_CLUSTER))) {
+ MQClientException exception = new
MQClientException(ClientErrorCode.CREATE_REPLY_MESSAGE_EXCEPTION,
+ "create reply message fail, requestMessage error,
property[" + MessageConst.PROPERTY_CLUSTER + "] is null.");
+ response.setErrorMessage(exception.getMessage());
+ response.setStatus(Response.BAD_REQUEST);
+ logger.error(exception);
+ } else {
+ HeapChannelBuffer heapChannelBuffer = new
HeapChannelBuffer(messageExt.getBody());
+ Object object = rocketmqCountCodec.decode(channel,
heapChannelBuffer);
+ String topic = messageExt.getTopic();
+ Invocation inv = (Invocation) ((Request)
object).getData();
+ if (timeout < System.currentTimeMillis()) {
+ logger.warn(String.format("message timeoute time
is %d invocation is %s ", timeout, inv));
+ continue;
+ }
+ Invoker<?> invoker =
exporterMap.get(topic).getInvoker();
+
+
RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
+ Result result = invoker.invoke(inv);
+ response.setStatus(Response.OK);
+ response.setResult(result);
+ }
+ } catch (Exception e) {
+ String exceptionInfo = String.format("data decode or
invoke fail, url is %s cause is %s", url, e.getMessage());
+ response.setErrorMessage(exceptionInfo);
+ response.setStatus(Response.BAD_REQUEST);
+ logger.error(exceptionInfo, e);
+ }
+ ChannelBuffer buffer = new DynamicChannelBuffer(2048);
+ try {
+ rocketmqCountCodec.encode(channel, buffer, response);
+ } catch (Exception e) {
+ String exceptionInfo = String.format("encode fail, url is
%s cause is %s", url, e.getMessage());
+ response.setErrorMessage(exceptionInfo);
+ response.setStatus(Response.BAD_REQUEST);
+ logger.error(exceptionInfo, e);
+ try {
+ buffer = new DynamicChannelBuffer(2048);
+ rocketmqCountCodec.encode(channel, buffer, response);
+ } catch (IOException e1) {
+ String exceptionInfo1 = String.format("encode
exception response fail, url is %s cause is %s", url, e.getMessage());
+ logger.error(exceptionInfo1, e1);
+ continue;
+ }
+ }
+ try {
+ Message newMessage =
MessageUtil.createReplyMessage(messageExt, buffer.array());
+
newMessage.putUserProperty(RocketMQProtocolConstant.SEND_ADDRESS,
RocketMQProtocolConstant.LOCAL_ADDRESS.getHostString());
+
newMessage.putUserProperty(RocketMQProtocolConstant.URL_STRING, urlString);
+ SendResult sendResult = defaultMQProducer.send(newMessage,
3000);
+ if (logger.isDebugEnabled()) {
+ logger.debug(String.format("send result is : %s",
sendResult));
+ }
+ } catch (Exception e) {
+ String exceptionInfo = String.format("send response fail,
url is %s cause is %s", url, e.getMessage());
+ logger.error(exceptionInfo, e);
+ }
+ }
+ return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+ }
+
+ }
+
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocolConstant.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocolConstant.java
new file mode 100644
index 0000000..a6c773b
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocolConstant.java
@@ -0,0 +1,38 @@
+/*
+ * 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.dubbo.rpc.rocketmq;
+
+import java.net.InetSocketAddress;
+
+import org.apache.dubbo.common.utils.NetUtils;
+
+public interface RocketMQProtocolConstant {
+
+
+ static final String CONSUMER_CROUP_NAME = "dubbo-roucketmq-consumer-group";
+
+ static final String PRODUCER_CROUP_NAME = "dubbo-roucketmq-producer-group";
+
+ static final String DUBBO_DEFAULT_PROTOCOL_TOPIC =
"dubbo_default_protocol_topic";
+
+ static final String SEND_ADDRESS = "send_address";
+
+ static final String URL_STRING = "url_string";
+
+ static final InetSocketAddress LOCAL_ADDRESS =
InetSocketAddress.createUnresolved(NetUtils.getLocalHost(), 9876);
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocolServer.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocolServer.java
new file mode 100644
index 0000000..d7c5ade
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocolServer.java
@@ -0,0 +1,164 @@
+/*
+ * 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.dubbo.rpc.rocketmq;
+
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.remoting.RemotingServer;
+import org.apache.dubbo.rpc.ProtocolServer;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+
+public class RocketMQProtocolServer implements ProtocolServer {
+
+ private MessageListenerConcurrently messageListenerConcurrently;
+
+ private DefaultMQProducer defaultMQProducer;
+
+ private DefaultMQPushConsumer defaultMQPushConsumer;
+
+ private String address;
+
+ private String namespace;
+
+ private URL url;
+
+ private String model;
+
+ private String inistanceName;
+
+ private String producerGroup;
+
+ private String consumerGroup;
+
+ private boolean enableMsgTrace;
+
+ private String customizedTraceTopic;
+
+ private int sendMsgTimeout;
+
+
+ public void setMessageListenerConcurrently(MessageListenerConcurrently
messageListenerConcurrently) {
+ this.messageListenerConcurrently = messageListenerConcurrently;
+ }
+
+ public void setModel(String model) {
+ this.model = model;
+ }
+
+ @Override
+ public String getAddress() {
+ return this.address;
+ }
+
+ @Override
+ public void setAddress(String address) {
+ this.address = address;
+ }
+
+ @Override
+ public void close() {
+ if (Objects.nonNull(defaultMQProducer)) {
+ defaultMQProducer.shutdown();
+ }
+
+ if (Objects.nonNull(defaultMQPushConsumer)) {
+ defaultMQPushConsumer.shutdown();
+ }
+ }
+
+ public RemotingServer getRemotingServer() {
+ return null;
+ }
+
+ public URL getUrl() {
+ return url;
+ }
+
+ private void getConfig() {
+ this.address = url.getAddress();
+ this.enableMsgTrace = url.getParameter("enableMsgTrace", false);
+ this.namespace = url.getParameter("namespace");
+ this.customizedTraceTopic = url.getParameter("customizedTraceTopic");
+ this.sendMsgTimeout = url.getParameter("timeout", 3000);
+
+ this.inistanceName = url.getParameter("inistanceName", "default-" +
System.currentTimeMillis());
+ this.producerGroup = url.getParameter("producerGroup",
RocketMQProtocolConstant.PRODUCER_CROUP_NAME);
+ this.consumerGroup = url.getParameter("consumerGroup",
RocketMQProtocolConstant.CONSUMER_CROUP_NAME);
+ }
+
+ public synchronized void reset(URL url) {
+ try {
+ this.url = url;
+ this.getConfig();
+
+ DefaultMQProducer defaultMQProducer = new
DefaultMQProducer(this.namespace, this.producerGroup, null, this.enableMsgTrace,
+ customizedTraceTopic);
+ defaultMQProducer.setNamesrvAddr(this.address);
+ defaultMQProducer.setSendMsgTimeout(this.sendMsgTimeout);
+ defaultMQProducer.setInstanceName("producer- " + inistanceName);
+ defaultMQProducer.setSendMsgTimeout(this.sendMsgTimeout);
+
+ defaultMQProducer.start();
+
+
+ this.defaultMQProducer = defaultMQProducer;
+ if (Objects.equals(this.model, CommonConstants.PROVIDER) ||
Objects.equals(this.model, CommonConstants.CALLBACK_INSTANCES_LIMIT_KEY)) {
+ this.createConsumer();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public synchronized void createConsumer() throws MQClientException {
+ DefaultMQPushConsumer defaultMQPushConsumer = new
DefaultMQPushConsumer(this.namespace,
+ this.consumerGroup);
+
+ defaultMQPushConsumer.setNamesrvAddr(this.address);
+ defaultMQPushConsumer.setInstanceName("consumer- " + inistanceName);
+ defaultMQPushConsumer.setConsumeThreadMin(16);
+ defaultMQPushConsumer.setConsumeThreadMax(200);
+
defaultMQPushConsumer.subscribe(RocketMQProtocolConstant.DUBBO_DEFAULT_PROTOCOL_TOPIC,
defaultMQPushConsumer.buildMQClientId());
+
defaultMQPushConsumer.setMessageListener(this.messageListenerConcurrently);
+
+ defaultMQPushConsumer.start();
+
+ this.defaultMQPushConsumer = defaultMQPushConsumer;
+ }
+
+ public DefaultMQProducer getDefaultMQProducer() {
+ return defaultMQProducer;
+ }
+
+ public DefaultMQPushConsumer getDefaultMQPushConsumer() {
+ return defaultMQPushConsumer;
+ }
+
+ @Override
+ public Map<String, Object> getAttributes() {
+ return null;
+ }
+
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/DecodeableRpcInvocation.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/DecodeableRpcInvocation.java
new file mode 100644
index 0000000..cfabf13
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/DecodeableRpcInvocation.java
@@ -0,0 +1,226 @@
+/*
+ * 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.dubbo.rpc.rocketmq.codec;
+
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.serialize.Cleanable;
+import org.apache.dubbo.common.serialize.ObjectInput;
+import org.apache.dubbo.common.utils.Assert;
+import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ReflectUtils;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.remoting.Channel;
+import org.apache.dubbo.remoting.Codec;
+import org.apache.dubbo.remoting.Decodeable;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.transport.CodecSupport;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.FrameworkServiceRepository;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.ModuleModel;
+import org.apache.dubbo.rpc.model.ProviderModel;
+import org.apache.dubbo.rpc.model.ServiceDescriptor;
+import org.apache.dubbo.rpc.support.RpcUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.dubbo.common.BaseServiceMetadata.keyWithoutGroup;
+import static org.apache.dubbo.common.URL.buildKey;
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_VERSION_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+import static org.apache.dubbo.rpc.Constants.SERIALIZATION_ID_KEY;
+import static org.apache.dubbo.rpc.Constants.SERIALIZATION_SECURITY_CHECK_KEY;
+
+@SuppressWarnings({"deprecation", "serial"})
+public class DecodeableRpcInvocation extends RpcInvocation implements Codec,
Decodeable {
+
+ private static final Logger log =
LoggerFactory.getLogger(DecodeableRpcInvocation.class);
+ protected final FrameworkModel frameworkModel;
+ private Channel channel;
+ private byte serializationType;
+ private InputStream inputStream;
+ private Request request;
+ private volatile boolean hasDecoded;
+
+
+ public DecodeableRpcInvocation(FrameworkModel frameworkModel, Channel
channel, Request request, InputStream is, byte id) {
+ this.frameworkModel = frameworkModel;
+ Assert.notNull(channel, "channel == null");
+ Assert.notNull(request, "request == null");
+ Assert.notNull(is, "inputStream == null");
+ this.channel = channel;
+ this.request = request;
+ this.inputStream = is;
+ this.serializationType = id;
+ }
+
+ @Override
+ public void decode() throws Exception {
+ if (!hasDecoded && channel != null && inputStream != null) {
+ try {
+ decode(channel, inputStream);
+ } catch (Throwable e) {
+ if (log.isWarnEnabled()) {
+ log.warn("Decode rpc invocation failed: " +
e.getMessage(), e);
+ }
+ request.setBroken(true);
+ request.setData(e);
+ } finally {
+ hasDecoded = true;
+ }
+ }
+ }
+
+ @Override
+ public void encode(Channel channel, OutputStream output, Object message)
throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+
+ @Override
+ public Object decode(Channel channel, InputStream input) throws
IOException {
+ ObjectInput in = CodecSupport.getSerialization(channel.getUrl(),
serializationType)
+ .deserialize(channel.getUrl(), input);
+ this.put(SERIALIZATION_ID_KEY, serializationType);
+
+ String dubboVersion = in.readUTF();
+ request.setVersion(dubboVersion);
+ setAttachment(DUBBO_VERSION_KEY, dubboVersion);
+
+ String path = in.readUTF();
+ setAttachment(PATH_KEY, path);
+ String version = in.readUTF();
+ setAttachment(VERSION_KEY, version);
+
+ setMethodName(in.readUTF());
+
+ String desc = in.readUTF();
+ setParameterTypesDesc(desc);
+
+ ClassLoader originClassLoader =
Thread.currentThread().getContextClassLoader();
+ try {
+ if
(Boolean.parseBoolean(System.getProperty(SERIALIZATION_SECURITY_CHECK_KEY,
"true"))) {
+
CodecSupport.checkSerialization(frameworkModel.getServiceRepository(), path,
version, serializationType);
+ }
+ Object[] args = RocketMQCodec.EMPTY_OBJECT_ARRAY;
+ Class<?>[] pts = RocketMQCodec.EMPTY_CLASS_ARRAY;
+ if (desc.length() > 0) {
+ FrameworkServiceRepository repository =
frameworkModel.getServiceRepository();
+ List<ProviderModel> providerModels =
repository.lookupExportedServicesWithoutGroup(keyWithoutGroup(path, version));
+ ServiceDescriptor serviceDescriptor = null;
+ if (CollectionUtils.isNotEmpty(providerModels)) {
+ for (ProviderModel providerModel : providerModels) {
+ serviceDescriptor = providerModel.getServiceModel();
+ if (serviceDescriptor != null) {
+ break;
+ }
+ }
+ }
+ if (serviceDescriptor == null) {
+ // Unable to find ProviderModel from Exported Services
+ for (ApplicationModel applicationModel :
frameworkModel.getApplicationModels()) {
+ for (ModuleModel moduleModel :
applicationModel.getModuleModels()) {
+ serviceDescriptor =
moduleModel.getServiceRepository().lookupService(path);
+ if (serviceDescriptor != null) {
+ break;
+ }
+ }
+ }
+ }
+
+ if (serviceDescriptor != null) {
+ MethodDescriptor methodDescriptor =
serviceDescriptor.getMethod(getMethodName(), desc);
+ if (methodDescriptor != null) {
+ pts = methodDescriptor.getParameterClasses();
+ this.setReturnTypes(methodDescriptor.getReturnTypes());
+
+ // switch TCCL
+ if (CollectionUtils.isNotEmpty(providerModels)) {
+ if (providerModels.size() == 1) {
+
Thread.currentThread().setContextClassLoader(providerModels.get(0).getClassLoader());
+ } else {
+ // try all providerModels' classLoader can
load pts, use the first one
+ for (ProviderModel providerModel :
providerModels) {
+ ClassLoader classLoader =
providerModel.getClassLoader();
+ boolean match = true;
+ for (Class<?> pt : pts) {
+ try {
+ if
(!pt.equals(classLoader.loadClass(pt.getName()))) {
+ match = false;
+ }
+ } catch (ClassNotFoundException e) {
+ match = false;
+ }
+ }
+ if (match) {
+
Thread.currentThread().setContextClassLoader(classLoader);
+ break;
+ }
+ }
+ }
+ }
+ }
+ }
+
+ if (pts == RocketMQCodec.EMPTY_CLASS_ARRAY) {
+ if (!RpcUtils.isGenericCall(desc, getMethodName()) &&
!RpcUtils.isEcho(desc, getMethodName())) {
+ throw new IllegalArgumentException("Service not
found:" + path + ", " + getMethodName());
+ }
+ pts = ReflectUtils.desc2classArray(desc);
+ }
+// }
+
+ args = new Object[pts.length];
+ for (int i = 0; i < args.length; i++) {
+ args[i] = in.readObject(pts[i]);
+ }
+ }
+ setParameterTypes(pts);
+
+ Map<String, Object> map = in.readAttachments();
+ if (CollectionUtils.isNotEmptyMap(map)) {
+ addObjectAttachments(map);
+ }
+
+ setArguments(args);
+ String targetServiceName = buildKey(getAttachment(PATH_KEY),
+ getAttachment(GROUP_KEY),
+ getAttachment(VERSION_KEY));
+ setTargetServiceUniqueName(targetServiceName);
+ } catch (ClassNotFoundException e) {
+ throw new IOException(StringUtils.toString("Read invocation data
failed.", e));
+ } finally {
+ Thread.currentThread().setContextClassLoader(originClassLoader);
+ if (in instanceof Cleanable) {
+ ((Cleanable) in).cleanup();
+ }
+ }
+ return this;
+ }
+
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/DecodeableRpcResult.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/DecodeableRpcResult.java
new file mode 100644
index 0000000..a4e5205
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/DecodeableRpcResult.java
@@ -0,0 +1,179 @@
+/*
+ * 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.dubbo.rpc.rocketmq.codec;
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.serialize.Cleanable;
+import org.apache.dubbo.common.serialize.ObjectInput;
+import org.apache.dubbo.common.utils.ArrayUtils;
+import org.apache.dubbo.common.utils.Assert;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.remoting.Channel;
+import org.apache.dubbo.remoting.Codec;
+import org.apache.dubbo.remoting.Decodeable;
+import org.apache.dubbo.remoting.exchange.Response;
+import org.apache.dubbo.remoting.transport.CodecSupport;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.support.RpcUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.reflect.Type;
+
+
+@SuppressWarnings({"deprecation", "serial"})
+public class DecodeableRpcResult extends AppResponse implements Codec,
Decodeable {
+
+ private static final Logger log =
LoggerFactory.getLogger(DecodeableRpcResult.class);
+
+ private Channel channel;
+
+ private byte serializationType;
+
+ private InputStream inputStream;
+
+ private Response response;
+
+ private Invocation invocation;
+
+ private volatile boolean hasDecoded;
+
+ public DecodeableRpcResult(Channel channel, Response response, InputStream
is, Invocation invocation, byte id) {
+ Assert.notNull(channel, "channel == null");
+ Assert.notNull(response, "response == null");
+ Assert.notNull(is, "inputStream == null");
+ this.channel = channel;
+ this.response = response;
+ this.inputStream = is;
+ this.invocation = invocation;
+ this.serializationType = id;
+ }
+
+ @Override
+ public void encode(Channel channel, OutputStream output, Object message)
throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object decode(Channel channel, InputStream input) throws
IOException {
+ if (log.isDebugEnabled()) {
+ Thread thread = Thread.currentThread();
+ log.debug("Decoding in thread -- [" + thread.getName() + "#" +
thread.getId() + "]");
+ }
+
+ // switch TCCL
+ if (invocation != null && invocation.getServiceModel() != null) {
+
Thread.currentThread().setContextClassLoader(invocation.getServiceModel().getClassLoader());
+ }
+ ObjectInput in = CodecSupport.getSerialization(channel.getUrl(),
serializationType)
+ .deserialize(channel.getUrl(), input);
+
+ byte flag = in.readByte();
+ switch (flag) {
+ case RocketMQCodec.RESPONSE_NULL_VALUE:
+ break;
+ case RocketMQCodec.RESPONSE_VALUE:
+ handleValue(in);
+ break;
+ case RocketMQCodec.RESPONSE_WITH_EXCEPTION:
+ handleException(in);
+ break;
+ case RocketMQCodec.RESPONSE_NULL_VALUE_WITH_ATTACHMENTS:
+ handleAttachment(in);
+ break;
+ case RocketMQCodec.RESPONSE_VALUE_WITH_ATTACHMENTS:
+ handleValue(in);
+ handleAttachment(in);
+ break;
+ case RocketMQCodec.RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS:
+ handleException(in);
+ handleAttachment(in);
+ break;
+ default:
+ throw new IOException("Unknown result flag, expect '0' '1' '2'
'3' '4' '5', but received: " + flag);
+ }
+ if (in instanceof Cleanable) {
+ ((Cleanable) in).cleanup();
+ }
+ return this;
+ }
+
+ @Override
+ public void decode() throws Exception {
+ if (!hasDecoded && channel != null && inputStream != null) {
+ try {
+ decode(channel, inputStream);
+ } catch (Throwable e) {
+ if (log.isWarnEnabled()) {
+ log.warn("Decode rpc result failed: " + e.getMessage(), e);
+ }
+ response.setStatus(Response.CLIENT_ERROR);
+ response.setErrorMessage(StringUtils.toString(e));
+ } finally {
+ hasDecoded = true;
+ }
+ }
+ }
+
+ private void handleValue(ObjectInput in) throws IOException {
+ try {
+ Type[] returnTypes;
+ if (invocation instanceof RpcInvocation) {
+ returnTypes = ((RpcInvocation) invocation).getReturnTypes();
+ } else {
+ returnTypes = RpcUtils.getReturnTypes(invocation);
+ }
+ Object value;
+ if (ArrayUtils.isEmpty(returnTypes)) {
+ // This almost never happens?
+ value = in.readObject();
+ } else if (returnTypes.length == 1) {
+ value = in.readObject((Class<?>) returnTypes[0]);
+ } else {
+ value = in.readObject((Class<?>) returnTypes[0],
returnTypes[1]);
+ }
+ setValue(value);
+ } catch (ClassNotFoundException e) {
+ rethrow(e);
+ }
+ }
+
+ private void handleException(ObjectInput in) throws IOException {
+ try {
+ setException(in.readThrowable());
+ } catch (ClassNotFoundException e) {
+ rethrow(e);
+ }
+ }
+
+ private void handleAttachment(ObjectInput in) throws IOException {
+ try {
+ addObjectAttachments(in.readAttachments());
+ } catch (ClassNotFoundException e) {
+ rethrow(e);
+ }
+ }
+
+ private void rethrow(Exception e) throws IOException {
+ throw new IOException(StringUtils.toString("Read response data
failed.", e));
+ }
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCodec.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCodec.java
new file mode 100644
index 0000000..36a509c
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCodec.java
@@ -0,0 +1,230 @@
+/*
+ * 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.dubbo.rpc.rocketmq.codec;
+
+import org.apache.dubbo.common.Version;
+import org.apache.dubbo.common.io.Bytes;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.serialize.ObjectInput;
+import org.apache.dubbo.common.serialize.ObjectOutput;
+import org.apache.dubbo.common.serialize.Serialization;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.remoting.Channel;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.exchange.Response;
+import org.apache.dubbo.remoting.exchange.codec.ExchangeCodec;
+import org.apache.dubbo.remoting.transport.CodecSupport;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_VERSION_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+
+
+/**
+ * Dubbo codec.
+ */
+public class RocketMQCodec extends ExchangeCodec {
+
+ public static final String NAME = "dubbo";
+ public static final String DUBBO_VERSION = Version.getProtocolVersion();
+ public static final byte RESPONSE_WITH_EXCEPTION = 0;
+ public static final byte RESPONSE_VALUE = 1;
+ public static final byte RESPONSE_NULL_VALUE = 2;
+ public static final byte RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS = 3;
+ public static final byte RESPONSE_VALUE_WITH_ATTACHMENTS = 4;
+ public static final byte RESPONSE_NULL_VALUE_WITH_ATTACHMENTS = 5;
+ public static final Object[] EMPTY_OBJECT_ARRAY = new Object[0];
+ public static final Class<?>[] EMPTY_CLASS_ARRAY = new Class<?>[0];
+ private static final Logger log =
LoggerFactory.getLogger(RocketMQCodec.class);
+ private FrameworkModel frameworkModel;
+
+ public RocketMQCodec(FrameworkModel frameworkModel) {
+ this.frameworkModel = frameworkModel;
+ }
+
+ @Override
+ protected Object decodeBody(Channel channel, InputStream is, byte[]
header) throws IOException {
+ byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
+ // get request id.
+ long id = Bytes.bytes2long(header, 4);
+ if ((flag & FLAG_REQUEST) == 0) {
+ // decode response.
+ Response res = new Response(id);
+ if ((flag & FLAG_EVENT) != 0) {
+ res.setEvent(true);
+ }
+ // get status.
+ byte status = header[3];
+ res.setStatus(status);
+ try {
+ if (status == Response.OK) {
+ Object data;
+ if (res.isEvent()) {
+ byte[] eventPayload = CodecSupport.getPayload(is);
+ if (CodecSupport.isHeartBeat(eventPayload, proto)) {
+ // heart beat response data is always null;
+ data = null;
+ } else {
+ ObjectInput in =
CodecSupport.deserialize(channel.getUrl(), new
ByteArrayInputStream(eventPayload), proto);
+ data = decodeEventData(channel, in, eventPayload);
+ }
+ } else {
+ DecodeableRpcResult result = new
DecodeableRpcResult(channel, res, is, (Invocation) getRequestData(id), proto);
+ result.decode();
+ data = result;
+ }
+ res.setResult(data);
+ } else {
+ ObjectInput in =
CodecSupport.deserialize(channel.getUrl(), is, proto);
+ res.setErrorMessage(in.readUTF());
+ }
+ } catch (Throwable t) {
+ if (log.isWarnEnabled()) {
+ log.warn("Decode response failed: " + t.getMessage(), t);
+ }
+ res.setStatus(Response.CLIENT_ERROR);
+ res.setErrorMessage(StringUtils.toString(t));
+ }
+ return res;
+ } else {
+ // decode request.
+ Request req = new Request(id);
+ req.setVersion(Version.getProtocolVersion());
+ req.setTwoWay((flag & FLAG_TWOWAY) != 0);
+ if ((flag & FLAG_EVENT) != 0) {
+ req.setEvent(true);
+ }
+ try {
+ Object data;
+ if (req.isEvent()) {
+ byte[] eventPayload = CodecSupport.getPayload(is);
+ if (CodecSupport.isHeartBeat(eventPayload, proto)) {
+ // heart beat response data is always null;
+ data = null;
+ } else {
+ ObjectInput in =
CodecSupport.deserialize(channel.getUrl(), new
ByteArrayInputStream(eventPayload), proto);
+ data = decodeEventData(channel, in, eventPayload);
+ }
+ } else {
+ DecodeableRpcInvocation inv = new
DecodeableRpcInvocation(frameworkModel, channel, req, is, proto);
+ inv.decode();
+ data = inv;
+ }
+ req.setData(data);
+ } catch (Throwable t) {
+ if (log.isWarnEnabled()) {
+ log.warn("Decode request failed: " + t.getMessage(), t);
+ }
+ // bad request
+ req.setBroken(true);
+ req.setData(t);
+ }
+
+ return req;
+ }
+ }
+
+ @Override
+ protected void encodeRequestData(Channel channel, ObjectOutput out, Object
data) throws IOException {
+ encodeRequestData(channel, out, data, DUBBO_VERSION);
+ }
+
+ @Override
+ protected void encodeResponseData(Channel channel, ObjectOutput out,
Object data) throws IOException {
+ encodeResponseData(channel, out, data, DUBBO_VERSION);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ protected void encodeRequestData(Channel channel, ObjectOutput out, Object
data, String version) throws IOException {
+ RpcInvocation inv = (RpcInvocation) data;
+
+ out.writeUTF(version);
+ // https://github.com/apache/dubbo/issues/6138
+ String serviceName = inv.getAttachment(INTERFACE_KEY);
+ if (serviceName == null) {
+ serviceName = inv.getAttachment(PATH_KEY);
+ }
+ out.writeUTF(serviceName);
+ out.writeUTF(inv.getAttachment(VERSION_KEY));
+
+ out.writeUTF(inv.getMethodName());
+ out.writeUTF(inv.getParameterTypesDesc());
+ Object[] args = inv.getArguments();
+ if (args != null) {
+ for (int i = 0; i < args.length; i++) {
+ out.writeObject(args[i]);
+ }
+ }
+ out.writeAttachments(inv.getObjectAttachments());
+ }
+
+ @Override
+ protected void encodeResponseData(Channel channel, ObjectOutput out,
Object data, String version) throws IOException {
+ Result result = (Result) data;
+ // currently, the version value in Response records the version of
Request
+ boolean attach = Version.isSupportResponseAttachment(version);
+ Throwable th = result.getException();
+ if (th == null) {
+ Object ret = result.getValue();
+ if (ret == null) {
+ out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS :
RESPONSE_NULL_VALUE);
+ } else {
+ out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS :
RESPONSE_VALUE);
+ out.writeObject(ret);
+ }
+ } else {
+ out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS :
RESPONSE_WITH_EXCEPTION);
+ out.writeThrowable(th);
+ }
+
+ if (attach) {
+ // returns current version of Response to consumer side.
+ result.getObjectAttachments().put(DUBBO_VERSION_KEY,
Version.getProtocolVersion());
+ out.writeAttachments(result.getObjectAttachments());
+ }
+ }
+
+ @Override
+ protected Serialization getSerialization(Channel channel, Request req) {
+ if (!(req.getData() instanceof Invocation)) {
+ return super.getSerialization(channel, req);
+ }
+ return RocketMQCodecSupport.getRequestSerialization(channel.getUrl(),
(Invocation) req.getData());
+ }
+
+ @Override
+ protected Serialization getSerialization(Channel channel, Response res) {
+ if (!(res.getResult() instanceof AppResponse)) {
+ return super.getSerialization(channel, res);
+ }
+ return RocketMQCodecSupport.getResponseSerialization(channel.getUrl(),
(AppResponse) res.getResult());
+ }
+
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCodecSupport.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCodecSupport.java
new file mode 100644
index 0000000..79cc2fc
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCodecSupport.java
@@ -0,0 +1,55 @@
+/*
+ * 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.dubbo.rpc.rocketmq.codec;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.serialize.Serialization;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.transport.CodecSupport;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.Invocation;
+
+import static org.apache.dubbo.rpc.Constants.INVOCATION_KEY;
+import static org.apache.dubbo.rpc.Constants.SERIALIZATION_ID_KEY;
+
+public class RocketMQCodecSupport {
+
+ private final static String DEFAULT_REMOTING_SERIALIZATION_PROPERTY =
"hessian2";
+
+ public static Serialization getRequestSerialization(URL url, Invocation
invocation) {
+ Object serializationTypeObj = invocation.get(SERIALIZATION_ID_KEY);
+ if (serializationTypeObj != null) {
+ return CodecSupport.getSerializationById((byte)
serializationTypeObj);
+ }
+ return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
+
url.getParameter(org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY,
DEFAULT_REMOTING_SERIALIZATION_PROPERTY));
+ }
+
+ public static Serialization getResponseSerialization(URL url, AppResponse
appResponse) {
+ Object invocationObj = appResponse.getAttribute(INVOCATION_KEY);
+ if (invocationObj != null) {
+ Invocation invocation = (Invocation) invocationObj;
+ Object serializationTypeObj = invocation.get(SERIALIZATION_ID_KEY);
+ if (serializationTypeObj != null) {
+ return CodecSupport.getSerializationById((byte)
serializationTypeObj);
+ }
+ }
+ return
url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
+ url.getParameter(Constants.SERIALIZATION_KEY,
DEFAULT_REMOTING_SERIALIZATION_PROPERTY));
+ }
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCountCodec.java
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCountCodec.java
new file mode 100644
index 0000000..872b54f
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/java/org/apache/dubbo/rpc/rocketmq/codec/RocketMQCountCodec.java
@@ -0,0 +1,45 @@
+/*
+ * 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.dubbo.rpc.rocketmq.codec;
+
+import org.apache.dubbo.remoting.Channel;
+import org.apache.dubbo.remoting.Codec2;
+import org.apache.dubbo.remoting.buffer.ChannelBuffer;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import java.io.IOException;
+
+
+public final class RocketMQCountCodec implements Codec2 {
+
+ private RocketMQCodec codec;
+
+ public RocketMQCountCodec(FrameworkModel frameworkModel) {
+ codec = new RocketMQCodec(frameworkModel);
+ }
+
+ @Override
+ public void encode(Channel channel, ChannelBuffer buffer, Object msg)
throws IOException {
+ codec.encode(channel, buffer, msg);
+ }
+
+ @Override
+ public Object decode(Channel channel, ChannelBuffer buffer) throws
IOException {
+ return codec.decode(channel, buffer);
+ }
+}
diff --git
a/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol
new file mode 100644
index 0000000..a965f70
--- /dev/null
+++
b/dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol
@@ -0,0 +1 @@
+rocketmq=org.apache.dubbo.rpc.rocketmq.RocketMQProtocol
\ No newline at end of file
diff --git a/dubbo-rpc-extensions/pom.xml b/dubbo-rpc-extensions/pom.xml
index d4a1800..6de59b3 100644
--- a/dubbo-rpc-extensions/pom.xml
+++ b/dubbo-rpc-extensions/pom.xml
@@ -39,6 +39,7 @@
<module>dubbo-rpc-hessian</module>
<module>dubbo-rpc-memcached</module>
<module>dubbo-rpc-redis</module>
+ <module>dubbo-rpc-rocketmq</module>
</modules>
</project>
