/[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.4 by benja, Mon Apr 21 18:15:43 2003 UTC revision 1.5 by benja, Mon Apr 21 21:07:41 2003 UTC
# Line 63  public class P2PPool extends AbstractLoc Line 63  public class P2PPool extends AbstractLoc
63              Collector c = map.get(id.toString());              Collector c = map.get(id.toString());
64              for(Iterator i=c.blockingIterator(); i.hasNext();) {              for(Iterator i=c.blockingIterator(); i.hasNext();) {
65                  String url = (String)i.next();                  String url = (String)i.next();
66                  URLConnection conn = new URL(url).openConnection();                  try {
67                  BlockOutputStream bos =                      URLConnection conn = new URL(url).openConnection();
68                      cache.getBlockOutputStream(conn.getContentType());                      String contentType = conn.getContentType();
69                  CopyUtil.copy(conn.getInputStream(), bos);                      if(contentType == null) continue;
70                  if(bos.getBlockId().equals(id))                      BlockOutputStream bos =
71                      return bos.getBlock();                          cache.getBlockOutputStream(conn.getContentType());
72                  else                      CopyUtil.copy(conn.getInputStream(), bos);
73                      p("Block ids didn't match: "+id+" / "+                      if(bos.getBlockId().equals(id))
74                        bos.getBlockId()+" (from "+url+").");                          return bos.getBlock();
75                        else
76                            p("Block ids didn't match: "+id+" / "+
77                              bos.getBlockId()+" (from "+url+").");
78                    } catch(IOException e) {
79                        // next iteration
80                    }
81              }              }
82              throw new FileNotFoundException(""+id);              throw new FileNotFoundException(""+id);
83          }          }
# Line 97  public class P2PPool extends AbstractLoc Line 103  public class P2PPool extends AbstractLoc
103          final String prefix = indexType.getIndexTypeURI()+" ";          final String prefix = indexType.getIndexTypeURI()+" ";
104          return new DB() {          return new DB() {
105                  public Collector get(String key) throws IOException {                  public Collector get(String key) throws IOException {
106                      return map.get(prefix + key);                      Collector collector = map.get(prefix+key);
107                        AsyncSetCollector result = new AsyncSetCollector();
108                        CollectionListener l = new MyListener(key, result);
109                        collector.addCollectionListener(l);
110                        return result;
111                  }                  }
112              };              };
113      }      }
114    
115        private class MyListener implements CollectionListener {
116            String key;
117            AsyncSetCollector addTo;
118            MyListener(String key, AsyncSetCollector addTo) {
119                this.key = key;
120                this.addTo = addTo;
121            }
122            public boolean item(Object o) {
123                try {
124                    String s = (String)o;
125                    int sp = s.indexOf(' ');
126                    if(sp < 0) return true; // ignore bad data
127                    BlockId id = new BlockId(s.substring(0, sp));
128                    String value = s.substring(sp+1);
129                    addTo.receive(new IndexedPool.Mapping(id, key, value));
130                } catch(IllegalArgumentException _) {
131                    // When there's an exception when creating the id,
132                    // assume that the data we received from the net
133                    // was bad; simply ignore.
134                }
135                return true;
136            }
137            public void finish(boolean timeout) {
138                addTo.finish(timeout);
139            }
140        }
141  }  }

Legend:
Removed from v.1.4  
changed lines
  Added in v.1.5

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