[Gzz-commits] storm/org/nongnu/storm impl/AsyncSetCollector.j...

[Benja Fallenstein]

CVSROOT:        /cvsroot/storm
Module name:    storm
Changes by:     Benja Fallenstein <address@hidden>      03/05/09 06:54:56

Modified files:
        org/nongnu/storm/impl: AsyncSetCollector.java 
        org/nongnu/storm/impl/p2p: P2PPool.java Peer.java 
        org/nongnu/storm/util: HTTPProxy.java 

Log message:
        Give correct list of blocks in p2p gateway.
        This is done by putting them all
        in the DHT under a given hashtable key.
        Not too scalable, though, I guess:
        if there are millions of blocks, a singe node
        will need to know about them all (because
        they're all put with that same hashtable key).

CVSWeb URLs:
http://savannah.gnu.org/cgi-bin/viewcvs/storm/storm/org/nongnu/storm/impl/AsyncSetCollector.java.diff?tr1=1.9&tr2=1.10&r1=text&r2=text
http://savannah.gnu.org/cgi-bin/viewcvs/storm/storm/org/nongnu/storm/impl/p2p/P2PPool.java.diff?tr1=1.7&tr2=1.8&r1=text&r2=text
http://savannah.gnu.org/cgi-bin/viewcvs/storm/storm/org/nongnu/storm/impl/p2p/Peer.java.diff?tr1=1.5&tr2=1.6&r1=text&r2=text
http://savannah.gnu.org/cgi-bin/viewcvs/storm/storm/org/nongnu/storm/util/HTTPProxy.java.diff?tr1=1.27&tr2=1.28&r1=text&r2=text

Patches:
Index: storm/org/nongnu/storm/impl/AsyncSetCollector.java
diff -u storm/org/nongnu/storm/impl/AsyncSetCollector.java:1.9 
storm/org/nongnu/storm/impl/AsyncSetCollector.java:1.10
--- storm/org/nongnu/storm/impl/AsyncSetCollector.java:1.9      Fri May  2 
09:28:38 2003
+++ storm/org/nongnu/storm/impl/AsyncSetCollector.java  Fri May  9 06:54:56 2003
@@ -58,7 +58,9 @@
        if(state != 0)
            throw new IllegalStateException("Cannot receive more elements");
        
-       set.add(o);
+       synchronized(set) {
+           set.add(o);
+       }
 
        for(Iterator i=listeners.iterator(); i.hasNext();) {
            CollectionListener l = (CollectionListener)i.next();
@@ -113,7 +115,7 @@
        listeners.add(l);
        synchronized(set) {
            for(Iterator i=set.iterator(); i.hasNext();) {
-               if(l.item(i.next())) return;
+               if(!l.item(i.next())) return;
            }
        }
     }
Index: storm/org/nongnu/storm/impl/p2p/P2PPool.java
diff -u storm/org/nongnu/storm/impl/p2p/P2PPool.java:1.7 
storm/org/nongnu/storm/impl/p2p/P2PPool.java:1.8
--- storm/org/nongnu/storm/impl/p2p/P2PPool.java:1.7    Wed May  7 04:21:57 2003
+++ storm/org/nongnu/storm/impl/p2p/P2PPool.java        Fri May  9 06:54:56 2003
@@ -89,7 +89,22 @@
     }
 
     public SetCollector getIds() throws IOException {
-       return cache.getIds();
+       Collector c = map.get("http://fenfire.org/2003/05/published-blocks";);
+       final AsyncSetCollector result = new AsyncSetCollector();
+
+       for(Iterator i=cache.getIds().iterator(); i.hasNext();)
+           result.receive(i.next());
+
+       c.addCollectionListener(new CollectionListener() {
+               public boolean item(Object item) {
+                   result.receive(new BlockId((String)item));
+                   return true;
+               }
+               public void finish(boolean timeout) {
+                   result.finish(timeout);
+               }
+           });
+       return result;
     }
 
     public void add(Block b) {
Index: storm/org/nongnu/storm/impl/p2p/Peer.java
diff -u storm/org/nongnu/storm/impl/p2p/Peer.java:1.5 
storm/org/nongnu/storm/impl/p2p/Peer.java:1.6
--- storm/org/nongnu/storm/impl/p2p/Peer.java:1.5       Wed May  7 05:15:23 2003
+++ storm/org/nongnu/storm/impl/p2p/Peer.java   Fri May  9 06:54:56 2003
@@ -79,6 +79,8 @@
            BlockId id = (BlockId)i.next();
            Block block = publishedPool.get(id);
            map.put(id.toString(), server.getURL()+id.toString());
+           map.put("http://fenfire.org/2003/05/published-blocks";,
+                   id.toString());
            
            p("Published: " + id.toString() + "(key):" + server.getURL() +
            id.toString()+ "(value)\n");
@@ -95,7 +97,6 @@
                }
            }
        }
-       // XXX put indexing information into pool!
     }
 
     public IndexedPool getPool() {
Index: storm/org/nongnu/storm/util/HTTPProxy.java
diff -u storm/org/nongnu/storm/util/HTTPProxy.java:1.27 
storm/org/nongnu/storm/util/HTTPProxy.java:1.28
--- storm/org/nongnu/storm/util/HTTPProxy.java:1.27     Thu May  8 12:13:15 2003
+++ storm/org/nongnu/storm/util/HTTPProxy.java  Fri May  9 06:54:56 2003
@@ -203,12 +203,23 @@
            
            writeRewriteLinks(w, rewrite);
 
-           for(Iterator i=pool.getIds().block().iterator(); 
-               i.hasNext();) {
-               BlockId id = (BlockId)i.next();
-               String s = id.getURI();
-               if(rewrite) s = base + s;
-               w.write("<a href=\""+s+"\">"+id+"</a><br>\n");
+           SetCollector ids = pool.getIds();
+
+           try {
+               // wait for DHT to receive information from network
+               Thread.sleep(2000);
+           } catch(InterruptedException _) {}
+
+           System.out.println("start recv'n");
+
+           synchronized(ids) {
+               for(Iterator i=ids.iterator(); i.hasNext();) {
+                   BlockId id = (BlockId)i.next();
+                   System.out.println("recvd: "+id);
+                   String s = id.getURI();
+                   if(rewrite) s = base + s;
+                   w.write("<a href=\""+s+"\">"+id+"</a><br>\n");
+               }
            }
            w.write("</body></html>\n");
            w.close();