lib/java/dolda/dolcon/HubManager.java

   1 package dolda.dolcon;
   2
   3 import java.util.*;
   4 import dolda.dolcon.protocol.*;
   5
   6 class HubManager implements NotifyListener {
   7     private Set<HubListener> hubls = new HashSet<HubListener>();
   8     private Set<HubListener> delayed = new HashSet<HubListener>();
   9     private Map<Integer, Hub> hubs = new TreeMap<Integer, Hub>();
  10     private String state = "none";
  11     private Session sess;
  12
  13     HubManager(Session sess) {
  14         this.sess = sess;
  15     }
  16
  17     private int atoi(String a) {
  18         return(Integer.parseInt(a));
  19     }
  20
  21     private void addall(final HubListener ls) {
  22         for(final Hub hub : hubs.values()) {
  23             sess.dispatch(new Runnable() {
  24                     public void run() {
  25                         ls.added(hub);
  26                     }
  27                 });
  28         }
  29     }
  30
  31     private void fetchhubs() {
  32         synchronized(this) {
  33             if(state != "none")
  34                 return;
  35             state = "fetch";
  36         }
  37         Command cmd = new Command("lsnodes");
  38         cmd.new Listener() {
  39                 public void done(Response r) {
  40                     if(r.code != 200)
  41                         return;
  42                     for(List<String> line : r.lines) {
  43                         Hub h = new Hub(atoi(line.get(0)), line.get(1));
  44                         h.name = line.get(2);
  45                         h.numpeers = atoi(line.get(3));
  46                         h.state = new String[] {"syn", "hs", "est", "dead"}[atoi(line.get(4))];
  47                         h.gid = line.get(5);
  48                         hubs.put(h.id, h);
  49                     }
  50                     synchronized(HubManager.this) {
  51                         state = "";
  52                         HubManager.this.notifyAll();
  53                         for(HubListener ls : delayed) {
  54                             addall(ls);
  55                         }
  56                     }
  57                 }
  58
  59                 public void error(Exception e) {
  60                     synchronized(HubManager.this) {
  61                         state = "closed";
  62                     }
  63                 }
  64             };
  65         sess.conn.qcmd(new Command("notify", "fn:act", "on"), cmd);
  66         sess.conn.addNotifyListener(this);
  67     }
  68
  69     public Collection<Hub> gethubs() throws InterruptedException {
  70         fetchhubs();
  71         synchronized(this) {
  72             while((state != "") && (state != "closed"))
  73                 wait();
  74         }
  75         Collection<Hub> ret = new LinkedList<Hub>();
  76         synchronized(hubs) {
  77             for(Hub h : hubs.values())
  78                 ret.add(h.copy());
  79         }
  80         return(ret);
  81     }
  82
  83     public void addls(HubListener hl, boolean addexisting) {
  84         fetchhubs();
  85         synchronized(hubls) {
  86             hubls.add(hl);
  87         }
  88         if(addexisting) {
  89             synchronized(this) {
  90                 if(state != "")
  91                     delayed.add(hl);
  92                 else
  93                     addall(hl);
  94             }
  95         }
  96     }
  97
  98     public void rmls(HubListener hl) {
  99         synchronized(sess) {
 100             synchronized(hubls) {
 101                 hubls.remove(hl);
 102                 if(hubls.isEmpty()) {
 103                     synchronized(hubs) {
 104                         hubs.clear();
 105                     }
 106                     state = "closed";
 107                     sess.conn.removeNotifyListener(this);
 108                     sess.hm = null;
 109                 }
 110             }
 111         }
 112     }
 113
 114     public void notified(Response resp) {
 115         synchronized(this) {
 116             if(state != "")
 117                 return;
 118         }
 119         if(resp.code == 604) {
 120             final Hub h = new Hub(atoi(resp.token(0, 0)), resp.token(0, 1));
 121             synchronized(hubs) {
 122                 hubs.put(h.id, h);
 123             }
 124             sess.dispatch(new Runnable() {
 125                     public void run() {
 126                         synchronized(hubls) {
 127                             for(HubListener ls : hubls)
 128                                 ls.added(h);
 129                         }
 130                     }
 131                 });
 132         } else if(resp.code == 603) {
 133             final Hub h;
 134             synchronized(hubs) {
 135                 h = hubs.remove(atoi(resp.token(0, 0)));
 136             }
 137             sess.dispatch(new Runnable() {
 138                     public void run() {
 139                         synchronized(hubls) {
 140                             for(HubListener ls : hubls)
 141                                 ls.removed(h);
 142                         }
 143                     }
 144                 });
 145         } else if(resp.code == 601) {
 146             final Hub h;
 147             final String state = new String[] {"syn", "hs", "est", "dead"}[atoi(resp.token(0, 1))];
 148             synchronized(hubs) {
 149                 h = hubs.get(atoi(resp.token(0, 0)));
 150             }
 151             h.state = state;
 152             sess.dispatch(new Runnable() {
 153                     public void run() {
 154                         synchronized(h.ls) {
 155                             for(Hub.Listener ls : h.ls) {
 156                                 ls.chState(h);
 157                             }
 158                         }
 159                     }
 160                 });
 161         } else if(resp.code == 602) {
 162             final Hub h;
 163             final String name = resp.token(0, 1);
 164             synchronized(hubs) {
 165                 h = hubs.get(atoi(resp.token(0, 0)));
 166             }
 167             h.name = name;
 168             sess.dispatch(new Runnable() {
 169                     public void run() {
 170                         synchronized(h.ls) {
 171                             for(Hub.Listener ls : h.ls) {
 172                                 ls.chName(h);
 173                             }
 174                         }
 175                     }
 176                 });
 177         } else if(resp.code == 605) {
 178             final Hub h;
 179             final int np = atoi(resp.token(0, 1));
 180             synchronized(hubs) {
 181                 h = hubs.get(atoi(resp.token(0, 0)));
 182             }
 183             h.numpeers = np;
 184             sess.dispatch(new Runnable() {
 185                     public void run() {
 186                         synchronized(h.ls) {
 187                             for(Hub.Listener ls : h.ls) {
 188                                 ls.chNumPeers(h);
 189                             }
 190                         }
 191                     }
 192                 });
 193         }
 194     }
 195 }