rendezvousserviceprovider.java

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

     * as appropriate. This generally only makes sense for multicast or
     * broadcast scenarios.
     *
     * @param msg          the message to be repropagated.
     * @param propHdr      It's current propagation header.
     * @param serviceName  The destination service.
     * @param serviceParam The destination service parameter.
     */
    protected abstract void repropagate(Message msg, RendezVousPropagateMessage propHdr, String serviceName, String serviceParam);

    public abstract void propagateInGroup(Message msg, String serviceName, String serviceParam, int ttl) throws IOException;

    /**
     * Propagates the message via endpoint propagation (multicast/broadcast) on
     * all Message Transports.
     * <p/>
     * Note: The original msg is not modified and may be reused upon return.
     *
     * @param msg     The message to be propagated.
     * @param propHdr It's current propagation header.
     * @throws java.io.IOException if an io error occurs
     */
    protected void sendToNetwork(Message msg, RendezVousPropagateMessage propHdr) throws IOException {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Endpoint propagating " + msg + " (" + propHdr.getMsgId() + ")");
        }

        rdvService.endpoint.propagate(msg, PropSName, PropPName);
    }

    /**
     * Convenience method for constructing an endpoint address from an id
     *
     * @param destPeer peer id
     * @param serv     the service name (if any)
     * @param parm     the service param (if any)
     * @return endpointAddress for this peer id.
     */
    protected static EndpointAddress mkAddress(String destPeer, String serv, String parm) {
        ID asID;
        try {
            asID = IDFactory.fromURI(new URI(destPeer));
        } catch (URISyntaxException caught) {
            throw new IllegalArgumentException(caught.getMessage());
        }

        return mkAddress(asID, serv, parm);
    }

    /**
     * Convenience method for constructing an endpoint address from an id
     *
     * @param destPeer peer id
     * @param serv     the service name (if any)
     * @param parm     the service param (if any)
     * @return endpointAddress for this peer id.
     */
    protected static EndpointAddress mkAddress(ID destPeer, String serv, String parm) {
        EndpointAddress addr = new EndpointAddress(RDV_MSG_NAMESPACE_NAME, destPeer.getUniqueValue().toString(), serv, parm);
        return addr;
    }

    /**
     * Get propagate header from the message.
     *
     * @param msg The source message.
     * @return The message's propagate header if any, otherwise null.
     */
    protected RendezVousPropagateMessage getPropHeader(Message msg) {

        MessageElement elem = msg.getMessageElement(RDV_MSG_NAMESPACE_NAME, PROP_HDR_ELEMENT_NAME);

        if (elem == null) {
            return null;
        }

        try {
            StructuredDocument asDoc = StructuredDocumentFactory.newStructuredDocument(elem);
            return new RendezVousPropagateMessage(asDoc);
        } catch (IOException failed) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Could not get prop header of " + msg, failed);
            }
            IllegalArgumentException failure = new IllegalArgumentException("Could not get prop header of " + msg);
            failure.initCause(failed);
            throw failure;
        }
    }

    /**
     * Check and updates the header message element
     *
     * @param msg the message to check
     * @return an upadate message
     */
    protected RendezVousPropagateMessage checkPropHeader(Message msg) {
        RendezVousPropagateMessage propHdr;

        try {
            propHdr = getPropHeader(msg);
            if (null == propHdr) {
                // No header. Discard the message
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.warning("Discarding " + msg + " -- missing propagate header.");
                }
                if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                    rendezvousMeter.invalidMessageReceived();
                }
                return null;
            }
        } catch (Exception failure) {
            // Bad header. Discard the message
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Discarding " + msg + " -- bad propagate header.", failure);
            }

            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.invalidMessageReceived();
            }
            return null;
        }
        // Look at the Propagate header if any and check for loops.
        // Do not remove it; we do not have to change it yet, and we have
        // do look at it at different places and looking costs less on
        // incoming elements than on outgoing.

        // TTL detection. A message arriving with TTL <= 0 should not even
        // have been sent. Kill it.

        if (propHdr.getTTL() <= 0) {
            // This message is dead on arrival. Drop it.
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Discarding " + msg + "(" + propHdr.getMsgId() + ") -- dead on arrival (TTl=" + propHdr.getTTL() + ").");
            }

            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.receivedDeadMessage();
            }
            return null;
        }

        if (!rdvService.addMsgId(propHdr.getMsgId())) {
            // We already received this message - discard
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Discarding " + msg + "(" + propHdr.getMsgId() + ") -- feedback.");
            }

            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.receivedDuplicateMessage();
            }
            return null;
        }

        // Loop detection
        if (propHdr.isVisited(group.getPeerID().toURI())) {
            // Loop is detected - discard.
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Discarding " + msg + "(" + propHdr.getMsgId() + ") -- loopback.");
            }
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.receivedLoopbackMessage();
            }
            return null;
        }
        // Message is valid
        return propHdr;
    }

    protected RendezVousPropagateMessage updatePropHeader(Message msg, RendezVousPropagateMessage propHdr, String serviceName, String serviceParam, int initialTTL) {

        boolean newHeader = false;

        if (null == propHdr) {
            propHdr = newPropHeader(serviceName, serviceParam, initialTTL);
            newHeader = true;
        } else {
            if (null == updatePropHeader(propHdr, initialTTL)) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("TTL expired for " + msg + " (" + propHdr.getMsgId() + ") TTL=" + propHdr.getTTL());
                }
                return null;
            }
        }

        XMLDocument propHdrDoc = (XMLDocument) propHdr.getDocument(MimeMediaType.XMLUTF8);
        MessageElement elem = new TextDocumentMessageElement(PROP_HDR_ELEMENT_NAME, propHdrDoc, null);

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine((newHeader ? "Added" : "Updated") + " prop header for " + msg + " (" + propHdr.getMsgId() + ") TTL = "
                    + propHdr.getTTL());
        }

        msg.replaceMessageElement(RDV_MSG_NAMESPACE_NAME, elem);
        return propHdr;
    }

    /**
     * Adds a propagation header to the given message with the given default
     * TTL. Also adds this peer to the path recorded in the message.
     *
     * @param serviceName the service name
     * @param serviceParam the parameter
     * @param initialTTL initial TTL
     * @return  a updated message with the proper TTL and ID
     */
    private RendezVousPropagateMessage newPropHeader(String serviceName, String serviceParam, int initialTTL) {

        RendezVousPropagateMessage propHdr = new RendezVousPropagateMessage();

        propHdr.setTTL(initialTTL);
        propHdr.setDestSName(serviceName);
        propHdr.setDestSParam(serviceParam);

        UUID msgID = rdvService.createMsgId();

        propHdr.setMsgId(msgID);
        rdvService.addMsgId(msgID);

        // Add this peer to the path.
        propHdr.addVisited(group.getPeerID().toURI());

        return propHdr;
    }

    /**
     * Updates the propagation header of the message. Also adds this peer to the
     * path recorded in the message. Returns true if the message should be
     * repropagated, false otherwise.
     *
     * @param propHdr The propHdr for the message.
     * @param maxTTL  The maximum TTL which will be allowed.
     * @return The updated propagate header if the message should be
     *         repropagated otherwise null.
     */
    private RendezVousPropagateMessage updatePropHeader(RendezVousPropagateMessage propHdr, int maxTTL) {

        int msgTTL = propHdr.getTTL();
        URI me = group.getPeerID().toURI();

        int useTTL = msgTTL;

        if (!propHdr.isVisited(me)) {
            // only reduce TTL if message has not previously visited us.
            useTTL--;
        }

        // ensure TTL does not exceed maxTTL
        useTTL = Math.min(useTTL, maxTTL);

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Updating propagation header (" + propHdr.getMsgId() + ") TTL: " + msgTTL + "-->" + useTTL);
        }

        propHdr.setTTL(useTTL);

        // Add this peer to the path.
        propHdr.addVisited(me);

        // If message came in with TTL one or less, it was last trip. It can not go any further.
        return (useTTL <= 0) ? null : propHdr;
    }
}