/[storm]/storm/org/nongnu/storm/impl/p2p/P2PPool.java
ViewVC logotype

Diff of /storm/org/nongnu/storm/impl/p2p/P2PPool.java

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.2 by benja, Thu Apr 17 08:36:34 2003 UTC revision 1.3 by benja, Sat Apr 19 12:20:32 2003 UTC
# Line 28  P2PPool.java Line 28  P2PPool.java
28  package org.nongnu.storm.impl.p2p;  package org.nongnu.storm.impl.p2p;
29  import org.nongnu.storm.*;  import org.nongnu.storm.*;
30  import org.nongnu.storm.impl.*;  import org.nongnu.storm.impl.*;
31    import org.nongnu.storm.util.*;
32  import java.io.*;  import java.io.*;
33    import java.net.*;
34  import java.util.*;  import java.util.*;
35    
36  /** P2P-based, indexed Storm pool.  /** P2P-based, indexed Storm pool.
37   *  "Componentalized" into a "P2PMap" (an peer-to-peer indexing mechanism)   *  This pool retrieves blocks from a Storm P2P network.
38   *  and a "Server" (actually a matching client and server implementation,   *  <p>
39   *  for retrieving actual files). The simplest Server impl is HTTP   *  The distributed search network is accessed through the
40   *  (see <code>HttpP2PServer.py</code>, a Python module because   *  <code>P2PMap</code> interface. Different implementations
41   *  we use Python's existing HTTP server code).   *  are possible.
42   *  <p>   *  <p>
  *  XXX finish  
43   *  XXX Become asynchronous -> don't subclass AbstractLocalPool...   *  XXX Become asynchronous -> don't subclass AbstractLocalPool...
44   */   */
45  public abstract class P2PPool extends AbstractLocalPool {  public class P2PPool extends AbstractLocalPool {
46        static public boolean dbg = false;
47        static private void p(String s) { System.out.println(s); }
48    
49      protected P2PMap map;      protected P2PMap map;
50      protected Server server;      protected StormPool cache;
   
     /** A distributed search network.  
      *  Can be a DHT; can also be a Gnutella-like network  
      *  if desired. Abstracts over time-to-live in DHTs:  
      *  When an item has been <code>put()</code>,  
      *  this object will try to keep it published  
      *  in the network for its whole lifetime,  
      *  even if it has to re-publish it every  
      *  thirty minutes or so (e.g. in GISP).  
      *  This abstraction knows nothing about  
      *  actually downloading blocks.  
      */  
     public interface P2PMap {  
         Collector get(String key) throws IOException;  
         void put(String key, String value) throws IOException;  
     }  
   
     /** A method for sharing/downloading blocks on the network.  
      *  This is a server serving all blocks in a given pool  
      *  as well as a client for downloading blocks  
      *  from servers like this. This abstraction  
      *  knows nothing about indexing.  
      *  <p>  
      *  E.g. HTTP-based: This computer's blocks are served through HTTP,  
      *  and to download, HTTP connections to other servers are made.  
      */  
     public interface Server {  
         /** Set the pool whose blocks are served  
          *  by this <code>Server</code>.  
          *  @throws IllegalStateException if this method  
          *          has already been called.  
          */  
         void setPool(StormPool pool);  
   
         /** Get the URL this server offers the given block under.  
          *  Could be e.g. a <code>http://</code> URL. It is  
          *  published through the <code>P2PMap</code> so that  
          *  other clients with the same kind of <code>Server</code>  
          *  are able to download this block.  
          */  
         String getURL(BlockId blockId);  
   
         /** Download the raw block data stored at a given URL.  
          *  Simply throws an IOException if the kind of URL  
          *  cannot be handled by this kind of server.  
          */  
         InputStream download(String url) throws IOException;  
     }  
51    
52      public P2PPool(P2PMap map, Server server,      public P2PPool(P2PMap map, StormPool cache,
53                     Set indexTypes) throws IOException {                     Set indexTypes) throws IOException {
54          super(indexTypes);          super(indexTypes);
55          this.map = map;          this.map = map;
56          this.server = server;          this.cache = cache;
57        }
58    
59        public Block get(BlockId id) throws IOException {
60            try {
61                return cache.get(id);
62            } catch(FileNotFoundException _) {
63                Collection c = map.get(id.toString()).block();
64                for(Iterator i=c.iterator(); i.hasNext();) {
65                    String url = (String)i.next();
66                    URLConnection conn = new URL(url).openConnection();
67                    BlockOutputStream bos =
68                        cache.getBlockOutputStream(conn.getContentType());
69                    CopyUtil.copy(conn.getInputStream(), bos);
70                    if(bos.getBlockId().equals(id))
71                        return bos.getBlock();
72                    else
73                        p("Block ids didn't match: "+id+" / "+
74                          bos.getBlockId()+" (from "+url+").");
75                }
76                throw new FileNotFoundException(""+id);
77            }
78        }
79    
80        public SetCollector getIds() throws IOException {
81            return cache.getIds();
82        }
83    
84        public void add(Block b) {
85            throw new UnsupportedOperationException("P2PPool is retrieval only");
86        }
87    
88        public void delete(Block b) {
89            throw new UnsupportedOperationException("P2PPool is retrieval only");
90        }
91    
92        public BlockOutputStream getBlockOutputStream(String contentType) {
93            throw new UnsupportedOperationException("P2PPool is retrieval only");
94        }
95    
96        protected DB getDB(IndexType indexType) {
97            final String prefix = indexType.getIndexTypeURI()+" ";
98            return new DB() {
99                    public Collector get(String key) throws IOException {
100                        return map.get(prefix + key);
101                    }
102                };
103      }      }
104  }  }

Legend:
Removed from v.1.2  
changed lines
  Added in v.1.3

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26