@Test public void testSendPeerListCluster() throws Exception { final SocketFlowFileServerProtocol protocol = getDefaultSocketFlowFileServerProtocol(); final List<NodeInformation> nodeInfoList = new ArrayList<>(); final ClusterNodeInformation clusterNodeInformation = new ClusterNodeInformation(); clusterNodeInformation.setNodeInformation(nodeInfoList); final Optional<ClusterNodeInformation> clusterNodeInfo = Optional.of(clusterNodeInformation); for (int i = 0; i < 3; i++) { final String siteToSiteHostname = String.format("node%d.example.com", i); final Integer siteToSitePort = 8081; final Integer siteToSiteHttpPort = null; final int apiPort = 8080; final boolean isSiteToSiteSecure = true; final int numOfQueuedFlowFiles = 100 + i; final NodeInformation nodeInformation = new NodeInformation(siteToSiteHostname, siteToSitePort, siteToSiteHttpPort, apiPort, isSiteToSiteSecure, numOfQueuedFlowFiles); nodeInfoList.add(nodeInformation); } final NodeInformation self = nodeInfoList.get(0); final HandshakeProperties handshakeProperties = new HandshakeProperties(); handshakeProperties.setCommsIdentifier("communication-identifier"); handshakeProperties.setTransitUriPrefix("uri-prefix"); final ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); final Peer peer = getDefaultPeer(handshakeProperties, outputStream); protocol.handshake(peer); protocol.sendPeerList(peer, clusterNodeInfo, self); try (final DataInputStream dis = new DataInputStream(new ByteArrayInputStream(outputStream.toByteArray()))) { final Response handshakeResponse = Response.read(dis); assertEquals(ResponseCode.PROPERTIES_OK, handshakeResponse.getCode()); final int numPeers = dis.readInt(); assertEquals(nodeInfoList.size(), numPeers); for (int i = 0; i < nodeInfoList.size(); i++) { final NodeInformation node = nodeInfoList.get(i); assertEquals(node.getSiteToSiteHostname(), dis.readUTF()); assertEquals(node.getSiteToSitePort().intValue(), dis.readInt()); assertEquals(node.isSiteToSiteSecure(), dis.readBoolean()); assertEquals(node.getTotalFlowFiles(), dis.readInt()); } } }
@Override public void sendPeerList( final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInfo, final NodeInformation self) throws IOException { if (!handshakeCompleted) { throw new IllegalStateException("Handshake has not been completed"); } if (shutdown) { throw new IllegalStateException("Protocol is shutdown"); } logger.debug("{} Sending Peer List to {}", this, peer); final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); logger.debug("{} Advertising Remote Input host name {}", this, peer); List<NodeInformation> nodeInfos; if (clusterNodeInfo.isPresent()) { nodeInfos = new ArrayList<>(clusterNodeInfo.get().getNodeInformation()); } else { nodeInfos = Collections.singletonList(self); } int numPeers = 0; for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() != null) { numPeers++; } } dos.writeInt(numPeers); for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() == null) { continue; } if (peerDescriptionModifier != null && peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.RAW)) { final PeerDescription target = new PeerDescription(nodeInfo.getSiteToSiteHostname(), nodeInfo.getSiteToSitePort(), nodeInfo.isSiteToSiteSecure()); final PeerDescription modifiedTarget = peerDescriptionModifier.modify(peer.getDescription(), target, SiteToSiteTransportProtocol.RAW, PeerDescriptionModifier.RequestType.Peers, new HashMap<>()); dos.writeUTF(modifiedTarget.getHostname()); dos.writeInt(modifiedTarget.getPort()); dos.writeBoolean(modifiedTarget.isSecure()); } else { dos.writeUTF(nodeInfo.getSiteToSiteHostname()); dos.writeInt(nodeInfo.getSiteToSitePort()); dos.writeBoolean(nodeInfo.isSiteToSiteSecure()); } dos.writeInt(nodeInfo.getTotalFlowFiles()); } logger.info("Sending list of {} peers back to client {}", numPeers, peer); dos.flush(); }
SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol implements PeerDescriptionModifiable { @Override public void sendPeerList( final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInfo, final NodeInformation self) throws IOException { if (!handshakeCompleted) { throw new IllegalStateException("Handshake has not been completed"); } if (shutdown) { throw new IllegalStateException("Protocol is shutdown"); } logger.debug("{} Sending Peer List to {}", this, peer); final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); logger.debug("{} Advertising Remote Input host name {}", this, peer); List<NodeInformation> nodeInfos; if (clusterNodeInfo.isPresent()) { nodeInfos = new ArrayList<>(clusterNodeInfo.get().getNodeInformation()); } else { nodeInfos = Collections.singletonList(self); } int numPeers = 0; for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() != null) { numPeers++; } } dos.writeInt(numPeers); for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() == null) { continue; } if (peerDescriptionModifier != null && peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.RAW)) { final PeerDescription target = new PeerDescription(nodeInfo.getSiteToSiteHostname(), nodeInfo.getSiteToSitePort(), nodeInfo.isSiteToSiteSecure()); final PeerDescription modifiedTarget = peerDescriptionModifier.modify(peer.getDescription(), target, SiteToSiteTransportProtocol.RAW, PeerDescriptionModifier.RequestType.Peers, new HashMap<>()); dos.writeUTF(modifiedTarget.getHostname()); dos.writeInt(modifiedTarget.getPort()); dos.writeBoolean(modifiedTarget.isSecure()); } else { dos.writeUTF(nodeInfo.getSiteToSiteHostname()); dos.writeInt(nodeInfo.getSiteToSitePort()); dos.writeBoolean(nodeInfo.isSiteToSiteSecure()); } dos.writeInt(nodeInfo.getTotalFlowFiles()); } logger.info("Sending list of {} peers back to client {}", numPeers, peer); dos.flush(); } }
SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol implements PeerDescriptionModifiable { @Override public void sendPeerList( final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInfo, final NodeInformation self) throws IOException { if (!handshakeCompleted) { throw new IllegalStateException("Handshake has not been completed"); } if (shutdown) { throw new IllegalStateException("Protocol is shutdown"); } logger.debug("{} Sending Peer List to {}", this, peer); final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); logger.debug("{} Advertising Remote Input host name {}", this, peer); List<NodeInformation> nodeInfos; if (clusterNodeInfo.isPresent()) { nodeInfos = new ArrayList<>(clusterNodeInfo.get().getNodeInformation()); } else { nodeInfos = Collections.singletonList(self); } int numPeers = 0; for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() != null) { numPeers++; } } dos.writeInt(numPeers); for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() == null) { continue; } if (peerDescriptionModifier != null && peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.RAW)) { final PeerDescription target = new PeerDescription(nodeInfo.getSiteToSiteHostname(), nodeInfo.getSiteToSitePort(), nodeInfo.isSiteToSiteSecure()); final PeerDescription modifiedTarget = peerDescriptionModifier.modify(peer.getDescription(), target, SiteToSiteTransportProtocol.RAW, PeerDescriptionModifier.RequestType.Peers, new HashMap<>()); dos.writeUTF(modifiedTarget.getHostname()); dos.writeInt(modifiedTarget.getPort()); dos.writeBoolean(modifiedTarget.isSecure()); } else { dos.writeUTF(nodeInfo.getSiteToSiteHostname()); dos.writeInt(nodeInfo.getSiteToSitePort()); dos.writeBoolean(nodeInfo.isSiteToSiteSecure()); } dos.writeInt(nodeInfo.getTotalFlowFiles()); } logger.info("Sending list of {} peers back to client {}", numPeers, peer); dos.flush(); }  }
SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol implements PeerDescriptionModifiable { @Override public void sendPeerList( final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInfo, final NodeInformation self) throws IOException { if (!handshakeCompleted) { throw new IllegalStateException("Handshake has not been completed"); } if (shutdown) { throw new IllegalStateException("Protocol is shutdown"); } logger.debug("{} Sending Peer List to {}", this, peer); final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); logger.debug("{} Advertising Remote Input host name {}", this, peer); List<NodeInformation> nodeInfos; if (clusterNodeInfo.isPresent()) { nodeInfos = new ArrayList<>(clusterNodeInfo.get().getNodeInformation()); } else { nodeInfos = Collections.singletonList(self); } int numPeers = 0; for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() != null) { numPeers++; } } dos.writeInt(numPeers); for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() == null) { continue; } if (peerDescriptionModifier != null && peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.RAW)) { final PeerDescription target = new PeerDescription(nodeInfo.getSiteToSiteHostname(), nodeInfo.getSiteToSitePort(), nodeInfo.isSiteToSiteSecure()); final PeerDescription modifiedTarget = peerDescriptionModifier.modify(peer.getDescription(), target, SiteToSiteTransportProtocol.RAW, PeerDescriptionModifier.RequestType.Peers, new HashMap<>()); dos.writeUTF(modifiedTarget.getHostname()); dos.writeInt(modifiedTarget.getPort()); dos.writeBoolean(modifiedTarget.isSecure()); } else { dos.writeUTF(nodeInfo.getSiteToSiteHostname()); dos.writeInt(nodeInfo.getSiteToSitePort()); dos.writeBoolean(nodeInfo.isSiteToSiteSecure()); } dos.writeInt(nodeInfo.getTotalFlowFiles()); } logger.info("Sending list of {} peers back to client {}", numPeers, peer); dos.flush(); }  @Override void setPeerDescriptionModifier(PeerDescriptionModifier modifier); @Override FlowFileCodec negotiateCodec(final Peer peer); @Override RequestType getRequestType(final Peer peer); @Override void sendPeerList(
            final Peer peer,
            final Optional<ClusterNodeInformation> clusterNodeInfo,
            final NodeInformation self); @Override String getResourceName(); @Override VersionNegotiator getVersionNegotiator(); }
SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol implements PeerDescriptionModifiable { @Override public void sendPeerList( final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInfo, final NodeInformation self) throws IOException { if (!handshakeCompleted) { throw new IllegalStateException("Handshake has not been completed"); } if (shutdown) { throw new IllegalStateException("Protocol is shutdown"); } logger.debug("{} Sending Peer List to {}", this, peer); final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); logger.debug("{} Advertising Remote Input host name {}", this, peer); List<NodeInformation> nodeInfos; if (clusterNodeInfo.isPresent()) { nodeInfos = new ArrayList<>(clusterNodeInfo.get().getNodeInformation()); } else { nodeInfos = Collections.singletonList(self); } int numPeers = 0; for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() != null) { numPeers++; } } dos.writeInt(numPeers); for (final NodeInformation nodeInfo : nodeInfos) { if (nodeInfo.getSiteToSitePort() == null) { continue; } if (peerDescriptionModifier != null && peerDescriptionModifier.isModificationNeeded(SiteToSiteTransportProtocol.RAW)) { final PeerDescription target = new PeerDescription(nodeInfo.getSiteToSiteHostname(), nodeInfo.getSiteToSitePort(), nodeInfo.isSiteToSiteSecure()); final PeerDescription modifiedTarget = peerDescriptionModifier.modify(peer.getDescription(), target, SiteToSiteTransportProtocol.RAW, PeerDescriptionModifier.RequestType.Peers, new HashMap<>()); dos.writeUTF(modifiedTarget.getHostname()); dos.writeInt(modifiedTarget.getPort()); dos.writeBoolean(modifiedTarget.isSecure()); } else { dos.writeUTF(nodeInfo.getSiteToSiteHostname()); dos.writeInt(nodeInfo.getSiteToSitePort()); dos.writeBoolean(nodeInfo.isSiteToSiteSecure()); } dos.writeInt(nodeInfo.getTotalFlowFiles()); } logger.info("Sending list of {} peers back to client {}", numPeers, peer); dos.flush(); }  @Override void setPeerDescriptionModifier(PeerDescriptionModifier modifier); @Override FlowFileCodec negotiateCodec(final Peer peer); @Override RequestType getRequestType(final Peer peer); @Override void sendPeerList(
            final Peer peer,
            final Optional<ClusterNodeInformation> clusterNodeInfo,
            final NodeInformation self); @Override String getResourceName(); @Override VersionNegotiator getVersionNegotiator(); static final String RESOURCE_NAME; }