Method subscribe

ZezeJava/ZezeJava/src/main/java/Zeze/MQ/MQAgent.java:41–62

(String topic, long sessionId, MQConsumer consumer, HashSet<Connector> managers)

	}

	public void subscribe(String topic, long sessionId, MQConsumer consumer, HashSet<Connector> managers) {
		if (consumers.putIfAbsent(sessionId, consumer) == null) {
			var futures = new ArrayList<Subscribe>();
			for (var manager : managers) {
				var r = new Subscribe();
				r.Argument.setTopic(topic);
				r.Argument.setSessionId(sessionId);
				r.SendForWait(manager.GetReadySocket());
				futures.add(r);
			}
			// await all
			for (var future : futures) {
				assert future.getFuture() != null;
				future.getFuture().await();
			}
			// check all result code
			for (var future : futures) {
				if (future.getResultCode() != 0)
					throw new RuntimeException("subscribe consumer error=" + IModule.getErrorCode(future.getResultCode()));
			}
		}
	}

	public void unsubscribe(MQConsumer consumer, HashSet<Connector> managers) {
		// unsubscribe all