proxyservice.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

    }

    private PeerGroupAdvertisement createGroupAdvertisement(String name, String id) {
        PeerGroupAdvertisement adv = null;

        PeerGroupID gid = null;
        if (id != null) {
            try {
                ID tempId = IDFactory.fromURI(new URI(id));
                gid = (PeerGroupID)tempId;
            } catch (URISyntaxException e) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Invalid peergroupId", e);
                }
            } catch (ClassCastException e) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("id was not a peergroup id", e);
                }
            }
        }

        if (gid == null) {
            gid = IDFactory.newPeerGroupID();
        }

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("newPeerGroupAdvertisement name="+name+" id="+gid.toString());
        }

        adv = (PeerGroupAdvertisement) group.getPeerGroupAdvertisement().clone();

        try {
            // Create a PeerGroup Advertisement for this pipe.
            adv = (PeerGroupAdvertisement) AdvertisementFactory.newAdvertisement(
                      PeerGroupAdvertisement.getAdvertisementType());
            adv.setName(name);
            adv.setPeerGroupID(gid);
            adv.setModuleSpecID(PeerGroup.allPurposePeerGroupSpecID);
            adv.setDescription("PeerGroup Advertisement created for a jxme device");
            ModuleImplAdvertisement groupImplAdv = group.getAllPurposePeerGroupImplAdvertisement();
            discovery.publish(groupImplAdv);
            discovery.publish(adv);
        } catch (Exception e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("newPeerGroupAdvertisement Exception", e);
            }
        }

        return adv;
    }

    private PipeAdvertisement createPipeAdvertisement(String pipeName, String pipeId, String pipeType) {
        PipeAdvertisement adv = null;

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("newPipeAdvertisement name="+pipeName+" pipeId="+pipeId+" pipeType="+pipeType);
        }

        if (pipeType == null || pipeType.length() == 0) {
            pipeType = PipeService.UnicastType;
        }

        if (pipeId ==  null) {
            pipeId = IDFactory.newPipeID(group.getPeerGroupID()).toString();
        }

        try {
            // Create a pipe advertisement for this pipe.
            adv = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(
                      PipeAdvertisement.getAdvertisementType());

            adv.setName(pipeName);
            adv.setPipeID(IDFactory.fromURI(new URI(pipeId)));
            adv.setType(pipeType);
        } catch (Exception e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("newPipeAdvertisement Exception", e);
            }
        }

        return adv;
    }

    private PipeAdvertisement findPipeAdvertisement(String name, String id, String arg) {
        String attribute, value;

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("findPipeAdvertisement name=" + name +
                      " id=" + id + " arg=" + arg);
        }

        if (id != null) {
            attribute = PipeAdvertisement.IdTag;
            value = id;
        } else if (name != null) {
            attribute = PipeAdvertisement.NameTag;
            value = name;
        } else {
            // the id or the name must be specified
            return null;
        }

        if (arg == null) {
            // the default pipe type
            arg = PipeService.UnicastType;
        }

        Enumeration each = null;
        try {
            each = discovery.getLocalAdvertisements(DiscoveryService.ADV,
                                                    attribute,
                                                    value);
        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("IOException in getLocalAdvertisements()", e);
            }
            return null;
        }

        PipeAdvertisement pipeAdv = null;
        Advertisement adv = null;
        while (each != null && each.hasMoreElements()) {
            adv = (Advertisement) each.nextElement();

            // take the first match
            if (adv instanceof PipeAdvertisement) {
                pipeAdv = (PipeAdvertisement)adv;
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("found PipeAdvertisement = " + pipeAdv);
                }
                break;
            }
        }

        return pipeAdv;
    }

    public synchronized void discoveryEvent(DiscoveryEvent event) {

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("discoveryEvent " + event);
        }

        Requestor requestor;

        requestor = (Requestor) searchRequests.get(new Integer(event.getQueryID()));

        if (requestor == null) {
            return;
        }

        DiscoveryResponseMsg response = event.getResponse();
        if (response == null) {
            return;
        }

        Enumeration each = response.getResponses();
        if (each == null || !each.hasMoreElements()) {
            return;
        }
        // we have a response remove it from the LRUCache
        searchRequests.remove(new Integer(event.getQueryID()));
        int i=0;
        while (each.hasMoreElements() && i < requestor.getThreshold()) {
            try {
                String str = (String) each.nextElement();

                // Create Advertisement from response.
                Advertisement adv = (Advertisement)
                                    AdvertisementFactory.newAdvertisement(MimeMediaType.XMLUTF8,
                                                                          new StringReader(str));

                // notify the requestor of the result
                requestor.send(adv, RESPONSE_RESULT);
            } catch (Exception e) {
                // this should not happen unless a bad result is returned
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Bad result returned by DiscoveryService", e);
                }
            }
        }
    }

    public synchronized void pipeMsgEvent(PipeMsgEvent event) {
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("pipeMsgEvent " + event.getPipeID());
        }

        String id = event.getPipeID().toString();

        PipeListenerList list = (PipeListenerList)pipeListeners.get(id);
        if (list != null) {
            Message message = event.getMessage();

            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("pipeMsgEvent: start sending to each requestor");
            }
            list.send((Message) message.clone(), id);
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("pipeMsgEvent: end sending to each requestor");
            }
        } else {
            // there are no listeners, close the input pipe
            ((InputPipe)event.getSource()).close();
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("close pipe id=" + id);
            }
        }
    }

    public synchronized void outputPipeEvent(OutputPipeEvent event) {

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("outputPipeEvent " + event);
        }
        PendingPipe p =
            (PendingPipe) pendingPipes.remove(event.getPipeID());

        // Noone cares (anylonger). TBD should it be removed then??
        if (p == null) {
            event.getOutputPipe().close();
            return;
        }

        resolvedPipes.put(event.getPipeID(), event.getOutputPipe());
        p.sendPending(event.getOutputPipe());
    }

    private static String popString(String name, Message message) {
        MessageElement el = message.getMessageElement(PROXYNS, name);
        if (el != null) {
            message.removeMessageElement(el);
            return el.toString();
        }
        return null;
    }

    static class PipeListenerList {
        LinkedList list = new LinkedList();
        InputPipe inputPipe = null;
        Map pipeListeners = null;
        String id = null;

        PipeListenerList(InputPipe inputPipe,
                         Map pipeListeners,
                         String id) {
            this.inputPipe = inputPipe;
            this.pipeListeners = pipeListeners;
            this.id = id;

            if (pipeListeners != null) {
                if (LOG.isEnabledFor(Level.INFO)) {
                    LOG.info("number of pipeListeners = " + pipeListeners.size());
                }
            }
        }

        void add(Requestor requestor) {
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info("add " + requestor + " from " + toString());
            }

            if (!list.contains(requestor)) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("requestor add");
                }
                list.add(requestor);
            } else {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("requestor exits already");
                }
            }
        }

        void remove(Requestor requestor) {
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info("remove " + requestor + " from " + toString());
            }

            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("removed = " + list.remove(requestor));
            }

            if (list.size() == 0) {
                // close the pipe and remove from the listenerList
                if (inputPipe != null) {
                    inputPipe.close();
                }

                if (id != null && pipeListeners != null) {
                    pipeListeners.remove(id);
                }
            }
        }

        int size() {
            int size = list.size();
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("size " + size);
            }
            return size;
        }

        private static StringMessageElement sme = new StringMessageElement(RESPONSE_TAG, RESPONSE_MESSAGE, null);

        void send(Message message, String id) {
            LOG.debug("send list.size = " + list.size());

            message.addMessageElement(PROXYNS,sme);

            message.addMessageElement(PROXYNS, new StringMessageElement(ID_TAG, id, null));

            // removed all element that are known to be not needed
            Iterator elements = message.getMessageElements();

            while (elements.hasNext()) {
                MessageElement el = (MessageElement) elements.next();
                String name = el.getElementName();

                if (name.startsWith("RendezVousPropagate")) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("JxtaWireHeader")) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("RdvIncarnjxta")) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("JxtaEndpointRouter")) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("EndpointRouterMsg")) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("removeMessageElement " + name);
                    }
                    elements.remove();
                }  else if (name.startsWith("EndpointHeaderSrcPeer")) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("removeMessageElement " + name);
                    }
                    elements.remove();
                }
            }

            Iterator iterator = list.iterator();
            try {
                while (iterator.hasNext()) {
                    Requestor requestor = (Requestor)iterator.next();
                    if (requestor.send((Message)message.clone()) == false) {
                        // could not send to listener, remove them from the list
                        remove(requestor);
                    }
                }
            } catch (Exception ex) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Error sending" + ex);
                }
            }
        }

        public String toString() {
            String str = "PipeListenerList size=" + list.size();
            return str;
        }
    }

    protected static void logMessage(Message message, Logger log) {
        String out = "\n**************** begin ****************\n";

        Message.ElementIterator elements = message.getMessageElements();

        while (elements.hasNext()) {
            MessageElement  element = (MessageElement) elements.next();
            out += "[" + elements.getNamespace() + "," +
                   element.getElementName() + "]=" +
                   element.toString() + "\n";
        }

        if (log.isEnabledFor(Level.DEBUG)) {
            log.debug(out + "****************  end  ****************\n");
        }
    }


    /****************************************************************
     * Implement the CacheEntryListener                             *
     ***************************************************************/

    public void purged(CacheEntry ce) {
        // A resolved pipe was purged from the cache because we have to
        // many pre-resolved pipes hanging around. Close it, because
        // it may be holding critical resources that the GC will not be
        // sensitive to.
        ((OutputPipe)(ce.getValue())).close();
    }
}