genericpeergroup.java

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

        if ((gid == null) || ID.nullID.equals(gid)) {
            throw new IllegalArgumentException("Advertisement did not contain a peer group ID");
        }
        
        PeerGroup theNewGroup = globalRegistry.lookupInstance(gid);
        
        if (theNewGroup != null) {
            return theNewGroup;
        }
        
        // We do not know if the grp adv had been previously published or not...  Since it may contain information essential to
        // the configuration of services, we need to make sure it is published localy, rather than letting the group publish
        // itself after the fact.
        
        // FIXME [e-mail] 20040713 : The downside is that we're publishing the adv even before making sure that this group
        // can really be instantiated. We're basically using the cm as a means to pass parameters to the module because it is a
        // group. We have the same parameter issue with the config adv. Eventually we need to find a clean way of passing
        // parameters specific to a certain types of module.
        
        try {
            discovery.publish(adv, DEFAULT_LIFETIME, DEFAULT_EXPIRATION);
        } catch (Exception any) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Could not publish the group advertisement: ", any);
            }
        }
        
        theNewGroup = (PeerGroup) loadModule(adv.getPeerGroupID(), adv.getModuleSpecID(), Here, false);
        
        if (theNewGroup == null) {
            throw new PeerGroupException("Could not find group implementation with " + adv.getModuleSpecID());
        }
        
        return (PeerGroup) theNewGroup.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public PeerGroup newGroup(PeerGroupID gid, Advertisement impl, String name, String description) throws PeerGroupException {
        PeerGroup theNewGroup = null;
        
        if (null != gid) {
            theNewGroup = globalRegistry.lookupInstance(gid);
        }
        
        if (theNewGroup != null) {
            return theNewGroup;
        }
        
        try {
            theNewGroup = (PeerGroup) loadModule(gid, (ModuleImplAdvertisement) impl, false);
        } catch (Throwable any) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Could not load group implementation", any);
            }
            
            throw new PeerGroupException("Could not load group implementation", any);
        }
        
        try {
            // The group adv definitely needs to be published.
            theNewGroup.publishGroup(name, description);
        } catch (Exception any) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Could not publish group or implementation:", any);
            }
        }
        return (PeerGroup) theNewGroup.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public PeerGroup newGroup(PeerGroupID gid) throws PeerGroupException {
        
        if ((gid == null) || ID.nullID.equals(gid)) {
            throw new IllegalArgumentException("Invalid peer group ID");
        }
        
        PeerGroup result = globalRegistry.lookupInstance(gid);
        
        if (result != null) {
            return result;
        }
        
        PeerGroupAdvertisement adv;
        
        try {
            adv = (PeerGroupAdvertisement)
                    discoverOne(DiscoveryService.GROUP, "GID", gid.toString(), 120, PeerGroupAdvertisement.class);
        } catch (Throwable any) {
            throw new PeerGroupException("Failed finding group advertisement for " + gid, any);
        }
        
        if (adv == null) {
            throw new PeerGroupException("Could not find group advertisement for group " + gid);
        }
        
        return newGroup(adv);
    }
    
    /**
     * {@inheritDoc}
     */
    public JxtaLoader getLoader() {
        return loader;
    }
    
    /**
     * {@inheritDoc}
     */
    public String getPeerName() {
        // before init we must fail.
        if (null == peerAdvertisement) {
            throw new IllegalStateException("PeerGroup not sufficiently initialized");
        }
        return peerAdvertisement.getName();
    }
    
    /**
     * {@inheritDoc}
     */
    public String getPeerGroupName() {
        // before init we must fail.
        if (null == peerGroupAdvertisement) {
            throw new IllegalStateException("PeerGroup not sufficiently initialized");
        }
        return peerGroupAdvertisement.getName();
    }
    
    /**
     * {@inheritDoc}
     */
    public PeerGroupID getPeerGroupID() {
        // before init we must fail.
        if (null == peerGroupAdvertisement) {
            throw new IllegalStateException("PeerGroup not sufficiently initialized");
        }
        
        return peerGroupAdvertisement.getPeerGroupID();
    }
    
    /**
     * {@inheritDoc}
     */
    public PeerID getPeerID() {
        // before init we must fail.
        if (null == peerAdvertisement) {
            throw new IllegalStateException("PeerGroup not sufficiently initialized");
        }
        return peerAdvertisement.getPeerID();
    }
    
    /**
     * {@inheritDoc}
     */
    public PeerAdvertisement getPeerAdvertisement() {
        return peerAdvertisement;
    }
    
    /**
     * {@inheritDoc}
     */
    public PeerGroupAdvertisement getPeerGroupAdvertisement() {
        return peerGroupAdvertisement;
    }
    
    /**
     * {@inheritDoc}
     */
    public boolean isRendezvous() {
        if (rendezvous == null) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Rendezvous service null");
            }
        }
        return (rendezvous != null) && rendezvous.isRendezVous();
    }
    
    /*
     * shortcuts to the well-known services, in order to avoid calls to lookup.
     */
    
    /**
     * {@inheritDoc}
     */
    public EndpointService getEndpointService() {
        if (endpoint == null) {
            return null;
        }
        return (EndpointService) endpoint.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public ResolverService getResolverService() {
        if (resolver == null) {
            return null;
        }
        return (ResolverService) resolver.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public DiscoveryService getDiscoveryService() {
        if (discovery == null) {
            return null;
        }
        return (DiscoveryService) discovery.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public PeerInfoService getPeerInfoService() {
        if (peerinfo == null) {
            return null;
        }
        return (PeerInfoService) peerinfo.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public MembershipService getMembershipService() {
        if (membership == null) {
            return null;
        }
        return (MembershipService) membership.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public PipeService getPipeService() {
        if (pipe == null) {
            return null;
        }
        return (PipeService) pipe.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public RendezVousService getRendezVousService() {
        if (rendezvous == null) {
            return null;
        }
        return (RendezVousService) rendezvous.getInterface();
    }
    
    /**
     * {@inheritDoc}
     */
    public AccessService getAccessService() {
        if (access == null) {
            return null;
        }
        return (AccessService) access.getInterface();
    }

    /**
     * Returns the executor pool
     *
     * @return the executor pool
     */
    public Executor getExecutor() {
        return threadPool;
    }

    /**
     * Returns the scheduled executor. The
     *
     * @return the scheduled executor
     */
    public ScheduledExecutorService getScheduledExecutor() {
        // FIXME 20070815 bondolo We should return a proxy object to disable shutdown()
        return scheduledExecutor;
    }
    
    /**
     * Our rejected execution handler which has the effect of pausing the
     * caller until the task can be queued.
     */
    private static class CallerBlocksPolicy implements RejectedExecutionHandler {
        
        private CallerBlocksPolicy() {
        }
        
        /**
         * {@inheritDoc}
         */
        public void rejectedExecution(Runnable runnable, ThreadPoolExecutor executor) {
            BlockingQueue<Runnable> queue = executor.getQueue();

            while (!executor.isShutdown()) {
                executor.purge();

                try {
                    boolean pushed = queue.offer(runnable, 500, TimeUnit.MILLISECONDS);
                    
                    if (pushed) {
                        break;
                    }
                } catch (InterruptedException woken) {
                    throw new RejectedExecutionException("Interrupted while attempting to enqueue", woken);
                }
            }
        }
    }
    
    /**
     * Our thread factory that adds the threads to our thread group and names
     * the thread to something recognizable.
     */
    static class PeerGroupThreadFactory implements ThreadFactory {
        final AtomicInteger threadNumber = new AtomicInteger(1);
        final String name;
        final ThreadGroup threadgroup;

        PeerGroupThreadFactory(String name, ThreadGroup threadgroup) {
            this.name = name;
            this.threadgroup = threadgroup;
        }

        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(threadgroup, runnable,  name + " - " + threadNumber.getAndIncrement(), 0);
            if(thread.isDaemon()) {
                thread.setDaemon(false);
            }
            if (thread.getPriority() != Thread.NORM_PRIORITY) {
                thread.setPriority(Thread.NORM_PRIORITY);
            }
            return thread;
        }
    }
}