|
266
|
1
|
|
0
|
2 package rep;
|
|
|
3
|
|
|
4 import java.io.IOException;
|
|
|
5 import java.net.InetSocketAddress;
|
|
267
|
6 import java.nio.channels.ClosedChannelException;
|
|
0
|
7 import java.nio.channels.SelectionKey;
|
|
83
|
8 import java.util.LinkedList;
|
|
144
|
9 import java.util.List;
|
|
231
|
10 import java.util.Set;
|
|
178
|
11 import java.util.concurrent.BlockingQueue;
|
|
192
|
12 import java.util.concurrent.LinkedBlockingQueue;
|
|
0
|
13
|
|
346
|
14 import org.xml.sax.SAXException;
|
|
|
15
|
|
198
|
16
|
|
353
|
17
|
|
337
|
18 import rep.channel.REPLogger;
|
|
123
|
19 import rep.channel.REPServerSocketChannel;
|
|
133
|
20 import rep.channel.REPSocketChannel;
|
|
144
|
21 import rep.handler.PacketSet;
|
|
146
|
22 import rep.handler.REPHandler;
|
|
158
|
23 import rep.channel.REPSelector;
|
|
56
|
24 import rep.xml.SessionXMLDecoder;
|
|
45
|
25 import rep.xml.SessionXMLEncoder;
|
|
198
|
26 import rep.channel.REPSelectionKey;
|
|
264
|
27
|
|
198
|
28 /*
|
|
264
|
29 +-------+--------+--------+-------+--------+---------+------+
|
|
|
30 | cmd | session| editor | seqid | lineno | textsiz | text |
|
|
|
31 | | id | id | | | | |
|
|
|
32 +-------+--------+--------+-------+--------+---------+------+
|
|
|
33 o---------- header section (network order) ----------o
|
|
|
34
|
|
|
35 int cmd; kind of command
|
|
|
36 int sid; session ID : uniqu to editing file
|
|
|
37 int eid; editor ID : owner editor ID = 1。Session に対して unique
|
|
308
|
38 -1 session manager command
|
|
|
39 -2 merge command
|
|
264
|
40 int seqno; Sequence number : sequence number はエディタごとに管理
|
|
|
41 int lineno; line number
|
|
|
42 int textsize; textsize : bytesize
|
|
|
43 byte[] text;
|
|
198
|
44 */
|
|
1
|
45
|
|
250
|
46 public class SessionManager implements SessionManagerEventListener{
|
|
337
|
47 static public REPLogger logger = REPLogger.singleton();
|
|
|
48
|
|
319
|
49 LinkedList<Session> sessionList;
|
|
280
|
50 private SessionManagerGUI gui;
|
|
198
|
51 private REPSelector<REPCommand> selector;
|
|
319
|
52 SessionManagerList smList;
|
|
|
53 List<Editor> editorList;
|
|
304
|
54 // editorList は、sessionList に入っているeditorとは別なeditorのlistらしい。
|
|
349
|
55 // private String maxHost;
|
|
212
|
56 private List<PacketSet> waitingCommandInMerge;
|
|
308
|
57 private BlockingQueue<SessionManagerEvent> waitingEventQueue = new LinkedBlockingQueue<SessionManagerEvent>();;
|
|
319
|
58 String myHost;
|
|
317
|
59 private LinkedList<PacketSet> writeQueue = new LinkedList<PacketSet>();
|
|
336
|
60 private int receive_port;
|
|
|
61 private int parent_port;
|
|
101
|
62 static final int DEFAULT_PORT = 8766;
|
|
332
|
63 private static final int packetLimit = 200;
|
|
353
|
64 SessionXMLDecoder decoder = new SessionXMLDecoder();
|
|
355
|
65 private Forwarder sm_join_channel;
|
|
349
|
66
|
|
316
|
67 public static void main(String[] args) throws InterruptedException, IOException {
|
|
|
68
|
|
|
69 int port = DEFAULT_PORT;
|
|
|
70 int port_s = DEFAULT_PORT;
|
|
|
71 //System.setProperty("file.encoding", "UTF-8");
|
|
|
72 if(args.length > 0){
|
|
|
73 port = Integer.parseInt(args[0]);
|
|
|
74 port_s = Integer.parseInt(args[1]);
|
|
|
75 }
|
|
|
76 SessionManager sm = new SessionManager();
|
|
336
|
77 sm.setReceivePort(port);
|
|
|
78 sm.setParentPort(port_s);
|
|
316
|
79 sm.init(port,new SessionManagerGUIimpl(sm));
|
|
|
80
|
|
|
81
|
|
|
82 }
|
|
|
83
|
|
2
|
84
|
|
336
|
85 public void setReceivePort(int port) {
|
|
|
86 receive_port = port;
|
|
|
87 }
|
|
|
88
|
|
|
89
|
|
2
|
90 public void openSelector() throws IOException{
|
|
231
|
91 selector = REPSelector.<REPCommand>create();
|
|
2
|
92 }
|
|
280
|
93
|
|
|
94 public void init(int port, SessionManagerGUI gui) throws IOException, InterruptedException {
|
|
|
95 this.gui = gui;
|
|
|
96 openSelector();
|
|
|
97 init(port);
|
|
|
98 mainLoop();
|
|
|
99 }
|
|
|
100
|
|
|
101
|
|
|
102 private void init(int port) throws InterruptedException, IOException {
|
|
2
|
103
|
|
186
|
104 REPServerSocketChannel<REPCommand> ssc = REPServerSocketChannel.<REPCommand>open(new REPCommandPacker());
|
|
122
|
105 ssc.configureBlocking(false); //reuse address 必須
|
|
101
|
106 ssc.socket().setReuseAddress(true);
|
|
212
|
107 //getAllByNameで取れた全てのアドレスに対してbindする
|
|
|
108 ssc.socket().bind(new InetSocketAddress(port));
|
|
349
|
109 ssc.register(selector, SelectionKey.OP_ACCEPT,
|
|
|
110 new Forwarder(this));
|
|
6
|
111
|
|
144
|
112 sessionList = new LinkedList<Session>();
|
|
7
|
113 smList = new SessionManagerList();
|
|
144
|
114 editorList = new LinkedList<Editor>();
|
|
212
|
115 waitingCommandInMerge = new LinkedList<PacketSet>();
|
|
228
|
116
|
|
215
|
117
|
|
155
|
118 }
|
|
313
|
119
|
|
|
120 /*
|
|
|
121 * We wrote everything in one thread, but we can assign
|
|
|
122 * one thread for each communication channel and GUI event.
|
|
|
123 */
|
|
155
|
124
|
|
231
|
125 public void mainLoop() throws IOException {
|
|
0
|
126 while(true){
|
|
328
|
127 checkWaitingCommandInMerge();
|
|
313
|
128 if (checkInputEvent() ||
|
|
328
|
129 checkWaitingWrite()) {
|
|
313
|
130 // try to do fair execution for waiting task
|
|
|
131 if(selector.selectNow() > 0) select();
|
|
|
132 continue;
|
|
300
|
133 }
|
|
313
|
134 // now we can wait for input packet or event
|
|
233
|
135 selector.select();
|
|
144
|
136 select();
|
|
|
137 }
|
|
|
138 }
|
|
|
139
|
|
313
|
140 private boolean checkInputEvent() {
|
|
|
141 SessionManagerEvent e;
|
|
|
142 if((e = waitingEventQueue.poll())!=null){
|
|
334
|
143 e.exec(this);
|
|
313
|
144 return true;
|
|
|
145 }
|
|
|
146 return false;
|
|
|
147 }
|
|
|
148
|
|
|
149 private boolean checkWaitingWrite() throws IOException {
|
|
317
|
150 PacketSet p = writeQueue.poll();
|
|
|
151 if (p!=null) {
|
|
327
|
152 p.channel.write(p.command);
|
|
317
|
153 return true;
|
|
313
|
154 }
|
|
|
155 return false;
|
|
|
156 }
|
|
|
157
|
|
308
|
158 /**
|
|
|
159 * Check waiting command in merge
|
|
|
160 * @return true if there is a processed waiting command
|
|
|
161 * @throws IOException
|
|
|
162 */
|
|
346
|
163 private void checkWaitingCommandInMerge() {
|
|
328
|
164 List<PacketSet> w = waitingCommandInMerge;
|
|
|
165 waitingCommandInMerge = new LinkedList<PacketSet>();
|
|
|
166 for(PacketSet p: w) {
|
|
|
167 Editor e = p.getEditor();
|
|
353
|
168 if(e.isMerging()) { // still merging do nothing
|
|
328
|
169 waitingCommandInMerge.add(p);
|
|
|
170 } else {
|
|
346
|
171 try {
|
|
353
|
172 if (sessionManage(e, p.command)) { // we don't need this
|
|
|
173 assert false;
|
|
|
174 return;
|
|
|
175 }
|
|
|
176 e.manage(p.command);
|
|
347
|
177 } catch (Exception e1) {
|
|
349
|
178 // should be e.close()?
|
|
346
|
179 close(p.channel);
|
|
|
180 }
|
|
328
|
181 }
|
|
|
182 }
|
|
|
183 }
|
|
|
184
|
|
346
|
185 private void close(REPSocketChannel<REPCommand> channel) {
|
|
|
186 REPSelectionKey<REPCommand>key = channel.keyFor1(selector);
|
|
|
187 REPHandler handler = (REPHandler)key.attachment();
|
|
|
188 key.cancel();
|
|
|
189 handler.cancel(channel);
|
|
|
190 // we have to remove session/enditor
|
|
|
191 }
|
|
|
192
|
|
|
193
|
|
328
|
194 public boolean hasWaitingCommand(REPSocketChannel<REPCommand>c) {
|
|
|
195 for(PacketSet p:waitingCommandInMerge) {
|
|
|
196 if (p.channel==c) {
|
|
212
|
197 return true;
|
|
178
|
198 }
|
|
|
199 }
|
|
|
200 return false;
|
|
|
201 }
|
|
|
202
|
|
144
|
203 private void select() throws IOException {
|
|
231
|
204
|
|
|
205 Set<REPSelectionKey<REPCommand>> keys = selector.selectedKeys1();
|
|
|
206 for(REPSelectionKey<REPCommand> key : keys){
|
|
144
|
207 if(key.isAcceptable()){
|
|
199
|
208 REPSocketChannel<REPCommand> channel = key.accept(new REPCommandPacker());
|
|
337
|
209 logger.writeLog("SessionManager.select() : key.isAcceptable : channel = " + channel);
|
|
355
|
210 registerChannel (channel, new FirstConnector(this));
|
|
144
|
211 channel = null;
|
|
123
|
212
|
|
144
|
213 }else if(key.isReadable()){
|
|
212
|
214 REPHandler handler = (REPHandler)(key.attachment());
|
|
267
|
215 try {
|
|
|
216 handler.handle(key);
|
|
346
|
217 } catch (Exception x) {
|
|
267
|
218 key.cancel();
|
|
308
|
219 handler.cancel(key.channel1());
|
|
267
|
220 }
|
|
0
|
221 }
|
|
|
222 }
|
|
|
223 }
|
|
1
|
224
|
|
355
|
225 void registerChannel(REPSocketChannel<REPCommand> channel,Forwarder handler) throws IOException {
|
|
2
|
226 if(channel == null) {
|
|
|
227 return;
|
|
|
228 }
|
|
349
|
229 handler.setChannel(channel);
|
|
2
|
230 channel.configureBlocking(false);
|
|
355
|
231 channel.register(selector, SelectionKey.OP_READ, handler);
|
|
2
|
232 }
|
|
|
233
|
|
144
|
234
|
|
353
|
235 void cancel_sm_join() {
|
|
355
|
236 removeChannel(sm_join_channel);
|
|
|
237 sm_join_channel=null;
|
|
|
238 }
|
|
|
239
|
|
|
240
|
|
|
241 private void removeChannel(Forwarder sm_join_channel) {
|
|
|
242 REPSelectionKey<REPCommand> key = sm_join_channel.channel.keyFor1(selector);
|
|
|
243 key.cancel();
|
|
|
244 try {
|
|
|
245 sm_join_channel.channel.close();
|
|
|
246 } catch (IOException e) {
|
|
|
247 }
|
|
349
|
248 }
|
|
|
249
|
|
320
|
250
|
|
353
|
251 boolean hasSession(int sid) {
|
|
319
|
252 for(Session s:sessionList) {
|
|
|
253 if (s.getSID()==sid) return true;
|
|
|
254 }
|
|
|
255 return false;
|
|
|
256 }
|
|
|
257
|
|
|
258
|
|
|
259 void updateGUI() {
|
|
212
|
260 //リストのコピーをGUIに渡す
|
|
|
261 LinkedList<Session> sList = new LinkedList<Session>(sessionList);
|
|
|
262 LinkedList<Editor> eList = new LinkedList<Editor>(editorList);
|
|
|
263 //GUIに反映
|
|
|
264 Runnable doRun = new DoGUIUpdate(sList, eList, gui);
|
|
279
|
265 gui.invokeLater(doRun);
|
|
212
|
266 }
|
|
|
267
|
|
319
|
268 Forwarder getEditor(String hostport) {
|
|
178
|
269 for(Editor editor : editorList){
|
|
|
270 if(editor.getHost() == hostport){
|
|
|
271 return editor;
|
|
|
272 }
|
|
|
273 }
|
|
|
274 return null;
|
|
|
275 }
|
|
|
276
|
|
316
|
277 public Session getSession(int sid) throws IOException {
|
|
144
|
278 for(Session session : sessionList){
|
|
|
279 if(session.getSID() == sid) return session;
|
|
|
280 }
|
|
316
|
281 throw new IOException();
|
|
0
|
282 }
|
|
83
|
283
|
|
349
|
284 // private boolean setMaxHost(REPSocketChannel<REPCommand> channel, String maxHost2) {
|
|
|
285 // if(maxHost.compareTo(maxHost2) > 0){
|
|
|
286 // return false;
|
|
|
287 // }else{
|
|
|
288 // maxHost = maxHost2;
|
|
|
289 // return true;
|
|
|
290 // }
|
|
|
291 // }
|
|
139
|
292
|
|
353
|
293 void setMyHostName(String localHostName) {
|
|
308
|
294 myHost = localHostName + receive_port;
|
|
349
|
295 // if(maxHost == null) {
|
|
|
296 // maxHost = myHost;
|
|
|
297 // }
|
|
164
|
298 setHostToEditor(myHost);
|
|
|
299 }
|
|
|
300
|
|
|
301 private void setHostToEditor(String myHost2) {
|
|
|
302 for(Editor editor : editorList){
|
|
|
303 editor.setHost(myHost2);
|
|
|
304 }
|
|
76
|
305 }
|
|
0
|
306
|
|
355
|
307
|
|
|
308 /**
|
|
|
309 * Host 名のSession Manager に SM_JOIN する。自分は、Session を持っていては
|
|
|
310 * ならない。複数のSession Managerにjoinすることは出来ない。(NATを実装するまでは)。
|
|
|
311 * @param host
|
|
|
312 */
|
|
178
|
313 public void connectSession(String host) {
|
|
355
|
314 if (sm_join_channel!=null) return;
|
|
|
315 if (!sessionList.isEmpty()) return;
|
|
|
316 if (!smList.isMaster()) return;
|
|
334
|
317 int port = parent_port;
|
|
1
|
318 InetSocketAddress addr = new InetSocketAddress(host, port);
|
|
|
319 try {
|
|
186
|
320 REPSocketChannel<REPCommand> sessionchannel = REPSocketChannel.<REPCommand>create(new REPCommandPacker());
|
|
349
|
321
|
|
1
|
322 sessionchannel.connect(addr);
|
|
337
|
323 while(!sessionchannel.finishConnect());
|
|
349
|
324 Forwarder sm = new Forwarder(this);
|
|
355
|
325 registerChannel(sessionchannel, sm);
|
|
349
|
326 sm_join(sm);
|
|
1
|
327 }catch (IOException e) {
|
|
|
328 }
|
|
|
329 }
|
|
77
|
330
|
|
349
|
331 private void sm_join(Forwarder channel){
|
|
355
|
332 sm_join_channel = channel;
|
|
122
|
333 //SM_JOINコマンドを生成。
|
|
77
|
334 REPCommand command = new REPCommand();
|
|
|
335 command.setCMD(REP.SMCMD_SM_JOIN);
|
|
349
|
336 command.setEID(-1); // request Parent SessionManagerID
|
|
|
337 command.setSID(-1); // request SessionManagerID
|
|
79
|
338
|
|
122
|
339 //hostnameをセット。
|
|
349
|
340 setMyHostName(channel.getLocalHostName());
|
|
82
|
341
|
|
122
|
342 //XMLを生成。送信コマンドにセット。
|
|
355
|
343 //SessionXMLEncoder encoder = new SessionXMLEncoder(sessionList);
|
|
|
344 //String string = encoder.sessionListToXML();
|
|
|
345 String string = myHost;
|
|
77
|
346 command.setString(string);
|
|
|
347
|
|
122
|
348 //SM_JOINコマンドを送信。
|
|
349
|
349 channel.send(command);
|
|
122
|
350 //SessionManagerのListに追加。
|
|
349
|
351
|
|
77
|
352 }
|
|
349
|
353
|
|
316
|
354 public void selectSession(SelectButtonEvent event) throws IOException {
|
|
250
|
355 int sid = event.getSID();
|
|
164
|
356 Session session = getSession(sid);
|
|
227
|
357
|
|
320
|
358 Editor editor = (Editor)event.getEditor();
|
|
227
|
359 if(editor == null){
|
|
337
|
360 logger.writeLog("SessionManager.selectSession():editor = " + editor);
|
|
227
|
361 return;
|
|
|
362 }
|
|
324
|
363 if (editor.hasSession()) return;
|
|
349
|
364 //REPSocketChannel<REPCommand> channel = editor.getChannel();
|
|
319
|
365
|
|
324
|
366 // System.out.println("SessionManager.session.hasOnwer="+session.hasOwner());
|
|
158
|
367 if(session.hasOwner()){
|
|
320
|
368 editor.setEID(session.newEid());
|
|
322
|
369 editor.setSID(sid);
|
|
319
|
370 session.addForwarder(editor);
|
|
107
|
371 REPCommand sendCommand = new REPCommand();
|
|
|
372 sendCommand.setCMD(REP.SMCMD_JOIN_ACK);
|
|
148
|
373 sendCommand.setEID(editor.getEID());
|
|
107
|
374 sendCommand.setSID(sid);
|
|
286
|
375 sendCommand.string = "";
|
|
349
|
376 editor.send(sendCommand);
|
|
107
|
377 }else {
|
|
|
378 editor.setHost(myHost);
|
|
322
|
379 editor.setSID(sid);
|
|
320
|
380 session.addEditor(editor);
|
|
316
|
381 Forwarder owner = session.getOwner();
|
|
107
|
382
|
|
|
383 REPCommand command = new REPCommand();
|
|
|
384 command.setCMD(REP.SMCMD_SELECT);
|
|
|
385 command.setSID(sid);
|
|
178
|
386 command.setString(editor.getHost());
|
|
107
|
387 owner.send(command);
|
|
|
388 }
|
|
8
|
389 }
|
|
122
|
390
|
|
144
|
391 public void addWaitingCommand(PacketSet set) {
|
|
212
|
392 waitingCommandInMerge.add(set);
|
|
144
|
393 }
|
|
148
|
394
|
|
222
|
395 public void buttonPressed(SessionManagerEvent event) {
|
|
|
396 try {
|
|
308
|
397 waitingEventQueue.put(event);
|
|
222
|
398 } catch (InterruptedException e) {}
|
|
|
399 selector.wakeup();
|
|
|
400 }
|
|
281
|
401
|
|
|
402 public void syncExec(SessionManagerEvent event) {
|
|
|
403 try {
|
|
308
|
404 waitingEventQueue.put(event);
|
|
281
|
405 } catch (InterruptedException e) {
|
|
|
406 }
|
|
|
407 }
|
|
222
|
408
|
|
259
|
409 public void closeSession(SessionManagerEvent event) {
|
|
|
410 Session session = ((CloseButtonEvent) event).getSession();
|
|
|
411 session.closeSession();
|
|
|
412 sessionList.remove(session);
|
|
|
413 updateGUI();
|
|
|
414 }
|
|
|
415
|
|
274
|
416 public void remove(REPSocketChannel<REPCommand> channel) {
|
|
|
417 for(Session s:sessionList) {
|
|
|
418 if (s.deleteEditor(channel)) {
|
|
|
419 return ;
|
|
|
420 }
|
|
|
421 }
|
|
|
422 assert(false);
|
|
|
423 // can be other session manager? what should I do?
|
|
|
424 }
|
|
|
425
|
|
317
|
426
|
|
|
427 public void addWriteQueue(PacketSet packetSet) {
|
|
324
|
428 writeQueue.addLast(packetSet);
|
|
323
|
429 assert(writeQueue.size()<packetLimit) ;
|
|
317
|
430 }
|
|
|
431
|
|
318
|
432
|
|
|
433 public void remove(Editor editor) {
|
|
341
|
434 Session s0 = null;
|
|
|
435 editorList.remove(editor);
|
|
318
|
436 for(Session s:sessionList) {
|
|
341
|
437 if (s.deleteForwarder(editor)) {
|
|
|
438 if (editor.getEID()==0) s0=s;
|
|
|
439 }
|
|
318
|
440 }
|
|
341
|
441 if (s0!=null) removeSession(s0);
|
|
|
442 updateGUI();
|
|
|
443 }
|
|
|
444
|
|
|
445 private void removeSession(Session s0) {
|
|
|
446 sessionList.remove(s0);
|
|
|
447 // send UPDATE to all the session manager
|
|
318
|
448 }
|
|
|
449
|
|
334
|
450 public void setParentPort(int port) {
|
|
|
451 parent_port = port;
|
|
|
452 }
|
|
|
453 public int getParentPort() {
|
|
|
454 return parent_port;
|
|
|
455 }
|
|
|
456
|
|
|
457 public int getPort() {
|
|
|
458 return receive_port;
|
|
|
459 }
|
|
|
460
|
|
353
|
461
|
|
355
|
462 boolean sessionManage(Forwarder forwarder, REPCommand command) throws ClosedChannelException,
|
|
353
|
463 IOException, SAXException {
|
|
355
|
464 switch(command.cmd){
|
|
353
|
465
|
|
|
466 // Session Manager Command
|
|
|
467
|
|
|
468 case SMCMD_JOIN:
|
|
|
469 {
|
|
|
470 updateGUI();
|
|
|
471 }
|
|
|
472
|
|
|
473 break;
|
|
|
474
|
|
|
475 case SMCMD_JOIN_ACK:
|
|
|
476 assert (false);
|
|
|
477 break;
|
|
|
478
|
|
|
479 case SMCMD_PUT:
|
|
|
480 {
|
|
|
481 //Sessionを生成
|
|
|
482 // sessionIDってglobaly uniqueだから、本来は、
|
|
|
483 // 自分の親に作ってもらう必要がある。自分が親なら自分で作って良い。
|
|
|
484
|
|
|
485 int sid = sessionList.size();
|
|
355
|
486 Editor editor = (Editor) forwarder;
|
|
|
487 Session session = new Session(sid, command.string, (Editor)forwarder);
|
|
353
|
488 session.hasOwner(true);
|
|
|
489 sessionList.add(session);
|
|
|
490
|
|
|
491 updateGUI();
|
|
|
492
|
|
|
493 //エディタにAckを送信
|
|
355
|
494 REPCommand sendCommand = new REPCommand(command);
|
|
353
|
495 sendCommand.setCMD(REP.SMCMD_PUT_ACK);
|
|
|
496 sendCommand.setEID(editor.getEID());
|
|
|
497 sendCommand.setSID(session.getSID());
|
|
|
498 editor.send(sendCommand);
|
|
|
499
|
|
|
500 //他のSessionManagerへSessionの追加を報告
|
|
|
501 //親に送って、親から子へ
|
|
|
502 SessionXMLEncoder sessionEncoder = new SessionXMLEncoder(session);
|
|
355
|
503 REPCommand command1 = new REPCommand();
|
|
|
504 command1.setSID(session.getSID());
|
|
|
505 command1.setString(sessionEncoder.sessionListToXML());
|
|
|
506 command1.setCMD(REP.SMCMD_UPDATE);
|
|
353
|
507 smList.sendToSlaves( command);
|
|
|
508
|
|
|
509 }
|
|
|
510
|
|
|
511 break;
|
|
|
512
|
|
|
513 // SELECT is no longer used in a editor. Select
|
|
|
514 // operation is handled in Session Manager Only
|
|
|
515 case SMCMD_SELECT:
|
|
|
516 {
|
|
|
517 //他のSessionManagerをエディタとしてSessionに追加
|
|
|
518 Forwarder next = new Forwarder(this);
|
|
|
519 next.setChannel(forwarder.channel);
|
|
355
|
520 Session session = getSession(command.sid);
|
|
353
|
521 session.addForwarder(next);
|
|
|
522
|
|
|
523 if(session.hasOwner()){
|
|
|
524 //このSessionManagerがオーナーを持っている場合、Sessionにエディタを追加し、エディタへAckを返す
|
|
355
|
525 REPCommand sendCommand = new REPCommand(command);
|
|
353
|
526 sendCommand.setCMD(REP.SMCMD_SELECT_ACK);
|
|
|
527 sendCommand.setEID(next.getEID());
|
|
|
528 next.send(sendCommand);
|
|
|
529 }else{
|
|
|
530 //オーナーを持ってない場合は、オーナーを持っているSessionManagerへSELECTコマンドを中継する
|
|
|
531 Forwarder owner = session.getOwner();
|
|
355
|
532 owner.send(command);
|
|
353
|
533 }
|
|
|
534 }
|
|
|
535
|
|
|
536 break;
|
|
|
537
|
|
|
538 case SMCMD_SELECT_ACK:
|
|
|
539 {
|
|
355
|
540 String hostport = command.string;
|
|
353
|
541 Forwarder editor1 = getEditor(hostport);
|
|
|
542
|
|
|
543 if(editor1 != null) {
|
|
|
544 //host, port を見て、このコマンドが自分が送信したSelectコマンドのAckかどうかを判断する
|
|
355
|
545 REPCommand command1 = new REPCommand();
|
|
|
546 command1.setCMD(REP.SMCMD_JOIN_ACK);
|
|
|
547 command1.setSID(command.sid);
|
|
|
548 command1.setEID(command.eid);
|
|
353
|
549 editor1.send(command);
|
|
|
550
|
|
|
551 }else{
|
|
|
552 //自分が送信したコマンドでなければ、次のSessionManagerへ中継する
|
|
355
|
553 smList.sendToSlaves(command);
|
|
353
|
554 }
|
|
|
555 }
|
|
|
556
|
|
|
557 break;
|
|
|
558 case SMCMD_SM_JOIN:
|
|
|
559
|
|
|
560 {
|
|
355
|
561 // SM_JOIN中にSMCMD_SM_JOINが来たら、これはループなので、
|
|
|
562 ///自分のSM_JOINを取り消す。
|
|
|
563 if (sm_join_channel!=null) cancel_sm_join();
|
|
353
|
564 // SMCMD_SM_JOIN は、master まで上昇する。
|
|
|
565 // masterでなければ、自分のparentに転送する。
|
|
355
|
566 if(isMaster()) {
|
|
353
|
567 // master であれば、SessionManager IDを決めて、
|
|
|
568 // 自分のsmList に登録
|
|
355
|
569 Forwarder sm;
|
|
|
570 int psid = command.eid;
|
|
|
571 if (forwarder.sid!=-1) {
|
|
|
572 // すでに channelはSessionManager Idを持っていて、
|
|
|
573 // direct link ではないので、
|
|
|
574 // channel を持たないForwarderとして登録する
|
|
|
575 sm = new Forwarder(this);
|
|
|
576 } else {
|
|
|
577 sm = forwarder;
|
|
|
578 }
|
|
|
579 int sid = smList.addNewSessionManager(sm,command);
|
|
|
580 REPCommand sendCommand = makeREPCommandWithSessionList(REP.SMCMD_SM_JOIN_ACK);
|
|
|
581 // command.eid==smList.sesionManagerID() の場合は、
|
|
|
582 // 待っている自分の下のsessionManagerにsidをassignする必要がある。
|
|
353
|
583 sendCommand.setSID(sid); // new Session manager ID
|
|
|
584 // 複数のSM_JOIN_ACKを識別するには、最初にSM_JOINを受け付けた
|
|
|
585 // Session manager IDを使う。
|
|
355
|
586 sendCommand.setEID(psid);
|
|
|
587 send_sm_join_ack(psid, sid, sendCommand);
|
|
|
588 } else {
|
|
|
589 if (forwarder.sid==-1) {
|
|
|
590 // direct link の場合は、識別のために、EIDに直上の
|
|
|
591 // smid を入れておく。
|
|
|
592 command.setEID(smList.sessionManagerID());
|
|
|
593 }
|
|
|
594 smList.sendToMaster(command);
|
|
353
|
595 }
|
|
|
596 }
|
|
|
597 break;
|
|
|
598
|
|
355
|
599 case SMCMD_SM_JOIN_ACK:
|
|
|
600 send_sm_join_ack(command.eid, command.sid, command);
|
|
353
|
601 break;
|
|
|
602
|
|
|
603 case SMCMD_UPDATE:
|
|
|
604 {
|
|
355
|
605 SessionList receivedSessionList3 = decoder.decode(command.string);
|
|
353
|
606
|
|
|
607 //UPDATEコマンドにより送られてきたSessionの情報を追加する
|
|
|
608 LinkedList<Session> list = receivedSessionList3.getList();
|
|
|
609 for(Session session : list){
|
|
|
610 session.getEditorList().get(0).setChannel(forwarder.channel);
|
|
|
611 sessionList.add(session);
|
|
|
612 }
|
|
|
613
|
|
|
614 //他のSessionManagerへ中継する
|
|
355
|
615 smList.sendToSlaves(command);
|
|
353
|
616
|
|
|
617 updateGUI();
|
|
|
618 }
|
|
|
619 break;
|
|
|
620
|
|
|
621 case SMCMD_UPDATE_ACK:
|
|
|
622 {
|
|
355
|
623 if(!hasSession(command.sid)) {
|
|
353
|
624 // accept new Session
|
|
|
625 // ここで初めてsession id が決まる。
|
|
|
626 // このコマンドは、master session manager が出すはず
|
|
|
627 Forwarder sm = new Forwarder(this);
|
|
|
628 sm.setChannel(forwarder.channel);
|
|
355
|
629 Session session = new Session(command.sid,command.string,null);
|
|
353
|
630 session.addForwarder(sm);
|
|
|
631
|
|
|
632 sessionList.add(session);
|
|
|
633
|
|
|
634 updateGUI();
|
|
|
635 }
|
|
355
|
636 smList.sendToSlaves(command);
|
|
353
|
637 }
|
|
|
638 break;
|
|
|
639
|
|
|
640 // case SMCMD_CH_MASTER:
|
|
|
641 // {
|
|
|
642 // //maxHost を設定。
|
|
|
643 // if(setMaxHost(channel, receivedCommand.string)){
|
|
|
644 // REPCommand sendCommand = new REPCommand();
|
|
|
645 // sendCommand.setCMD(REP.SMCMD_CH_MASTER);
|
|
|
646 // sendCommand.setString(maxHost);
|
|
|
647 // smList.sendExcept(channel, sendCommand);
|
|
|
648 // }
|
|
|
649 // }
|
|
|
650 // break;
|
|
|
651
|
|
|
652
|
|
|
653 default:
|
|
|
654 return false;
|
|
|
655 }
|
|
|
656 return true;
|
|
|
657 }
|
|
|
658
|
|
355
|
659
|
|
|
660 void send_sm_join_ack(int psid, int sid,REPCommand sendCommand) {
|
|
|
661 if (psid==smList.sessionManagerID()) {
|
|
|
662 // 直下のsessionManagerにIDを割り振る必要がある。
|
|
|
663 smList.assignSessionManagerIDtoWaitingSM(sid);
|
|
|
664 // ここで smList に一つだけ追加されるので
|
|
|
665 // 待っている最初のsm一つにだけ、sm_joinが新たに送られる。
|
|
|
666 }
|
|
|
667 smList.sendToSlaves(sendCommand);
|
|
|
668 }
|
|
|
669
|
|
|
670
|
|
|
671 private REPCommand makeREPCommandWithSessionList(REP cmd) {
|
|
|
672 //SessionListからXMLを生成。
|
|
|
673 //joinしてきたSessionManagerに対してACKを送信。
|
|
|
674 SessionXMLEncoder sessionlistEncoder = new SessionXMLEncoder(sessionList);
|
|
|
675 REPCommand sendCommand = new REPCommand();
|
|
|
676 sendCommand.setCMD(cmd);
|
|
|
677 sendCommand.setString(sessionlistEncoder.sessionListToXML());
|
|
|
678 return sendCommand;
|
|
|
679 }
|
|
|
680
|
|
|
681
|
|
|
682 public boolean isMaster() {
|
|
|
683 return smList.isMaster();
|
|
|
684 }
|
|
|
685
|
|
|
686
|
|
|
687 public void setSessionManagerID(int sid) {
|
|
|
688 smList.setSessionManagerID(sid);
|
|
|
689 }
|
|
|
690
|
|
0
|
691 }
|
