(MQConsumer consumer, HashSet<Connector> managers)
| 62 | } |
| 63 | |
| 64 | public void unsubscribe(MQConsumer consumer, HashSet<Connector> managers) { |
| 65 | // unsubscribe all |
| 66 | var futures = new ArrayList<Unsubscribe>(); |
| 67 | for (var manager : managers) { |
| 68 | var r = new Unsubscribe(); |
| 69 | r.Argument.setTopic(consumer.getTopic()); |
| 70 | r.Argument.setSessionId(consumer.getSessionId()); |
| 71 | r.SendForWait(manager.GetReadySocket()); |
| 72 | futures.add(r); |
| 73 | } |
| 74 | // await all |
| 75 | for (var future : futures) { |
| 76 | assert future.getFuture() != null; |
| 77 | future.getFuture().await(); |
| 78 | } |
| 79 | // check all result code |
| 80 | for (var future : futures) { |
| 81 | if (future.getResultCode() != 0) |
| 82 | throw new RuntimeException("unsubscribe consumer error=" + IModule.getErrorCode(future.getResultCode())); |
| 83 | } |
| 84 | |
| 85 | consumers.remove(consumer.getSessionId()); |
| 86 | } |
| 87 | |
| 88 | @Override |
| 89 | protected long ProcessPushMessageRequest(PushMessage r) { |
no test coverage detected