### Question:
PubSub { public void sendMessage(TopicMessage msg, HeaderHelper header){ logger.trace("receive topic message from:{} has Subscribe:{}", NodeUtils.getNodeIdString(msg.getFromNodeId()), msg.hasSubscribe()); if (msg.hasSubscribe()) { logger.trace("announce msg:{} ", msg.getSubscribe().getSub()); if (msg.getSubscribe().getSub()){ synchronized (topics) { Set<Bytes> nodeIds = topics.get(msg.getSubscribe().getTopic()); if (nodeIds == null) { nodeIds = new ConcurrentSet<>(); } nodeIds.add(Bytes.valueOf(msg.getSubscribe().getNodeId())); topics.put(msg.getSubscribe().getTopic(), nodeIds); } } else { Set<Bytes> nodeIds = topics.get(msg.getSubscribe().getTopic()); if (nodeIds != null) { nodeIds.remove(msg.getSubscribe().getNodeId()); } } } if (msg.getPublishedEntryCount() != 0) { for (EntryMessage entryMessage : msg.getPublishedEntryList()){ logger.trace("receive remote:{} have publish message:{}", NodeUtils.getNodeIdString(entryMessage.getFromNodeId()), entryMessage.getKey()); pushMsg(entryMessage); } } pubSubRouter.handleControl(Bytes.valueOf(msg.getFromNodeId()), msg.getControl()); }  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 testSendMessageSubscribe() { createSession(); List<TopicMessage> topicMessageList = new ArrayList<>(); for (Bytes nodeID : nodeList) { TopicMessage.Builder topicMessageBuilder = TopicMessage.newBuilder(); topicMessageBuilder.setFromNodeId(ByteString.copyFrom(nodeID.getKey())); SubMessage.Builder subMessageBuilder = SubMessage.newBuilder(); subMessageBuilder.setTopic("hello"); subMessageBuilder.setSub(true); subMessageBuilder.setNodeId(ByteString.copyFrom(nodeID.getKey())); topicMessageBuilder.setSubscribe(subMessageBuilder); topicMessageList.add(topicMessageBuilder.build()); } for (TopicMessage topicMessage : topicMessageList) { pubSub.sendMessage(topicMessage, null); } Set<Bytes> nodeIDSet = topicsMock.get("hello"); for (Bytes nodeId : nodeList) { Asserts.check(nodeIDSet.contains(nodeId), "topics error expected:" + nodeId); } }