### Question:
PubSubRouter { private void sendMessage(Bytes peer, TopicMessage.Builder topicMessageBuilder) { logger.trace("send peer:{}", NodeUtils.getNodeIdString(peer.getKey())); CompletableFuture.runAsync(()-> { try { Map<String, IHaveMessage> iHaveMessageMap = unSendMessage.get(peer); if (iHaveMessageMap != null) { for (Map.Entry<String, IHaveMessage> entry : iHaveMessageMap.entrySet()) { topicMessageBuilder.getControlBuilder().addIhave(entry.getValue()); unSendMessage.remove(peer); } } logger.trace("send message local:{} -> remote:{}", NodeUtils.getNodeIdString(routingTable.getLocalNode().getId()), NodeUtils.getNodeIdString(peer.getKey())); messageRouter.sendRequest(topicMessageBuilder.build(), Collections.singletonList(NodeUtils.toRoutableID(peer, RoutableID.DestinationType.NODEIDTYPE)), MessageRouter.ForwardingOptionType.DIRECT_CONNECTION, false); }catch (Exception e) { logger.error("error:", e); } }); } PubSubRouter(); void setMessageRouter(MessageRouter messageRouter); void setRoutingTable(RoutingTable routingTable); void attach(PubSub pubSub); void removePeer(Bytes peer); void publish(Bytes from, EntryMessage msg); void handleControl(Bytes from, ControlMessage msg); void join(String topic); void leave(String topic); void heartbeat();  }### Answer:
@Test public void testControlMessage() throws InterruptedException { subscribe(); List<TopicMessage> topicMessageList = new ArrayList<>(); for (Bytes nodeId : nodeList) { TopicMessage.Builder topicMessageBuilder = TopicMessage.newBuilder(); topicMessageBuilder.setFromNodeId(ByteString.copyFrom(nodeId.getKey())); EntryMessage.Builder entryMessage = EntryMessage.newBuilder(); entryMessage.setFromNodeId(ByteString.copyFrom(nodeId.getKey())); entryMessage.setKey("123"); entryMessage.setTopic("hello"); entryMessage.setData(ByteString.copyFrom("world".getBytes())); topicMessageBuilder.addPublishedEntry(entryMessage); topicMessageList.add(topicMessageBuilder.build()); } for (TopicMessage topicMessage : topicMessageList) { pubSub.sendMessage(topicMessage, null); } Thread.sleep(2000); Map<RoutableID, List<Message>> messageMap = messageRouter.getMessage(ProtoBufHelper.getFullName(TopicMessage.class)); for (Map.Entry<RoutableID, List<Message>> x: messageMap.entrySet()) { Asserts.check(x.getValue().size() == nodeList.size(), "send request size error, expected:" + nodeList.size()); } }