feat:add Bidistreaming Api

jraft-core/pom.xml

@@ -127,5 +127,11 @@
             <groupId>org.openjdk.jmh</groupId>
             <artifactId>jmh-generator-annprocess</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-stub</artifactId>
+            <version>1.17.0</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 </project>

jraft-core/src/main/java/com/alipay/sofa/jraft/rpc/RpcClient.java

@@ -21,10 +21,13 @@
 import com.alipay.sofa.jraft.error.RemotingException;
 import com.alipay.sofa.jraft.option.RpcOptions;
 import com.alipay.sofa.jraft.util.Endpoint;
+import com.google.protobuf.Message;
+import io.grpc.stub.StreamObserver;
 
 /**
  *
  * @author jiachun.fjc
+ * @author HH
  */
 public interface RpcClient extends Lifecycle<RpcOptions> {
 
@@ -107,4 +110,31 @@ default void invokeAsync(final Endpoint endpoint, final Object request, final In
      */
     void invokeAsync(final Endpoint endpoint, final Object request, final InvokeContext ctx, final InvokeCallback callback,
                      final long timeoutMs) throws InterruptedException, RemotingException;
+
+    /**
+     * Streaming invocation with a callback.
+     *
+     * @param endpoint  target address
+     * @param request   request object
+     * @param callback  invoke callback
+     * @param timeoutMs timeout millisecond
+     * @return request  stream observer.
+     */
+    default StreamObserver<Message> invokeBidiStreaming(final Endpoint endpoint, final Object request, final InvokeCallback callback,
+            final long timeoutMs) {
+        return invokeBidiStreaming(endpoint, request, null, callback, timeoutMs);
+    }
+
+    /**
+     * BidiStreaming invocation.
+     *
+     * @param endpoint  target address
+     * @param request   request object
+     * @param callback  invoke callback
+     * @param ctx       invoke context
+     * @param timeoutMs timeout millisecond
+     * @return request  stream observer.
+     */
+    StreamObserver<Message> invokeBidiStreaming(final Endpoint endpoint, final Object request, final InvokeContext ctx, final InvokeCallback callback,
+            final long timeoutMs);
 }

jraft-core/src/main/java/com/alipay/sofa/jraft/rpc/RpcServer.java

@@ -22,6 +22,7 @@
 /**
  *
  * @author jiachun.fjc
+ * @author HH
  */
 public interface RpcServer extends Lifecycle<Void> {
 
@@ -39,6 +40,13 @@ public interface RpcServer extends Lifecycle<Void> {
      */
     void registerProcessor(final RpcProcessor<?> processor);
 
+    /**
+     * Register bidiStreaming user processor.
+     *
+     * @param processor the user processor which has a interest
+     */
+    void registerBidiStreamingProcessor(final RpcProcessor<?> processor);
+
     /**
      *
      * @return bound port

jraft-core/src/main/java/com/alipay/sofa/jraft/rpc/impl/BoltRpcClient.java

@@ -32,11 +32,14 @@
 import com.alipay.sofa.jraft.rpc.impl.core.ClientServiceConnectionEventProcessor;
 import com.alipay.sofa.jraft.util.Endpoint;
 import com.alipay.sofa.jraft.util.Requires;
+import com.google.protobuf.Message;
+import io.grpc.stub.StreamObserver;
 
 /**
  * Bolt rpc client impl.
  *
  * @author jiachun.fjc
+ * @author HH
  */
 public class BoltRpcClient implements RpcClient {
 
@@ -119,6 +122,12 @@ public void invokeAsync(final Endpoint endpoint, final Object request, final Inv
         }
     }
 
+    @Override
+    public StreamObserver<Message> invokeBidiStreaming(Endpoint endpoint, Object request, InvokeContext ctx,
+                                                       InvokeCallback callback, long timeoutMs) {
+        return null;
+    }
+
     public com.alipay.remoting.rpc.RpcClient getRpcClient() {
         return rpcClient;
     }

jraft-core/src/main/java/com/alipay/sofa/jraft/rpc/impl/BoltRpcServer.java

@@ -34,6 +34,7 @@
  * Bolt RPC server impl.
  *
  * @author jiachun.fjc
+ * @author HH
  */
 public class BoltRpcServer implements RpcServer {
 
@@ -87,6 +88,11 @@ public void close() {
         });
     }
 
+    @Override
+    public void registerBidiStreamingProcessor(RpcProcessor<?> processor) {
+        //no-op
+    }
+
     @Override
     public int boundPort() {
         return this.rpcServer.port();

jraft-extension/rpc-grpc-impl/src/main/java/com/alipay/sofa/jraft/rpc/impl/GrpcClient.java

@@ -57,6 +57,7 @@
  *
  * @author nicholas.jxf
  * @author jiachun.fjc
+ * @author HH
  */
 public class GrpcClient implements RpcClient {
 
@@ -111,63 +112,112 @@ public void registerConnectEventListener(final ReplicatorGroup replicatorGroup)
     }
 
     @Override
-    public Object invokeSync(final Endpoint endpoint, final Object request, final InvokeContext ctx,
-                             final long timeoutMs) throws RemotingException {
-        final CompletableFuture<Object> future = new CompletableFuture<>();
-
-        invokeAsync(endpoint, request, ctx, (result, err) -> {
-            if (err == null) {
-                future.complete(result);
-            } else {
-                future.completeExceptionally(err);
-            }
-        }, timeoutMs);
-
-        try {
-            return future.get(timeoutMs, TimeUnit.MILLISECONDS);
-        } catch (final TimeoutException e) {
-            future.cancel(true);
-            throw new InvokeTimeoutException(e);
-        } catch (final Throwable t) {
-            future.cancel(true);
-            throw new RemotingException(t);
-        }
-    }
+	public Object invokeSync(final Endpoint endpoint, final Object request, final InvokeContext ctx,
+			final long timeoutMs) throws RemotingException {
+		final CompletableFuture<Object> future = new CompletableFuture<>();
+
+		invokeAsync(endpoint, request, ctx, (result, err) -> {
+			if (err == null) {
+				future.complete(result);
+			}
+			else {
+				future.completeExceptionally(err);
+			}
+		}, timeoutMs);
+
+		try {
+			return future.get(timeoutMs, TimeUnit.MILLISECONDS);
+		}
+		catch (final TimeoutException e) {
+			future.cancel(true);
+			throw new InvokeTimeoutException(e);
+		}
+		catch (final Throwable t) {
+			future.cancel(true);
+			throw new RemotingException(t);
+		}
+	}
 
     @Override
-    public void invokeAsync(final Endpoint endpoint, final Object request, final InvokeContext ctx,
-                            final InvokeCallback callback, final long timeoutMs) {
-        Requires.requireNonNull(endpoint, "endpoint");
-        Requires.requireNonNull(request, "request");
+	public void invokeAsync(final Endpoint endpoint, final Object request, final InvokeContext ctx,
+			final InvokeCallback callback, final long timeoutMs) {
+		nonNullCheck(endpoint, request);
 
-        final Executor executor = callback.executor() != null ? callback.executor() : DirectExecutor.INSTANCE;
+		final Executor executor = getExecutor(callback);
 
-        final Channel ch = getCheckedChannel(endpoint);
-        if (ch == null) {
-            executor.execute(() -> callback.complete(null, new RemotingException("Fail to connect: " + endpoint)));
-            return;
-        }
+		final Channel ch = getCheckedChannel(endpoint);
+		if (ch == null) {
+			executor.execute(() -> callback.complete(null, new RemotingException("Fail to connect: " + endpoint)));
+			return;
+		}
 
-        final MethodDescriptor<Message, Message> method = getCallMethod(request);
-        final CallOptions callOpts = CallOptions.DEFAULT.withDeadlineAfter(timeoutMs, TimeUnit.MILLISECONDS);
+		final MethodDescriptor<Message, Message> method = getCallMethod(request);
+		final CallOptions callOpts = getCallOpts(timeoutMs);
 
-        ClientCalls.asyncUnaryCall(ch.newCall(method, callOpts), (Message) request, new StreamObserver<Message>() {
+		ClientCalls.asyncUnaryCall(ch.newCall(method, callOpts), (Message) request, new StreamObserver<Message>() {
 
-            @Override
-            public void onNext(final Message value) {
-                executor.execute(() -> callback.complete(value, null));
-            }
+			@Override
+			public void onNext(final Message value) {
+				executor.execute(() -> callback.complete(value, null));
+			}
 
-            @Override
-            public void onError(final Throwable throwable) {
-                executor.execute(() -> callback.complete(null, throwable));
-            }
+			@Override
+			public void onError(final Throwable throwable) {
+				executor.execute(() -> callback.complete(null, throwable));
+			}
 
-            @Override
-            public void onCompleted() {
-                // NO-OP
-            }
-        });
+			@Override
+			public void onCompleted() {
+				// NO-OP
+			}
+		});
+	}
+
+    @Override
+	public StreamObserver<Message> invokeBidiStreaming(Endpoint endpoint, Object request, InvokeContext ctx, InvokeCallback callback, long timeoutMs) {
+		nonNullCheck(endpoint, request);
+
+		final Executor executor = getExecutor(callback);
+
+		final Channel ch = getCheckedChannel(endpoint);
+		if (ch == null) {
+			executor.execute(() -> callback.complete(null, new RemotingException("Fail to connect: " + endpoint)));
+			return null;
+		}
+
+		final MethodDescriptor<Message, Message> method = getCallMethod(request);
+		final CallOptions callOpts = getCallOpts(timeoutMs);
+
+		return ClientCalls
+				.asyncBidiStreamingCall(ch.newCall(method, callOpts), new StreamObserver<Message>() {
+					@Override
+					public void onNext(final Message value) {
+						executor.execute(() -> callback.complete(value, null));
+					}
+
+					@Override
+					public void onError(final Throwable throwable) {
+						executor.execute(() -> callback.complete(null, throwable));
+					}
+
+					@Override
+					public void onCompleted() {
+						// NO-OP
+					}
+				});
+	}
+
+    private void nonNullCheck(final Endpoint endpoint, final Object request) {
+        Requires.requireNonNull(endpoint, "endpoint");
+        Requires.requireNonNull(request, "request");
+    }
+
+    private Executor getExecutor(final InvokeCallback callback) {
+        return callback.executor() != null ? callback.executor() : DirectExecutor.INSTANCE;
+    }
+
+    private CallOptions getCallOpts(final long timeoutMs) {
+        return CallOptions.DEFAULT.withDeadlineAfter(timeoutMs, TimeUnit.MILLISECONDS);
     }
 
     private MethodDescriptor<Message, Message> getCallMethod(final Object request) {
@@ -195,12 +245,13 @@ private ManagedChannel getCheckedChannel(final Endpoint endpoint) {
     }
 
     private ManagedChannel getChannel(final Endpoint endpoint, final boolean createIfAbsent) {
-        if (createIfAbsent) {
-            return this.managedChannelPool.computeIfAbsent(endpoint, this::newChannel);
-        } else {
-            return this.managedChannelPool.get(endpoint);
-        }
-    }
+		if (createIfAbsent) {
+			return this.managedChannelPool.computeIfAbsent(endpoint, this::newChannel);
+		}
+		else {
+			return this.managedChannelPool.get(endpoint);
+		}
+	}
 
     private ManagedChannel newChannel(final Endpoint endpoint) {
         final ManagedChannel ch = ManagedChannelBuilder.forAddress(endpoint.getIp(), endpoint.getPort()) //
@@ -222,8 +273,8 @@ private ManagedChannel removeChannel(final Endpoint endpoint) {
     }
 
     private void notifyWhenStateChanged(final ConnectivityState state, final Endpoint endpoint, final ManagedChannel ch) {
-        ch.notifyWhenStateChanged(state, () -> onStateChanged(endpoint, ch));
-    }
+		ch.notifyWhenStateChanged(state, () -> onStateChanged(endpoint, ch));
+	}
 
     private void onStateChanged(final Endpoint endpoint, final ManagedChannel ch) {
         final ConnectivityState state = ch.getState(false);
@@ -252,27 +303,29 @@ private void onStateChanged(final Endpoint endpoint, final ManagedChannel ch) {
     }
 
     private void notifyReady(final Endpoint endpoint) {
-        LOG.info("The channel {} has successfully established.", endpoint);
-
-        clearConnFailuresCount(endpoint);
-
-        final ReplicatorGroup rpGroup = this.replicatorGroup;
-        if (rpGroup != null) {
-            try {
-                RpcUtils.runInThread(() -> {
-                    final PeerId peer = new PeerId();
-                    if (peer.parse(endpoint.toString())) {
-                        LOG.info("Peer {} is connected.", peer);
-                        rpGroup.checkReplicator(peer, true);
-                    } else {
-                        LOG.error("Fail to parse peer: {}.", endpoint);
-                    }
-                });
-            } catch (final Throwable t) {
-                LOG.error("Fail to check replicator {}.", endpoint, t);
-            }
-        }
-    }
+		LOG.info("The channel {} has successfully established.", endpoint);
+
+		clearConnFailuresCount(endpoint);
+
+		final ReplicatorGroup rpGroup = this.replicatorGroup;
+		if (rpGroup != null) {
+			try {
+				RpcUtils.runInThread(() -> {
+					final PeerId peer = new PeerId();
+					if (peer.parse(endpoint.toString())) {
+						LOG.info("Peer {} is connected.", peer);
+						rpGroup.checkReplicator(peer, true);
+					}
+					else {
+						LOG.error("Fail to parse peer: {}.", endpoint);
+					}
+				});
+			}
+			catch (final Throwable t) {
+				LOG.error("Fail to check replicator {}.", endpoint, t);
+			}
+		}
+	}
 
     private void notifyFailure(final Endpoint endpoint) {
         LOG.warn("There has been some transient failure on this channel {}.", endpoint);
@@ -310,8 +363,8 @@ private boolean checkChannel(final Endpoint endpoint, final boolean createIfAbse
     }
 
     private int incConnFailuresCount(final Endpoint endpoint) {
-        return this.transientFailures.computeIfAbsent(endpoint, ep -> new AtomicInteger()).incrementAndGet();
-    }
+		return this.transientFailures.computeIfAbsent(endpoint, ep -> new AtomicInteger()).incrementAndGet();
+	}
 
     private void clearConnFailuresCount(final Endpoint endpoint) {
         this.transientFailures.remove(endpoint);