Mercurial > hg > FederatedLinda

comparison src/fdl/test/debug/MetaProtocolEngine.java @ 85:d0d8aeaebccf

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
add routing table
author one
date Mon, 08 Feb 2010 11:07:57 +0900
parents c0575f877591
children c0591636a71a
comparison
equal deleted inserted replaced
84:c0575f877591 85:d0d8aeaebccf
40 private int managerPort = DEFAULTPORT; 40 private int managerPort = DEFAULTPORT;
41 private boolean running = true; 41 private boolean running = true;
42 private boolean connected = false; 42 private boolean connected = false;
43 private boolean debugConnected = false; 43 private boolean debugConnected = false;
44 private int nodeId; 44 private int nodeId;
45 private HashMap<Integer, Routing> nodes, debugNodes; 45 private HashMap<Integer, Routing> nodes;
46 46
47 // Callback class 47 // Callback class
48 class AcceptXMLCallback implements PSXCallback { 48 class AcceptXMLCallback implements PSXCallback {
49 int tid; 49 int tid;
50 50
94 int dstPort = new Integer(port.getAttribute("id")).intValue(); 94 int dstPort = new Integer(port.getAttribute("id")).intValue();
95 int dstId = new Integer(t.getAttribute("id")).intValue(); 95 int dstId = new Integer(t.getAttribute("id")).intValue();
96 try { 96 try {
97 PSXLindaImpl linda = (PSXLindaImpl) ml.open(dstHostName, dstPort); 97 PSXLindaImpl linda = (PSXLindaImpl) ml.open(dstHostName, dstPort);
98 Routing r = new Routing(linda, dstId); 98 Routing r = new Routing(linda, dstId);
99 if (tid == MANAGE) 99 nodes.put(new Integer(srcId), r);
100 nodes.put(new Integer(srcId), r); 100 ml.in(srcId, new RoutingCallback(srcId, r));
101 else if (tid == DEBUG)
102 debugNodes.put(new Integer(srcId), r);
103 // TODO: implements PSXCallback
104 ml.in(srcId);
105 } catch (IOException e) { 101 } catch (IOException e) {
106 e.printStackTrace(); 102 e.printStackTrace();
107 } 103 }
108 } 104 }
109 } else if (root.getTagName().equals("routing")) { 105 } else if (root.getTagName().equals("routing")) {
110 print("Routing xml received!"); 106 print("Routing xml received!");
111 HashMap<Integer, Routing> n; 107
112 if (tid == MANAGE) {
113 n = nodes;
114 } else if (tid == DEBUG) {
115 n = debugNodes;
116 }
117 NodeList routing = root.getElementsByTagName("source"); 108 NodeList routing = root.getElementsByTagName("source");
118 for (int i = 0; i < routing.getLength(); i++) { 109 for (int i = 0; i < routing.getLength(); i++) {
119 Element src = (Element) routing.item(i); 110 Element src = (Element) routing.item(i);
120 int srcId = new Integer(src.getAttribute("id")); 111 Integer srcId = new Integer(src.getAttribute("id"));
112 Routing r = nodes.get(srcId);
121 NodeList dest = src.getElementsByTagName("dest"); 113 NodeList dest = src.getElementsByTagName("dest");
122 for (int j = 0; j < dest.getLength(); i++) { 114 for (int j = 0; j < dest.getLength(); j++) {
123 Element dst = (Element) dest.item(i); 115 Element dst = (Element) dest.item(j);
124 int dstId = new Integer(dst.getAttribute("id")); 116 Integer dstId = new Integer(dst.getAttribute("id"));
117 r.route.add(dstId);
125 } 118 }
126 print(""); 119 }
127 }
128 //n.get(key);
129
130
131 120
132 } 121 }
133 if (tid == MANAGE) connected = true; 122 if (tid == MANAGE) connected = true;
134 else if (tid == DEBUG) debugConnected = true; 123 else if (tid == DEBUG) debugConnected = true;
135 if (connected && debugConnected) { 124 if (connected && debugConnected) {
139 } 128 }
140 } 129 }
141 130
142 } 131 }
143 132
133 private class RoutingCallback implements PSXCallback {
134 int tid;
135 Routing routing;
136 public RoutingCallback(int tid, Routing routing) {
137 this.tid = tid;
138 this.routing = routing;
139 }
140
141 public void callback(ByteBuffer reply) {
142 Iterator<Integer> it = routing.route.iterator();
143 while (it.hasNext()) {
144 Integer dstId = it.next();
145 Routing r = nodes.get(dstId);
146 r.linda.out(r.dstId, reply);
147 try {
148 r.linda.sync(1);
149 } catch (IOException e) {
150 e.printStackTrace();
151 }
152 ml.in(tid, this);
153 }
154 }
155
156 }
157
144 // Constructor 158 // Constructor
145 public MetaProtocolEngine(int port, MetaLinda ml, String managerHostName, int managerPort) { 159 public MetaProtocolEngine(int port, MetaLinda ml, String managerHostName, int managerPort) {
146 this.ml = ml; 160 this.ml = ml;
147 this.localPort = port; 161 this.localPort = port;
148 this.managerHostName = managerHostName; 162 this.managerHostName = managerHostName;
149 this.managerPort = managerPort; 163 this.managerPort = managerPort;
150 this.nodes = new HashMap<Integer, Routing>(); 164 this.nodes = new HashMap<Integer, Routing>();
151 this.debugNodes = new HashMap<Integer, Routing>();
152 try { 165 try {
153 this.localHostName = InetAddress.getLocalHost().getHostName(); 166 this.localHostName = InetAddress.getLocalHost().getHostName();
154 } catch (UnknownHostException e) { 167 } catch (UnknownHostException e) {
155 e.printStackTrace(); 168 e.printStackTrace();
156 } 169 }
166 } 179 }
167 180
168 protected void initPoller() { 181 protected void initPoller() {
169 ml.in(MANAGE, new AcceptXMLCallback(MANAGE)); 182 ml.in(MANAGE, new AcceptXMLCallback(MANAGE));
170 ml.in(DEBUG, new AcceptXMLCallback(DEBUG)); 183 ml.in(DEBUG, new AcceptXMLCallback(DEBUG));
171 //ml.in(debugId, new ConnectServerCallback(debugId, psxDebugServers, null));
172 //ml.in(registId, new RegistServerCallback());
173 } 184 }
174 185
175 protected void sendLocalHostName() { 186 protected void sendLocalHostName() {
176 // TopologyManager に自分のホストネームを送信して、起動を伝える 187 // TopologyManager に自分のホストネームを送信して、起動を伝える
177 ByteBuffer local; 188 ByteBuffer local;