srdiindex.java

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

            
            for (int i = 0; i < size; i++) {
                try {
                    String idstr = ois.readUTF();
                    PeerID pid = (PeerID) IDFactory.fromURI(new URI(idstr));
                    long exp = ois.readLong();
                    Entry entry = new Entry(pid, exp);
                    
                    result.add(entry);
                } catch (URISyntaxException badID) {
                    // ignored
                }
            }
            ois.close();
        } catch (EOFException eofe) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "Empty record", eofe);
            }
        } catch (IOException ie) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Exception while reading Entry", ie);
            }
        }
        return new SrdiIndexRecord(key, result);
    }
    
    /**
     * Empties the index completely.
     * The entries are abandoned to the GC.
     */
    public synchronized void clear() {
        // FIXME changing the behavior a bit
        // instead of dropping all srdi entries, we let them expire
        // if that is not a desired behavior the indexer could be dropped
        // simply close it, and remove all index db created
        try {
            srdiIndexer.close();
            cacheDB.close();
        } catch (Exception e) {
            // bad bits we are done
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "failed to close index", e);
            }
        }
    }
    
    /**
     * Garbage Collect expired entries
     */
    public void garbageCollect() {
        try {
            Map<String, NameIndexer> map = srdiIndexer.getIndexers();
            
            for(NameIndexer idxr : map.values()) {
                List<Long> list = new ArrayList<Long>();
                
                if(stop) {
                    break;
                }
                
                synchronized(this) {
                    idxr.query(null, new GcCallback(cacheDB, srdiIndexer, list, gcPeerTBL));
                    srdiIndexer.purge(list);
                }
            }
            gcPeerTBL.clear();
        } catch (Exception ex) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Failure during SRDI Garbage Collect", ex);
            }
        }
    }
    
    /**
     * Remove expired entries from a List
     *
     * @param list A list of entries.
     * @return The same list with the expired entries removed.
     */
    private static List<Entry> removeExpired(List<Entry> list) {
        Iterator<Entry> eachEntry = list.iterator();
        
        while(eachEntry.hasNext()) {
            Entry entry = eachEntry.next();
            
            if (entry.isExpired()) {
                eachEntry.remove();
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Removing expired Entry peerid :" + entry.peerid + " Expires at :" + entry.expiration);
                }
            }
        }
        return list;
    }
    
    private static boolean isExpired(long expiration) {
        return (TimeUtils.timeNow() > expiration);
    }
    
    /**
     * stop the current running thread
     */
    public synchronized void stop() {
        if(stop) {
            return;
        }
        
        stop = true;
        
        // wakeup and die
        try {
            Thread temp = gcThread;
            if (temp != null) {
                synchronized (temp) {
                    temp.notify();
                }
            }
        } catch (Exception ignored) {
            // ignored
        }
        
        // Stop the database
        
        try {
            srdiIndexer.close();
            cacheDB.close();
            gcPeerTBL.clear();
        } catch (Exception ex) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Unable to stop the Srdi Indexer", ex);
            }
        }
    }
    
    /**
     * {@inheritDoc}
     * <p/>
     * Periodic thread for GC
     */
    public void run() {
        try {
            while (!stop) {
                try {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Waiting for " + interval + "ms before garbage collection");
                    }
                    synchronized (gcThread) {
                        gcThread.wait(interval);
                    }
                } catch (InterruptedException woken) {
                    // The only reason we are woken is to stop.
                    Thread.interrupted();
                    continue;
                }
                
                if (stop) {
                    break;
                }
                
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Garbage collection started");
                }
                
                garbageCollect();
                
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Garbage collection completed");
                }
            }
        } catch (Throwable all) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all);
            }
        } finally {
            synchronized (this) {
                gcThread = null;
            }
        }
    }
    
    /**
     * Flushes the Srdi directory for a specified group
     * this method should only be called before initialization of a given group
     * calling this method on a running group would have undefined results
     *
     * @param group group context
     */
    public static void clearSrdi(PeerGroup group) {
        
        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("Clearing SRDI for " + group.getPeerGroupName());
        }
        
        try {
            String pgdir = null;
            
            if (group == null) {
                pgdir = "srdi-index";
            } else {
                pgdir = group.getPeerGroupID().getUniqueValue().toString();
            }
            File rootDir = null;
            
            if (group != null) {
                rootDir = new File(new File(new File(group.getStoreHome()), "cm"), pgdir);
            }
            
            rootDir = new File(rootDir, "srdi");
            if (rootDir.exists()) {
                // remove it along with it's content
                String[] list = rootDir.list();
                
                for (String aList : list) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Removing : " + aList);
                    }
                    File file = new File(rootDir, aList);
                    
                    if (!file.delete()) {
                        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.warning("Unable to delete the file");
                        }
                    }
                }
                rootDir.delete();
            }
        } catch (Throwable t) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Unable to clear Srdi", t);
            }
        }
    }
    
    /**
     * An entry in the index tables.
     */
    public final static class Entry {
        
        public final PeerID peerid;
        public final long expiration;
        
        /**
         * Peer Pointer reference
         *
         * @param peerid     PeerID for this entry
         * @param expiration the expiration for this entry
         */
        public Entry(PeerID peerid, long expiration) {
            this.peerid = peerid;
            this.expiration = expiration;
        }
        
        /**
         * {@inheritDoc}
         */
        @Override
        public boolean equals(Object obj) {
            return obj instanceof Entry && (peerid.equals(((Entry) obj).peerid));
        }
        
        /**
         * {@inheritDoc}
         */
        @Override
        public int hashCode() {
            return peerid.hashCode();
        }
        
        /**
         *  Return the absolute time in milliseconds at which this entry will
         *  expire.
         *
         *  @return The absolute time in milliseconds at which this entry will
         *  expire.
         */
        public long getExpiration() {
            return expiration;
        }
        
        /**
         *  Return {@code true} if this entry is expired.
         *
         *  @return {@code true} if this entry is expired otherwise {@code false}.
         */
        public boolean isExpired() {
            return TimeUtils.timeNow() > expiration;
        }
    }
    
    
    /**
     * an SrdiIndexRecord wrapper
     */
    public final static class SrdiIndexRecord {
        
        public final Key key;
        public final List<Entry> list;
        
        /**
         * SrdiIndex record
         *
         * @param key  record key
         * @param list record entries
         */
        public SrdiIndexRecord(Key key,List<Entry> list) {
            this.key = key;
            this.list = list;
        }
        
        /**
         * {@inheritDoc}
         */
        @Override
        public boolean equals(Object obj) {
            return obj instanceof SrdiIndexRecord && (key.equals(((SrdiIndexRecord) obj).key));
        }
        
        /**
         * {@inheritDoc}
         */
        @Override
        public int hashCode() {
            return key.hashCode();
        }
    }
}