### Question:
PubSub { public void subscribe(String topic, String serviceName, SubscribeCallback callback) { logger.trace("subscribe:{}, serviceName:{} local:{}", topic, serviceName, NodeUtils.getNodeIdString(routingTable.getLocalNode().getId())); boolean needAnnounce = false; synchronized (mytopics) { Map<String, SubscribeCallback> subscribeCallbackMap = mytopics.get(topic); if (subscribeCallbackMap != null) { if(!subscribeCallbackMap.containsKey(serviceName)) { subscribeCallbackMap.put(serviceName, callback); } } else { subscribeCallbackMap = new HashMap<>(); subscribeCallbackMap.put(serviceName, callback); mytopics.put(topic, subscribeCallbackMap); needAnnounce = true; } } logger.trace("needAnnounce:{}", needAnnounce); if (needAnnounce) { announce(topic, true); pubSubRouter.join(topic); } }  void setRoutingTable(RoutingTable routingTable); void setMessageRouter(MessageRouter messageRouter); void addPeer(ByteString peer); void removePeer(ByteString peer); Map<String, Set<Bytes>> getTopic(); void subscribe(String topic, String serviceName, SubscribeCallback callback); void unSubscribe(String topic, String serviceName); void publish(String topic, byte[] data); static String msgId(EntryMessage entryMessage); void notifyAll(String topic, byte[] data); Set<Bytes> listPeers(String topic); void sendMessage(TopicMessage msg, HeaderHelper header); boolean isSubscribe(String topic); public PubSubRouter pubSubRouter; }### Answer:
@Test public void testSubscribe() throws InterruptedException { createSession(); pubSub.subscribe("hello", "world", null); Thread.sleep(1000); int count = messageRouter.getRequestCount(ProtoBufHelper.getFullName(TopicMessage.class)); Asserts.check( count == nodeList.size(), "send subscribe request count error count:" + count + " , expected:"+nodeList.size()); messageRouter.clearRequest(ProtoBufHelper.getFullName(TopicMessage.class)); removeSession(); }