gossipservice.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

        return assignedID;
    }

    /**
     * {@inheritDoc}
     */
    public Advertisement getImplAdvertisement() {
        return implAdv;
    }

    /**
     * {@inheritDoc}
     */
    public void init(PeerGroup group, ID assignedID, Advertisement implAdvertisement) throws PeerGroupException {
        this.group = group;
        this.assignedID = assignedID;
        this.implAdv = (ModuleImplAdvertisement) implAdvertisement;

        // Get our configuration parameters.
        GossipServiceConfigAdv gossipConfig = null;
        ConfigParams confAdv = group.getConfigAdvertisement();

        if (confAdv != null) {
            Advertisement adv = null;

            try {
                XMLDocument configDoc = (XMLDocument) confAdv.getServiceParam(getAssignedID());

                if (null != configDoc) {
                    adv = AdvertisementFactory.newAdvertisement(configDoc);
                }
            } catch (NoSuchElementException failed) {
                //ignored
            }

            if (adv instanceof GossipServiceConfigAdv) {
                gossipConfig = (GossipServiceConfigAdv) adv;
            }
        }

        if (null == gossipConfig) {
            // Make a new advertisement for defaults.
            gossipConfig = (GossipServiceConfigAdv) AdvertisementFactory.newAdvertisement(GossipServiceConfigAdv.getAdvertisementType());
        }

        // If the config has a non-null gossip then use that.
        if (null != gossipConfig.getGossip()) {
            gossip = gossipConfig.getGossip();
        }

        // If the config has a non-null showOwn then use that.
        if (null != gossipConfig.getShowOwn()) {
            showOwn = gossipConfig.getShowOwn();
        }

        if (Logging.SHOW_CONFIG && LOG.isLoggable(Level.CONFIG)) {
            StringBuilder configInfo = new StringBuilder("Configuring Gossip Service : " + assignedID);

            configInfo.append("\n\tImplementation :");
            configInfo.append("\n\t\tModule Spec ID: ").append(implAdv.getModuleSpecID());
            configInfo.append("\n\t\tImpl Description : ").append(implAdv.getDescription());
            configInfo.append("\n\t\tImpl URI : ").append(implAdv.getUri());
            configInfo.append("\n\t\tImpl Code : ").append(implAdv.getCode());

            configInfo.append("\n\tGroup Params :");
            configInfo.append("\n\t\tGroup : ").append(group);
            configInfo.append("\n\t\tPeer ID : ").append(group.getPeerID());

            configInfo.append("\n\tConfiguration :");
            configInfo.append("\n\t\tShow own gossips : ").append(showOwn);
            configInfo.append("\n\t\tGossip : ").append(gossip);

            LOG.config(configInfo.toString());
        }
    }

    /**
     * {@inheritDoc}
     */
    public synchronized int startApp(String[] args) {
        /*  We require the peer group Endpoint service. Since the order in which
         *  services are initialized is random the Endpoint might not yet be
         *  initialized when we are first called. If the Endpoint service is not
         *  available then we tell our caller that we can not yet start. The
         *  peer group implementation will continue to start other services and
         *  call our <tt>startApp()</tt> method again.
         */
        endpoint = group.getEndpointService();

        if (null == endpoint) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Stalled until there is an endpoint service");
            }

            return Module.START_AGAIN_STALLED;
        }

        /*  Register our listener for gossip messages. The registered address is
         *  our assigned ID as a String as the <tt>serviceName</tt> and nothing
         *  as the <tt>serviceParam</tt>.
         */
        boolean registered = endpoint.addIncomingMessageListener(this, getAssignedID().toString(), null);

        if (!registered) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.severe("Failed to regiser endpoint listener.");
            }
            return -1;
        }

        // Create our timer task which will send our gossip messages.
        sendTask = new TimerTask() {

            /**
             * {@inheritDoc}
             */
            public void run() {
                sendGossip();
            }
        };

        // Register the timer task.
        SHARED_TIMER.schedule(sendTask, gossip_interval, gossip_interval);

        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("[" + group + "] Gossip Serivce (" + getAssignedID() + ") started");
        }

        return Module.START_OK;
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void stopApp() {
        /*  We have to assume that <tt>stopApp()</tt> might be called before
         * <tt>startApp()</tt> successfully completes. This means that fields
         * initialized in the <tt>startApp()</tt> method might not be
         * initialized.
         */
        if (null != endpoint) {
            endpoint.removeIncomingMessageListener(getAssignedID().toString(), null);
        }
        endpoint = null;

        // Cancel our sending timer task.
        TimerTask currentTask = sendTask;
        if (null != currentTask) {
            currentTask.cancel();
        }
        sendTask = null;

        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("[" + group + "] Gossip Serivce (" + getAssignedID() + ") stopped.");
        }
    }

    /**
     *  {@inheritDoc}
     */
    public void processIncomingMessage(Message message, EndpointAddress srcAddr, EndpointAddress dstAddr) {
        MessageElement sender = message.getMessageElement(GOSSIP_NAMESPACE, GOSSIP_SENDER_ELEMENT_NAME);
        MessageElement text = message.getMessageElement(GOSSIP_NAMESPACE, GOSSIP_GOSSIP_ELEMENT_NAME);

        // Make sure that the message contains the required elements.
        if ((null == sender) || (null == text)) {
            System.err.println("Someone sent us an incomplete message.");
            return;
        }

        // Check if the message is from ourself and should be ignored.
        if (!showOwn && group.getPeerID().toString().equals(sender.toString())) {
            // It's from ourself and we are configured to ignore it.
            return;
        }

        // Print the message's gossip text along with who it's from.
        System.out.println(sender.toString() + " says : " + text.toString());
    }

    /**
     *  Send a gossip message using the endpoint propagate method.
     */
    public void sendGossip() {
        try {
            EndpointService currentEndpoint = endpoint;

            if (null == currentEndpoint) {
                return;
            }
            
            // Create a new message.
            Message gossipMessage = new Message();

            // Add a "sender" element containing our peer id.
            MessageElement sender = new StringMessageElement(GOSSIP_SENDER_ELEMENT_NAME, group.getPeerID().toString(), null);
            gossipMessage.addMessageElement(GOSSIP_NAMESPACE, sender);

            // Add a "gossip" element containing our gossip text.
            MessageElement text = new StringMessageElement(GOSSIP_GOSSIP_ELEMENT_NAME, gossip, null);
            gossipMessage.addMessageElement(GOSSIP_NAMESPACE, text);

            // Send the message to the network using endpoint propagation.
            currentEndpoint.propagate(gossipMessage, getAssignedID().toString(), null);
        } catch (IOException ex) {
            Logger.getLogger(GossipService.class.getName()).log(Level.SEVERE, "Failed sending gossip message.", ex);
        }
    }
}