proxyservice.java

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

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

        PeerGroupID gid = null;

        if (id != null) {
            try {
                ID tempId = IDFactory.fromURI(new URI(id));

                gid = (PeerGroupID) tempId;
            } catch (URISyntaxException e) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Invalid peergroupId", e);
                }
            } catch (ClassCastException e) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "id was not a peergroup id", e);
                }
            }
        }

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

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("newPeerGroupAdvertisement name=" + name + " id=" + gid.toString());
        }

        adv = 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 (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "newPeerGroupAdvertisement Exception", e);
            }
        }

        return adv;
    }

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

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("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 (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "newPipeAdvertisement Exception", e);
            }
        }

        return adv;
    }

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

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("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;
        }

        Enumeration<Advertisement> each;

        try {
            each = discovery.getLocalAdvertisements(DiscoveryService.ADV, attribute, value);
        } catch (IOException e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "IOException in getLocalAdvertisements()", e);
            }
            return null;
        }

        PipeAdvertisement pipeAdv = null;

        while (each.hasMoreElements()) {
            Advertisement adv = each.nextElement();

            // take the first match
            if (adv instanceof PipeAdvertisement) {
                pipeAdv = (PipeAdvertisement) adv;
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("found PipeAdvertisement = " + pipeAdv);
                }
                break;
            }
        }

        return pipeAdv;
    }

    public synchronized void discoveryEvent(DiscoveryEvent event) {

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("discoveryEvent " + event);
        }

        Requestor requestor = searchRequests.get(event.getQueryID());
        if (requestor == null) {
            return;
        }

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

        Enumeration<Advertisement> each = response.getAdvertisements();
        if (each == null || !each.hasMoreElements()) {
            return;
        }
        // we have a response remove it from the LRUCache
        searchRequests.remove(event.getQueryID());
        int i = 0;

        while (each.hasMoreElements() && i < requestor.getThreshold()) {
            try {
                requestor.send(each.nextElement(), RESPONSE_RESULT);
            } catch (Exception e) {
                // this should not happen unless a bad result is returned
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Bad result returned by DiscoveryService", e);
                }
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void pipeMsgEvent(PipeMsgEvent event) {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("pipeMsgEvent " + event.getPipeID());
        }

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

        PipeListenerList list = pipeListeners.get(id);

        if (list != null) {
            Message message = event.getMessage();

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("pipeMsgEvent: start sending to each requestor");
            }
            list.send(message.clone(), id);
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("pipeMsgEvent: end sending to each requestor");
            }
        } else {
            // there are no listeners, close the input pipe
            ((InputPipe) event.getSource()).close();
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("close pipe id=" + id);
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void outputPipeEvent(OutputPipeEvent event) {

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("outputPipeEvent " + event);
        }
        PendingPipe p = (PendingPipe) pendingPipes.remove(event.getPipeID());

        // No one 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<Requestor> list = new LinkedList<Requestor>();
        InputPipe inputPipe = null;
        Map<String, PipeListenerList> pipeListeners = null;
        String id = null;

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

            if (pipeListeners != null) {
                if (Logging.SHOW_CONFIG && LOG.isLoggable(Level.CONFIG)) {
                    LOG.config("number of pipeListeners = " + pipeListeners.size());
                }
            }
        }

        void add(Requestor requestor) {
            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                LOG.info("add " + requestor + " from " + toString());
            }

            if (!list.contains(requestor)) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("requestor add");
                }
                list.add(requestor);
            } else {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("requestor exits already");
                }
            }
        }

        void remove(Requestor requestor) {
            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                LOG.info("remove " + requestor + " from " + toString());
            }

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("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 (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("size " + size);
            }
            return size;
        }

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

        void send(Message message, String id) {
            LOG.fine("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<MessageElement> elements = message.getMessageElements();

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

                if (name.startsWith("RendezVousPropagate")) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("JxtaWireHeader")) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("RdvIncarnjxta")) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("JxtaEndpointRouter")) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("EndpointRouterMsg")) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("removeMessageElement " + name);
                    }
                    elements.remove();
                } else if (name.startsWith("EndpointHeaderSrcPeer")) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("removeMessageElement " + name);
                    }
                    elements.remove();
                }
            }

            Iterator<Requestor> iterator = list.iterator();
            try {
                while (iterator.hasNext()) {
                    Requestor requestor = iterator.next();

                    if (!requestor.send(message.clone())) {
                        // could not send to listener, remove them from the list
                        remove(requestor);
                    }
                }
            } catch (Exception ex) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Error sending" + ex);
                }
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public String toString() {
            return "PipeListenerList size=" + list.size();
        }
    }

    protected static void logMessage(Message message, Logger log) {
        if (!Logging.SHOW_FINER || !log.isLoggable(Level.FINER)) {
            return;
        }

        StringBuilder out = new StringBuilder("\n**************** begin ****************\n");

        Message.ElementIterator elements = message.getMessageElements();

        while (elements.hasNext()) {
            MessageElement element = elements.next();

            out.append("[").append(elements.getNamespace()).append(",").append(element.getElementName()).append("]=").append(element.toString()).append(
                    "\n");
        }

        out.append("****************  end  ****************\n");
        log.finer(out.toString());
    }

    /**
     * {@inheritDoc}
     */
    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();
    }
}