resolverserviceimpl.java

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

                    srdiHandlerMeter.messageSentViaWalker(srdi);
                }
            } catch (IOException e) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Failure sending srdi message", e);
                }

                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    srdiHandlerMeter.errorPropagatingMessage();
                }
            }
        } else {
            try {
                boolean success = sendMessage(destPeer, null, handlerName, srdiQueName, srdiQueName,
                        (XMLDocument) srdi.getDocument(MimeMediaType.XMLUTF8),
                        // compression
                        true);

                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    if (success) {
                        srdiHandlerMeter.messageSentViaUnicast(destPeer, srdi);
                    } else {
                        srdiHandlerMeter.errorSendingMessage();
                    }
                }
            } catch (Exception e) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Error in sending srdi message", e);
                }
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    srdiHandlerMeter.errorSendingMessage();
                }
            }
        }
    }

    private void repropagateQuery(Message msg, ResolverQueryMsg query) {
        RendezVousService rendezvous = group.getRendezVousService();

        if ((null != rendezvous) && !group.isRendezvous()) {
            // not a Rendezvous peer? Let someone else forward it.
            return;
        }

        // just in case an excessive of attempt to forward a query
        // hopCount is used to determine forward counts independent of any other TTL
        if (query.getHopCount() > 3) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("discarding ResolverQuery. HopCount exceeded : " + query.getHopCount());
            }

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.propagationQueryDropped(query);
            }
            return;
        }

        XMLDocument asDoc = (XMLDocument) query.getDocument(MimeMediaType.XMLUTF8);
        MessageElement docElem = new TextDocumentMessageElement(outQueName, asDoc, null);

        msg.replaceMessageElement("jxta", docElem);

        // Re-propagate the message.
        // Loop and TTL control is done in demux and propagate(). The TTL
        // below is just a default it will be reduced appropriately.

        try {
            if (null != rendezvous) {
                if (rendezvous.getRendezVousStatus() != RendezVousStatus.ADHOC) {
                    rendezvous.walk(msg, handlerName, outQueName, RendezVousService.DEFAULT_TTL);
                }
                // propagate to local net as well
                rendezvous.propagateToNeighbors(msg, handlerName, outQueName, 2);
            } else {
                endpoint.propagate(msg, handlerName, inQueName);
            }

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.queryPropagatedViaWalker(query);
            }
        } catch (IOException e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Failure propagating query", e);

                if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                    resolverMeter.queryPropagationError(query);
                }
            }
        }
    }

    /**
     * Process a resolver query.
     *
     * @param query   The query.
     * @param srcAddr Who sent the query to us. May not be the same as the
     *                query originator.
     * @return the query id assigned to the query
     */
    private int processQuery(ResolverQueryMsg query, EndpointAddress srcAddr) {
        String queryHandlerName = query.getHandlerName();
        QueryHandler theHandler = getHandler(queryHandlerName);

        if (query.getHopCount() > 2) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Discarding query #" + query.getQueryId() + " hopCount > 2 : " + query.getHopCount());
            }

            // query has been forwarded too many times stop
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
                QueryHandlerMeter queryHandlerMeter = resolverServiceMonitor.getQueryHandlerMeter(queryHandlerName);
                if (queryHandlerMeter != null) {
                    queryHandlerMeter.queryHopCountDropped();
                } else {
                    resolverMeter.invalidQueryDiscarded(srcAddr);
                }
            }
            return ResolverService.OK;
        }

        if (theHandler == null) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Discarding query #" + query.getQueryId() + ", no handler for :" + queryHandlerName);
            }
            // If this peer is a rendezvous peer, it needs to repropagate the query to other rendezvous peer that
            // may have a handler.
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.unknownHandlerForQuery(query);
            }
            return ResolverService.Repropagate;
        }

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Handing query #" + query.getQueryId() + " to : " + queryHandlerName);
        }

        QueryHandlerMeter queryHandlerMeter = null;
        long startTime = 0;
        if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
            startTime = System.currentTimeMillis();
            queryHandlerMeter = resolverServiceMonitor.getQueryHandlerMeter(queryHandlerName);
        }

        try {
            int result;
            if (theHandler instanceof InternalQueryHandler) {
                result = ((InternalQueryHandler) theHandler).processQuery(query, srcAddr);
            } else {
                result = theHandler.processQuery(query);
            }

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                queryHandlerMeter.queryProcessed(query, result, System.currentTimeMillis() - startTime);
            }
            return result;
        } catch (Throwable any) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Uncaught Throwable from handler for : " + queryHandlerName, any);
            }
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                queryHandlerMeter.errorWhileProcessingQuery(query);
            }
            // stop repropagation
            return ResolverService.OK;
        }
    }

    /**
     * Process a resolver response.
     *
     * @param resp    The response.
     * @param srcAddr Who sent the response. May not be the same as the
     *                originator response.
     */
    private void processResponse(ResolverResponseMsg resp, EndpointAddress srcAddr) {
        String handlerName = resp.getHandlerName();

        if (handlerName == null) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Missing handlername in response");
            }

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.invalidResponseDiscarded(srcAddr);
            }
            return;
        }

        QueryHandler theHandler = getHandler(handlerName);
        if (theHandler == null) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("No handler for :" + handlerName);
            }
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.unknownHandlerForResponse(srcAddr, resp);
            }
            return;
        }

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Process response to query #" + resp.getQueryId() + " with " + handlerName);
        }

        QueryHandlerMeter queryHandlerMeter = null;
        long startTime = 0;

        if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
            startTime = System.currentTimeMillis();
            queryHandlerMeter = resolverServiceMonitor.getQueryHandlerMeter(handlerName);
        }

        try {
            if (theHandler instanceof InternalQueryHandler) {
                ((InternalQueryHandler) theHandler).processResponse(resp, srcAddr);
            } else {
                theHandler.processResponse(resp);
            }

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                queryHandlerMeter.responseProcessed(resp, System.currentTimeMillis() - startTime, srcAddr);
            }
        } catch (Throwable all) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Uncaught Throwable from handler for: " + handlerName, all);
            }

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                queryHandlerMeter.errorWhileProcessingResponse(srcAddr);
            }
        }
    }

    /**
     * propagate a response
     *
     * @param response response message to propagate
     */
    private void propagateResponse(ResolverResponseMsg response) {

        Message propagateMsg = new Message();

        String queryHandlerName = response.getHandlerName();
        QueryHandlerMeter queryHandlerMeter = null;

        try {
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
                queryHandlerMeter = resolverServiceMonitor.getQueryHandlerMeter(queryHandlerName);
            }

            XMLDocument responseDoc = (XMLDocument) response.getDocument(MimeMediaType.XMLUTF8);
            MessageElement elemDoc = new TextDocumentMessageElement(inQueName, responseDoc, null);
            propagateMsg.addMessageElement("jxta", elemDoc);
            RendezVousService rendezvous = group.getRendezVousService();

            if (null != rendezvous) {
                rendezvous.walk(propagateMsg, handlerName, inQueName, RendezVousService.DEFAULT_TTL);
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                    queryHandlerMeter.responseSentViaWalker(response);
                }
            } else {
                endpoint.propagate(propagateMsg, handlerName, inQueName);
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                    // FIXME bondolo 20040909 not technically the correct metric
                    queryHandlerMeter.responseSentViaWalker(response);
                }
            }
        } catch (IOException e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "failure during propagateResponse", e);
            }
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                queryHandlerMeter.responsePropagateError();
            }
        }
    }

    /**
     * Process an SRDI message.
     *
     * @param srdimsg The SRDI message.
     * @param srcAddr Who sent the message. May not be the same as the
     *                originator of the message.
     */
    private void processSrdi(ResolverSrdiMsgImpl srdimsg, EndpointAddress srcAddr) {
        String handlerName = srdimsg.getHandlerName();

        if (handlerName == null) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Missing handlername in response");
            }

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.invalidSrdiMessageDiscarded(srcAddr);
            }
            return;
        }

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Processing an SRDI msg for : " + handlerName + " in Group ID:" + group.getPeerGroupID());
        }

        SrdiHandler theHandler = getSrdiHandler(handlerName);
        if (theHandler != null) {
            SrdiHandlerMeter srdiHandlerMeter = null;

            try {
                long startTime = 0;
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
                    startTime = System.currentTimeMillis();
                    srdiHandlerMeter = resolverServiceMonitor.getSrdiHandlerMeter(handlerName);
                }

                theHandler.processSrdi(srdimsg);
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    srdiHandlerMeter.messageProcessed(srdimsg, System.currentTimeMillis() - startTime, srcAddr);
                }
            } catch (Throwable all) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Uncaught Throwable from handler for: " + handlerName, all);
                }

                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    srdiHandlerMeter.errorWhileProcessing(srcAddr);
                }