SchedulerHandler.java 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131
  1. package cn.sagacloud.server;
  2. /*
  3. * Author: Jxing
  4. * Create Time: 2019/3/20
  5. */
  6. import cn.sagacloud.pojo.ChannelHandlerContextWrapper;
  7. import cn.sagacloud.pojo.Command;
  8. import cn.sagacloud.pojo.TaskStatus;
  9. import cn.sagacloud.proto.MessageProto;
  10. import cn.sagacloud.proto.MessageUtil;
  11. import cn.sagacloud.utils.CommonUtil;
  12. import io.netty.channel.ChannelHandlerContext;
  13. import io.netty.channel.ChannelInboundHandlerAdapter;
  14. import java.net.SocketAddress;
  15. import java.util.ArrayList;
  16. import java.util.HashMap;
  17. import java.util.Map;
  18. import java.util.concurrent.ConcurrentHashMap;
  19. public class SchedulerHandler extends ChannelInboundHandlerAdapter {
  20. ArrayList<ChannelHandlerContextWrapper> clientList;
  21. private ChannelHandlerContextWrapper ctxw;
  22. // Map<当前状态, Map<可接受的命令, 接受完命令后的状态>>
  23. public static Map<TaskStatus, Map<Command, TaskStatus>> statusTransitionMap = new ConcurrentHashMap<>();
  24. static {
  25. Map<Command, TaskStatus> waitingMap = new HashMap<>();
  26. Map<Command, TaskStatus> sendingMap = new HashMap<>();
  27. Map<Command, TaskStatus> sentMap = new HashMap<>();
  28. Map<Command, TaskStatus> fileDownloadExceptionMap = new HashMap<>();
  29. Map<Command, TaskStatus> commandExceptionMap = new HashMap<>();
  30. statusTransitionMap.put(TaskStatus.Waiting, waitingMap);//CommonUtil.createSet(Command.TaskSuccess));
  31. statusTransitionMap.put(TaskStatus.Sending, sendingMap);//CommonUtil.createSet(Command.AcceptTask, Command.RefuseTask, Command.TaskSuccess));
  32. statusTransitionMap.put(TaskStatus.Sent, sentMap);//CommonUtil.createSet(Command.CommandError, Command.DownloadError, Command.TaskSuccess));
  33. statusTransitionMap.put(TaskStatus.FileDownloadException, fileDownloadExceptionMap);//CommonUtil.createSet(Command.TaskSuccess));
  34. statusTransitionMap.put(TaskStatus.CommandExecuteException, commandExceptionMap);//CommonUtil.createSet(Command.TaskSuccess));
  35. statusTransitionMap.put(TaskStatus.Finished, new HashMap<>());
  36. waitingMap.put(Command.TaskSuccess, TaskStatus.Finished);
  37. waitingMap.put(Command.DownloadError, TaskStatus.FileDownloadException);
  38. waitingMap.put(Command.CommandError, TaskStatus.CommandExecuteException);
  39. waitingMap.put(Command.AcceptTask, TaskStatus.Sent);
  40. sendingMap.put(Command.AcceptTask, TaskStatus.Sent);
  41. sendingMap.put(Command.RefuseTask, TaskStatus.Waiting);
  42. sendingMap.put(Command.TaskSuccess, TaskStatus.Finished);
  43. sendingMap.put(Command.DownloadError, TaskStatus.FileDownloadException);
  44. sendingMap.put(Command.CommandError, TaskStatus.CommandExecuteException);
  45. sentMap.put(Command.TaskSuccess, TaskStatus.Finished);
  46. sentMap.put(Command.DownloadError, TaskStatus.FileDownloadException);
  47. sentMap.put(Command.CommandError, TaskStatus.CommandExecuteException);
  48. fileDownloadExceptionMap.put(Command.TaskSuccess, TaskStatus.Finished);
  49. commandExceptionMap.put(Command.TaskSuccess, TaskStatus.Finished);
  50. }
  51. public SchedulerHandler(ArrayList<ChannelHandlerContextWrapper> clientList) {
  52. this.clientList = clientList;
  53. }
  54. /**
  55. * 刚添加上handler的时候 1
  56. * @param ctx
  57. */
  58. @Override
  59. public void handlerAdded(ChannelHandlerContext ctx) {
  60. SocketAddress addr = ctx.channel().remoteAddress();
  61. System.out.println("client incoming");
  62. }
  63. /**
  64. * 客户端活动状态 2
  65. * @param ctx
  66. */
  67. @Override
  68. public void channelActive(ChannelHandlerContext ctx) {
  69. ctxw = new ChannelHandlerContextWrapper(ctx);
  70. this.clientList.add(ctxw);
  71. //ctx.channel().writeAndFlush("connected !"+System.currentTimeMillis());
  72. }
  73. /**
  74. * 客户端服务器互发消息 3
  75. * @param ctx
  76. * @param msg
  77. * @throws Exception
  78. */
  79. @Override
  80. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  81. try {
  82. // Do something with msg
  83. MessageProto.Message message = (MessageProto.Message)msg;
  84. MessageUtil.printMessage(message);
  85. MessageHandler.handle(message, ctxw);
  86. //ctx.writeAndFlush(myMsg); // 异步
  87. } finally {
  88. // 如果ctx write 或 writeAndFlush过的话, 就不用释放msg
  89. //ReferenceCountUtil.release(msg);
  90. }
  91. }
  92. /**
  93. * 客户端异常原因 4
  94. * @param ctx
  95. * @param cause
  96. */
  97. @Override
  98. public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  99. System.out.println("client exceptionCaught : " + cause.getMessage());
  100. }
  101. /**
  102. * 客户端非活动状态 5
  103. * @param ctx
  104. */
  105. @Override
  106. public void channelInactive(ChannelHandlerContext ctx) {
  107. System.out.println("client channelInactive");
  108. }
  109. /**
  110. * 客户端断开连接 6
  111. * @param ctx
  112. */
  113. @Override
  114. public void handlerRemoved(ChannelHandlerContext ctx) {
  115. this.clientList.remove(ctxw);
  116. }
  117. }