32 |
import org.nongnu.storm.util.ByteArrayKey; |
import org.nongnu.storm.util.ByteArrayKey; |
33 |
|
|
34 |
import java.io.*; |
import java.io.*; |
35 |
|
import java.lang.ref.SoftReference; |
36 |
import java.util.*; |
import java.util.*; |
37 |
import com.axlight.jnushare.gisp.*; |
import com.axlight.jnushare.gisp.*; |
38 |
|
|
45 |
* That's also the main reason this is in a module. |
* That's also the main reason this is in a module. |
46 |
*/ |
*/ |
47 |
public class GispP2PMap implements P2PMap { |
public class GispP2PMap implements P2PMap { |
48 |
static public boolean dbg = false; |
static public boolean dbg = true; |
49 |
static private void pa(String s) { System.out.println(s); } |
static private void pa(String s) { System.out.println(s); } |
50 |
|
|
51 |
protected GISP gisp; |
protected GISP gisp; |
52 |
|
protected Set entries = new HashSet(); |
53 |
protected Map cache = new HashMap(); |
protected Map cache = new HashMap(); |
54 |
|
|
55 |
/** The "STORM" port. */ |
/** The "STORM" port. */ |
61 |
if(seedAddresses != null) |
if(seedAddresses != null) |
62 |
gisp.addSeedAddresses(seedAddresses); |
gisp.addSeedAddresses(seedAddresses); |
63 |
gisp.start(new String[] {"strength_min=1"}); |
gisp.start(new String[] {"strength_min=1"}); |
64 |
|
|
65 |
|
// thread used to republish items into the hashtable |
66 |
|
// GISP drops items after a 30 minute timeout |
67 |
|
keepaliveThread.start(); |
68 |
|
|
69 |
/** Doesn't work...maybe too quicly requested |
/** Doesn't work...maybe requested too quickly |
70 |
pa("Number of known peers in the network: " + String.valueOf(this.gisp.getNumOfPeers())); |
pa("Number of known peers in the network: " + String.valueOf(this.gisp.getNumOfPeers())); |
71 |
*/ |
*/ |
72 |
} |
} |
89 |
public void put(String key, String value) throws IOException { |
public void put(String key, String value) throws IOException { |
90 |
gisp.insert(key, value); |
gisp.insert(key, value); |
91 |
cache.remove(key); |
cache.remove(key); |
92 |
|
synchronized(entries) { |
93 |
|
entries.add(new Entry(key, value)); |
94 |
|
} |
95 |
|
} |
96 |
|
|
97 |
// XXX keep alive after ttl... |
/** Publish all entries from the map in the DHT again. |
98 |
|
* Used to keep them alive after timeout. |
99 |
|
*/ |
100 |
|
public void republish() { |
101 |
|
synchronized(entries) { |
102 |
|
for(Iterator i=entries.iterator(); i.hasNext();) { |
103 |
|
Entry e = (Entry)i.next(); |
104 |
|
gisp.insert(e.key, e.value); |
105 |
|
} |
106 |
|
} |
107 |
} |
} |
108 |
|
|
109 |
public Collector get(String key) throws IOException { |
public Collector get(String key) throws IOException { |
142 |
} |
} |
143 |
} |
} |
144 |
|
|
145 |
|
protected class Entry { |
146 |
|
protected final String key, value; |
147 |
|
protected Entry(String k, String v) { key=k; value=v; } |
148 |
|
} |
149 |
|
|
150 |
|
protected Thread keepaliveThread = new Thread(new Keepalive(this)); |
151 |
|
protected static class Keepalive implements Runnable { |
152 |
|
SoftReference ref; |
153 |
|
protected Keepalive(GispP2PMap map) { |
154 |
|
this.ref = new SoftReference(map); |
155 |
|
} |
156 |
|
public void run() { |
157 |
|
while(true) { |
158 |
|
// wait 25 minutes before republishing everything |
159 |
|
// GISP timeout is 30 minutes |
160 |
|
try { |
161 |
|
Thread.sleep(25 * 60 * 1000); |
162 |
|
} catch(InterruptedException _) {} |
163 |
|
GispP2PMap map = (GispP2PMap)ref.get(); |
164 |
|
if(map == null) break; |
165 |
|
if(dbg) pa("Republish GispP2PMap"); |
166 |
|
map.republish(); |
167 |
|
} |
168 |
|
if(dbg) pa("GispP2PMap object garbage collected"); |
169 |
|
} |
170 |
|
} |
171 |
|
|
172 |
|
protected void finalize() { |
173 |
|
keepaliveThread.interrupt(); |
174 |
|
} |
175 |
|
|
176 |
public static void main(String argv[]) throws Exception { |
public static void main(String argv[]) throws Exception { |
177 |
new GispP2PMap(PORT, argv); |
new GispP2PMap(PORT, argv); |
178 |
} |
} |