resolverserviceimpl.java

jxta平台的开发包

                            LOG.debug("Added route for peer " + route.getDestPeerID());
                        }
                    }
                }

                boolean success = sendMessage(destPeer,
                                              handlerName,
                                              inQueName,
                                              inQueName,
                                              (XMLDocument) response.getDocument(MimeMediaType.XMLUTF8),
                                              false);

                if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                    if (success) {
                        queryHandlerMeter.responseSentViaUnicast(destPeer, response);
                    } else {
                        queryHandlerMeter.responseSendError();
                    }
                }
            } catch (Exception e) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Error in sending response", e);
                }
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                    queryHandlerMeter.responseSendError();
                }
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public void sendSrdi(String destPeer, ResolverSrdiMsg srdi) {
        String srdiHandlerName = srdi.getHandlerName();
        SrdiHandlerMeter srdiHandlerMeter = null;

        if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
            srdiHandlerMeter = resolverServiceMonitor.getSrdiHandlerMeter(srdiHandlerName);
        }

        if (destPeer == null) {
            if (rendezvous == null) {
                return;
            }
            Message propagateMsg = new Message();
            try {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                GZIPOutputStream gos = new GZIPOutputStream(baos);

                srdi.getDocument(MimeMediaType.XMLUTF8).sendToStream(gos);
                gos.finish();
                gos.close();
                MessageElement zipElem = new ByteArrayMessageElement(srdiQueName, GZIP_MEDIA_TYPE, baos.toByteArray(), null);

                propagateMsg.addMessageElement("jxta", zipElem);
                rendezvous.walk(propagateMsg, handlerName, srdiQueName, RendezVousService.DEFAULT_TTL);

                // propagate to local net as well
                rendezvous.propagateToNeighbors(propagateMsg, handlerName, srdiQueName, 2);
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    srdiHandlerMeter.messageSentViaWalker(srdi);
                }
            } catch (IOException e) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Failure sending srdi message", e);
                }
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    srdiHandlerMeter.errorPropagatingMessage();
                }
            }
        } else {
            try {
                boolean success = sendMessage(destPeer, 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 (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Error in sending srdi message", e);
                }
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (srdiHandlerMeter != null)) {
                    srdiHandlerMeter.errorSendingMessage();
                }
            }
        }
    }

    private void repropagateQuery(Message msg, ResolverQueryMsg query) {

        if ((null != rendezvous) && !myGroup.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 (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("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) {
                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 (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("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.
     */
    private int processQuery(ResolverQueryMsg query, EndpointAddress srcAddr) {
        String queryHandlerName = query.getHandlerName();

        QueryHandler theHandler = getHandler(queryHandlerName);
        if (query.getHopCount() > 2) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("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 (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("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 (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("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 (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("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 = ((ResolverResponseMsg) resp).getHandlerName();
        if (handlerName == null) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Missing handlername in response");
            }
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.invalidResponseDiscarded(srcAddr);
            }
            return;
        }

        QueryHandler theHandler = getHandler(handlerName);

        if (theHandler == null) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("No handler for :" + handlerName);
            }
            if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverMeter != null)) {
                resolverMeter.unknownHandlerForResponse(srcAddr, resp);
            }
            return;
        }

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("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 (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("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);

            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 (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("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) {