test/dataflow
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

TODO: 设计一个简单易懂api

Browse Source
This commit is contained in:
huangsimin 2022-07-18 18:04:21 +08:00
parent ad78dd2613
commit 36fe0e6428
3 changed files with 4 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/com/yuandian/dataflow/controller/TaskLog.java
View File

@ -22,15 +22,7 @@ public class TaskLog {
@GetMapping(path = "/test")
public ResponseEntity<Response> Processing() throws InterruptedException {
SyncClosure<State> closure = new SyncClosure<State>() {
@Override
public void run(Status status) {
log.error("{} {}",getValue().toString(), getValue().getWorkers().size());
}
};
var state = new State();
closure.setValue(state);
StateServerFactory.getStateServer().useFsmState((fsmState)->{
log.error(fsmState.toString() );
return null;
@ -40,7 +32,7 @@ public class TaskLog {
// state.getWorker().setPeerId(StateServerFactory.getStateServer().getNode().getNodeId().getPeerId());
// state.getWorker().setTaskQueueSize(1);
StateServerFactory.getStateServer().readIndexState(true, closure);
final Response response = new Response();
response.Code = HttpStatus.OK;

4
src/main/java/com/yuandian/dataflow/statemachine/StateMachine.java
View File

@ -97,7 +97,7 @@ public class StateMachine extends StateMachineAdapter {
if (iter.done() != null) {
// This task is applied by this node, get value from closure to avoid additional
// parsing.
var closure = (SyncClosure<State>)iter.done();
var closure = (SyncClosure<State>)iter.done(); // 只支持单一个State. 全状态机只支持一种提交
state = closure.getValue();
log.info("SyncDataClosure(done) taskState:{} leaderTerm:{}",state, this.leaderTerm);
closure.success(state);
@ -109,7 +109,7 @@ public class StateMachine extends StateMachineAdapter {
state = SerializerManager.getSerializer(SerializerManager.Hessian2).deserialize(
data.array(), State.class.getName());
log.info("SyncDataClosure(null) taskState:{} leaderTerm:{}", state, this.leaderTerm);
log.info("SyncDataClosure(null) taskState:{} leaderTerm:{}", state, this.leaderTerm);
} catch (CodecException e) {

1
src/main/java/com/yuandian/dataflow/statemachine/StateServerFactory.java
View File

@ -185,7 +185,6 @@ public class StateServerFactory {
task.setData(ByteBuffer.wrap(SerializerManager.getSerializer(SerializerManager.Hessian2).serialize(state)));
task.setDone(closure); // 确认所有数据 一致, 不需要加锁
StateServerFactory.getStateServer().getNode().apply(task);
} catch (CodecException e) {
String errorMsg = "Fail to encode TaskState";
log.error(errorMsg, e);