class 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);
        }

        // determine how many nodes have Site-to-site enabled
        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();
    }

}