resolverserviceimpl.java

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

        try {
            // Register Query Listener
            queryListener = new DemuxQuery();
            if (!endpoint.addIncomingMessageListener(queryListener, handlerName, outQueName)) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.severe("Cannot register listener (already registered)");
                }
            }

            // Register Response Listener
            responseListener = new DemuxResponse();
            if (!endpoint.addIncomingMessageListener(responseListener, handlerName, inQueName)) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.severe("Cannot register listener (already registered)");
                }
            }

            // Register SRDI Listener
            srdiListener = new DemuxSrdi();
            if (!endpoint.addIncomingMessageListener(srdiListener, handlerName, srdiQueName)) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.severe("Cannot register listener (already registered)");
                }
            }
        } catch (Exception e) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "failed to add listeners", e);
            }
            return -1;
        }

        synchronized (this) {
            // register our credential listener.
            membership.addPropertyChangeListener(MembershipService.DEFAULT_CREDENTIAL_PROPERTY, membershipCredListener);
            try {
                // set the initial version of the default credential.
                currentCredential = null;
                Credential credential = membership.getDefaultCredential();
                XMLDocument credentialDoc;
                if (null != credential) {
                    credentialDoc = (XMLDocument) credential.getDocument(MimeMediaType.XMLUTF8);
                    currentCredential = new CurrentCredential(credential, credentialDoc);
                }
            } catch (Exception all) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "could not get default credential", all);
                }
            }
        }
        return Module.START_OK;
    }

    /**
     * {@inheritDoc}
     */
    public void stopApp() {
        endpoint.removeIncomingMessageListener(handlerName, outQueName);
        endpoint.removeIncomingMessageListener(handlerName, inQueName);

        if (null != srdiListener) {
            endpoint.removeIncomingMessageListener(handlerName, srdiQueName);
        }

        queryListener = null;
        responseListener = null;
        srdiListener = null;

        membership.removePropertyChangeListener("defaultCredential", membershipCredListener);
        currentCredential = null;

        routeControl = null;
        membership = null;
        group = null;
    }

    /**
     * {@inheritDoc}
     */
    public synchronized ResolverService getInterface() {
        if (resolverInterface == null) {
            resolverInterface = new ResolverServiceInterface(this);
        }
        return resolverInterface;
    }

    /**
     * {@inheritDoc}
     */
    public ModuleImplAdvertisement getImplAdvertisement() {
        return implAdvertisement;
    }

    /**
     * {@inheritDoc}
     */
    public QueryHandler registerHandler(String name, QueryHandler handler) {
        if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
            resolverServiceMonitor.registerQueryHandlerMeter(name);
        }
        return handlers.put(name, handler);
    }

    /**
     * {@inheritDoc}
     */
    public QueryHandler unregisterHandler(String name) {
        if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
            resolverServiceMonitor.unregisterQueryHandlerMeter(name);
        }
        return handlers.remove(name);
    }

    /**
     * given a name returns the query handler associated with it
     *
     * @param name the handler to lookup
     * @return returns the query handler
     */
    public QueryHandler getHandler(String name) {
        return handlers.get(name);
    }

    /**
     * {@inheritDoc}
     */
    public SrdiHandler registerSrdiHandler(String name, SrdiHandler handler) {
        if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
            resolverServiceMonitor.registerSrdiHandlerMeter(name);
        }
        return srdiHandlers.put(name, handler);
    }

    /**
     * {@inheritDoc}
     */
    public SrdiHandler unregisterSrdiHandler(String name) {
        if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
            resolverServiceMonitor.unregisterSrdiHandlerMeter(name);
        }
        return srdiHandlers.remove(name);
    }

    /**
     * given a name returns the srdi handler associated with it
     *
     * @param name the handler to lookup
     * @return returns the SRDI handler
     */
    public SrdiHandler getSrdiHandler(String name) {
        return srdiHandlers.get(name);
    }

    /**
     * {@inheritDoc}
     */
    public void sendQuery(String destPeer, ResolverQueryMsg query) {

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("sending query to resolver handler: " + query.getHandlerName());
        }

        // NOTE: Add route information about the issuing peer, so the
        // resolver query responding peer can respond to the issuer without
        // requiring any route discovery. In most case the responding peer
        // is unlikely to know the route to the query issuer. This is a good
        // optimization for edge peers. This optimzation is much less
        // important for RDV peers as they are more likely to have a route
        // to peers. Also, there is the concern that adding route info
        // in resolver query exchanged between RDV will increase overhead due
        // to the larger amount of information exchanged between RDV.
        // Only update query if the query does not already contain any route
        // information. We are mostly interested in the original src
        // route information.
        if (query.getSrcPeerRoute() == null) {
            if (getRouteControl() != null) {
                // FIXME tra 20031102 Until the new subscription service
                // is implemented, we use the Router Control IOCTL
                RouteAdvertisement route = routeControl.getMyLocalRoute();

                if (route != null) {
                    query.setSrcPeerRoute(route.clone());
                }

                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Sending query with route info to " + destPeer);
                }
            } else {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("No route control--could not set local route on query");
                }
            }
        }

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

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

        if (destPeer == null) {
            try {
                Message queryMsg = new Message();
                XMLDocument asDoc = (XMLDocument) query.getDocument(MimeMediaType.XMLUTF8);
                MessageElement docElem = new TextDocumentMessageElement(outQueName, asDoc, null);
                queryMsg.addMessageElement("jxta", docElem);
                RendezVousService rendezvous = group.getRendezVousService();

                if (null != rendezvous) {
                    if (rendezvous.getRendezVousStatus() != RendezVousStatus.ADHOC) {
                        // Walk the message
                        rendezvous.walk(queryMsg.clone(), handlerName, outQueName, RendezVousService.DEFAULT_TTL);
                    }
                    // propagate to local net as well
                    rendezvous.propagateToNeighbors(queryMsg, handlerName, outQueName, 2);
                } else {
                    endpoint.propagate(queryMsg, handlerName, outQueName);
                }

                if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                    queryHandlerMeter.querySentInGroup(query);
                }
            } catch (IOException e) {
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                    queryHandlerMeter.queryPropagateError();
                }
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Failure during propagate", e);
                }
            }
        } else {
            // unicast instead
            boolean success = sendMessage(destPeer, null, handlerName, outQueName, outQueName,
                    (XMLDocument) query.getDocument(MimeMediaType.XMLUTF8), false);

            if (ResolverMeterBuildSettings.RESOLVER_METERING && (queryHandlerMeter != null)) {
                if (success) {
                    queryHandlerMeter.querySentViaUnicast(destPeer, query);
                } else {
                    queryHandlerMeter.querySendError();
                }
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public void sendResponse(String destPeer, ResolverResponseMsg response) {

        if (destPeer == null) {
            propagateResponse(response);
        } else {
            QueryHandlerMeter queryHandlerMeter = null;
            try {
                if (ResolverMeterBuildSettings.RESOLVER_METERING && (resolverServiceMonitor != null)) {
                    queryHandlerMeter = resolverServiceMonitor.getQueryHandlerMeter(response.getHandlerName());
                }

                // Check if an optional route information is available to send the response
                RouteAdvertisement route = response.getSrcPeerRoute();
                boolean success = sendMessage(destPeer, route, 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 (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "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) {
            RendezVousService rendezvous = group.getRendezVousService();

            if (rendezvous == null) {
                // no rendezvous service, dump it.
                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();
                byte gzipBytes[] = baos.toByteArray();

                MessageElement zipElem = new ByteArrayMessageElement(srdiQueName, GZIP_MEDIA_TYPE, gzipBytes, null);

                propagateMsg.addMessageElement("jxta", zipElem);

                if (rendezvous.getRendezVousStatus() != RendezVousStatus.ADHOC) {
                    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)) {