From 159e25ddc67ae7408840ee424b7c969dcf54fecd Mon Sep 17 00:00:00 2001
From: huangsimin <474420502@qq.com>
Date: Tue, 26 Jul 2022 17:32:51 +0800
Subject: [PATCH] =?UTF-8?q?TODO:=20=E5=AE=8C=E5=96=84=E9=80=BB=E8=BE=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../dataflow/controller/PacketsProcessor.java |  7 +++-
 .../statemachine/state/StateFactory.java      | 41 +++++++++----------
 2 files changed, 25 insertions(+), 23 deletions(-)

diff --git a/src/main/java/com/yuandian/dataflow/controller/PacketsProcessor.java b/src/main/java/com/yuandian/dataflow/controller/PacketsProcessor.java
index a19e406..f2db24b 100644
--- a/src/main/java/com/yuandian/dataflow/controller/PacketsProcessor.java
+++ b/src/main/java/com/yuandian/dataflow/controller/PacketsProcessor.java
@@ -52,13 +52,14 @@ public class PacketsProcessor implements RpcProcessor<PacketsProcessor.PacketsRe
     @Override
     public void handleRequest(RpcContext rpcCtx, PacketsRequest request) {
             // StateServerFactory.getStateServer().updateFsmStateAsync(s, onCompleted);
-            log.info("request.packets.size(): {}",request.packets.size());
+            
             var resp = new RaftResponse();
             resp.setMsg(rpcCtx.getRemoteAddress());
             resp.setSuccess(true);
  
             var ss = StateServerFactory.getStateServer();
  
+            log.info("{} handler request.packets.size(): {}", ss.getNode().getNodeId().getPeerId() ,request.packets.size());
             ss.useFsmStateNotLock((state)->{
                 
                 var work = state.getWorkers().get( ss.getCluster().getServerId());
@@ -75,7 +76,9 @@ public class PacketsProcessor implements RpcProcessor<PacketsProcessor.PacketsRe
                     @Override
                     public void run(Status status) {
                         log.info("{}", this.getResponse());
-                        rpcCtx.sendResponse(this.getResponse());
+                        var resp = new RaftResponse();
+                        resp.setRedirect(StateServerFactory.getStateServer().getNode().getLeaderId());
+                        rpcCtx.sendResponse(resp);
                     }
                     
                 });
diff --git a/src/main/java/com/yuandian/dataflow/statemachine/state/StateFactory.java b/src/main/java/com/yuandian/dataflow/statemachine/state/StateFactory.java
index d6068a0..32af47f 100644
--- a/src/main/java/com/yuandian/dataflow/statemachine/state/StateFactory.java
+++ b/src/main/java/com/yuandian/dataflow/statemachine/state/StateFactory.java
@@ -80,9 +80,6 @@ public class StateFactory {
                                     ws.setUpdateAt(Instant.now());
                                     ws.setTaskQueueSize(100);
 
-                                    var op = new Operate(OperateType.PUT);
-                                    op.setValue(ws);
-
                                     var request = new PacketsRequest();
                                     for (int i = 0; i < cap; i++) {
                                         var p = Any.pack(
@@ -92,27 +89,29 @@ public class StateFactory {
                                         request.getPackets().add(p);
                                     }
 
-                                    // Operate.CallOperate(op, new OperateClosure() {
-                                    // @Override
-                                    // public void run(Status status) {
-                                    // // TODO Auto-generated method stub
-                                    // log.info("{}", status);
-                                    // }
-                                    // });
+                                    var op = new Operate(OperateType.PUT);
+                                    op.setValue(ws);
+                                    Operate.CallOperate(op, new OperateClosure() {
+                                        @Override
+                                        public void run(Status status) {
 
-                                    try {
-                                        ss.getRpcClient().invokeAsync(peer.getEndpoint(),
-                                                request, new InvokeCallback() {
+                                            log.info("{}", status);
 
-                                                    @Override
-                                                    public void complete(Object result, Throwable err) {
-                                                        log.info("{}", result);
-                                                    }
+                                            try {
+                                                ss.getRpcClient().invokeAsync(peer.getEndpoint(),
+                                                        request, new InvokeCallback() {
 
-                                                }, 5000);
-                                    } catch (InterruptedException | RemotingException e) {
-                                        log.info("error send packets {}", e.toString());
-                                    }
+                                                            @Override
+                                                            public void complete(Object result, Throwable err) {
+                                                                log.info("{}", result);
+                                                            }
+
+                                                        }, 5000);
+                                            } catch (InterruptedException | RemotingException e) {
+                                                log.info("error send packets {}", e.toString());
+                                            }
+                                        }
+                                    });
 
                                 }
                             });