Broadcast mode supports the collection of service response sent by ev…

…ery dubbo provider.

dubbo-cluster/pom.xml

@@ -56,5 +56,11 @@
             <version>${project.parent.version}</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+        </dependency>
+
     </dependencies>
 </project>

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java

@@ -16,18 +16,18 @@
  */
 package org.apache.dubbo.rpc.cluster.support;
 
+import com.google.gson.Gson;
+
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.rpc.Invocation;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.Result;
-import org.apache.dubbo.rpc.RpcContext;
-import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.*;
 import org.apache.dubbo.rpc.cluster.Directory;
 import org.apache.dubbo.rpc.cluster.LoadBalance;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.function.BiConsumer;
 
 /**
  * BroadcastClusterInvoker
@@ -36,6 +36,8 @@ public class BroadcastClusterInvoker<T> extends AbstractClusterInvoker<T> {
 
     private static final Logger logger = LoggerFactory.getLogger(BroadcastClusterInvoker.class);
     private static final String BROADCAST_FAIL_PERCENT_KEY = "broadcast.fail.percent";
+    private static final String BROADCAST_COLLECT_ALL_RESULT_KEY = "broadcast.collect.all.results";
+    private static final String BROADCAST_RESULTS_KEY = "broadcast_results";
     private static final int MAX_BROADCAST_FAIL_PERCENT = 100;
     private static final int MIN_BROADCAST_FAIL_PERCENT = 0;
 
@@ -55,6 +57,7 @@ public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, L
         // 100 means that an exception will be thrown last, and 0 means that as long as an exception occurs, it will be thrown.
         // see https://github.com/apache/dubbo/pull/7174
         int broadcastFailPercent = url.getParameter(BROADCAST_FAIL_PERCENT_KEY, MAX_BROADCAST_FAIL_PERCENT);
+        boolean collectAllResults = url.getParameter(BROADCAST_COLLECT_ALL_RESULT_KEY, false);
 
         if (broadcastFailPercent < MIN_BROADCAST_FAIL_PERCENT || broadcastFailPercent > MAX_BROADCAST_FAIL_PERCENT) {
             logger.info(String.format("The value corresponding to the broadcast.fail.percent parameter must be between 0 and 100. " +
@@ -64,30 +67,56 @@ public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, L
 
         int failThresholdIndex = invokers.size() * broadcastFailPercent / MAX_BROADCAST_FAIL_PERCENT;
         int failIndex = 0;
+        List<BroadcastResult> resultList = new ArrayList<>(invokers.size());
         for (Invoker<T> invoker : invokers) {
+            BroadcastResult br = new BroadcastResult();
             try {
+                br.setIp(invoker.getUrl().getIp());
+                br.setPort(invoker.getUrl().getPort());
                 result = invoker.invoke(invocation);
                 if (null != result && result.hasException()) {
                     Throwable resultException = result.getException();
                     if (null != resultException) {
                         exception = getRpcException(result.getException());
+                        br.setData(exception.getMessage());
+                        br.setExceptionMsg(exception.getMessage());
                         logger.warn(exception.getMessage(), exception);
                         if (failIndex == failThresholdIndex) {
                             break;
                         } else {
                             failIndex++;
                         }
                     }
+                } else {
+                    br.setData(result.getValue());
                 }
             } catch (Throwable e) {
                 exception = getRpcException(e);
+                br.setExceptionMsg(exception.getMessage());
                 logger.warn(exception.getMessage(), exception);
                 if (failIndex == failThresholdIndex) {
                     break;
                 } else {
                     failIndex++;
                 }
             }
+            if (collectAllResults) {
+                resultList.add(br);
+            }
+        }
+
+        if (collectAllResults) {
+            if (null != exception) {
+                result = new AppResponse(invocation) {
+                    @Override
+                    public Result whenCompleteWithContext(BiConsumer<Result, Throwable> fn) {
+                        RpcContext.getServerContext().setAttachment(BROADCAST_RESULTS_KEY, new Gson().toJson(resultList));
+                        return new AppResponse();
+                    }
+                };
+                result.setException(exception);
+                return result;
+            }
         }
 
         if (exception != null) {
@@ -100,7 +129,7 @@ public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, L
             }
             throw exception;
         }
-
+        result.setAttachment(BROADCAST_RESULTS_KEY, new Gson().toJson(resultList));
         return result;
     }
 

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastResult.java

@@ -0,0 +1,67 @@
+/*
+ * 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.cluster.support;
+
+/**
+ * @Author goodjava@qq.com
+ * @Date 2021/4/27 13:30
+ */
+public class BroadcastResult {
+
+
+    private String ip;
+
+    private int port;
+
+    private Object data;
+
+    private String exceptionMsg;
+
+
+    public String getIp() {
+        return ip;
+    }
+
+    public void setIp(String ip) {
+        this.ip = ip;
+    }
+
+    public Object getData() {
+        return data;
+    }
+
+    public void setData(Object data) {
+        this.data = data;
+    }
+
+    public String getExceptionMsg() {
+        return exceptionMsg;
+    }
+
+    public void setExceptionMsg(String exceptionMsg) {
+        this.exceptionMsg = exceptionMsg;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public void setPort(int port) {
+        this.port = port;
+    }
+}