Mercurial > hg > RemoteEditor > REPSessionManager
view rep/SessionManager.java @ 228:e6c7a56ff7f1
*** empty log message ***
author | pin |
---|---|
date | Sun, 31 Aug 2008 15:27:04 +0900 |
parents | cbd67817e9cd |
children | f816e0cbe6fd |
line wrap: on
line source
package rep; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.channels.SelectableChannel; import java.nio.channels.SelectionKey; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import javax.swing.SwingUtilities; import rep.channel.REPServerSocketChannel; import rep.channel.REPSocketChannel; import rep.handler.PacketSet; import rep.handler.REPHandler; import rep.handler.REPHandlerImpl; import rep.handler.REPHandlerInMerge; import rep.channel.REPSelector; import rep.xml.SessionXMLDecoder; import rep.xml.SessionXMLEncoder; import rep.channel.REPSelectionKey; /* //+-------+--------+--------+-------+--------+---------+------+ //| cmd | session| editor | seqid | lineno | textsiz | text | //| | id | id | | | | | //+-------+--------+--------+-------+--------+---------+------+ //o-------header section (network order)-------------o int cmd; // command int sid; // session ID : uniqu to editing file int eid; // editor ID : owner editor ID = 1。Session に対して unique int seqno; // Sequence number : sequence number はエディタごとに管理 int lineno; // line number int textsize; // textsize : bytesize byte[] text; */ public class SessionManager implements SessionManagerEventListener{ private LinkedList<Session> sessionList; private SessionManagerGUI gui; private REPSelector<REPCommand> selector; private SessionManagerList smList; private String myHost; private boolean isMaster = true; private List<Editor> editorList; private String maxHost; private List<PacketSet> waitingCommandInMerge; private BlockingQueue<SessionManagerEvent> waitingQueue; private static int temp_port; private static int send_port; static final int DEFAULT_PORT = 8766; public SessionManager(int port) { } public void openSelector() throws IOException{ selector = REPSelector.create(); } public void init(int port) throws InterruptedException, IOException { REPServerSocketChannel<REPCommand> ssc = REPServerSocketChannel.<REPCommand>open(new REPCommandPacker()); ssc.configureBlocking(false); //reuse address 必須 ssc.socket().setReuseAddress(true); //getAllByNameで取れた全てのアドレスに対してbindする ssc.socket().bind(new InetSocketAddress(port)); ssc.register(selector, SelectionKey.OP_ACCEPT, new REPHandlerImpl(-1, this)); sessionList = new LinkedList<Session>(); smList = new SessionManagerList(); editorList = new LinkedList<Editor>(); waitingCommandInMerge = new LinkedList<PacketSet>(); waitingQueue = new LinkedBlockingQueue<SessionManagerEvent>(); //デフォルトのSessionを作っておく(テスト用に?) if(sessionList.size() > 0) System.out.println("Error : SessionManager.init():"); Session defaultSession = new Session(sessionList.size(), "DefaultSession.txt", new Editor(0,null)); sessionList.add(defaultSession); } private void mainLoop() throws IOException { while(true){ if(checkSend()){ if(selector.selectNow() > 0){ select(); } continue; } int i = selector.select(); System.out.println(i); select(); } } private boolean checkSend() { for(Iterator<PacketSet> it = waitingCommandInMerge.iterator(); it.hasNext();){ PacketSet p = it.next(); if(p.getEditor().isMerging()) { continue; }else{ manage(p.channel, p.command); it.remove(); return true; } } return false; } private void select() throws IOException { SessionManagerEvent e; while((e = waitingQueue.poll())!=null){ e.exec(); } for(REPSelectionKey<REPCommand> key : selector.selectedKeys1()){ if(key.isAcceptable()){ /*** serverChannelはenableになったSelectionKeyのchannel ***/ REPSocketChannel<REPCommand> channel = key.accept(new REPCommandPacker()); registerChannel (selector, channel, SelectionKey.OP_READ); channel = null; }else if(key.isReadable()){ REPHandler handler = (REPHandler)(key.attachment()); handler.handle(key); } } } private void registerChannel(REPSelector selector, SelectableChannel channel, int ops) throws IOException { if(channel == null) { return; } channel.configureBlocking(false); REPHandler handler = new REPHandlerImpl(-1, this); channel.register(selector, ops, handler); } public void manage(REPSocketChannel<REPCommand> channel, REPCommand receivedCommand) { if(receivedCommand == null) return; //Session session; REPCommand sendCommand = new REPCommand(receivedCommand); REPSocketChannel<REPCommand> send = channel; switch(receivedCommand.cmd){ case REP.SMCMD_JOIN: { //どのSessionにも属さないエディタをリストに追加 //エディタとchannelは1対1 //エディタが新しくputする場合は新しくソケットを作る Editor editor = new Editor(editorList.size(), channel); editor.setHost(myHost); editorList.add(editor); guiUpdate(); } break; case REP.SMCMD_JOIN_ACK: assert (false); break; case REP.SMCMD_PUT: { //エディタのリストに追加 Editor editor = new Editor(editorList.size(), channel); //editorList.add(editor); //Sessionを生成 int sid = sessionList.size(); editor = new Editor(0, channel); editor.setHost(myHost); Session session = new Session(sid, receivedCommand.string, editor); session.hasOwner(true); sessionList.add(session); guiUpdate(); //エディタにAckを送信 sendCommand.setCMD(REP.SMCMD_PUT_ACK); sendCommand.setEID(editor.getEID()); sendCommand.setSID(session.getSID()); editor.send(sendCommand); //他のSessionManagerへSessionの追加を報告 //親に送って、親から子へ SessionXMLEncoder sessionEncoder = new SessionXMLEncoder(session); REPCommand command = new REPCommand(); command.setSID(session.getSID()); command.setString(sessionEncoder.sessionListToXML()); command.setCMD(REP.SMCMD_UPDATE); smList.sendExcept(channel, command); } break; case REP.SMCMD_SELECT: { //他のSessionManagerをエディタとしてSessionに追加 Editor editor = new Editor(channel); Session session = getSession(receivedCommand.sid); session.addEditor(editor); if(session.hasOwner()){ //このSessionManagerがオーナーを持っている場合、Sessionにエディタを追加し、エディタへAckを返す sendCommand.setCMD(REP.SMCMD_SELECT_ACK); sendCommand.setEID(editor.getEID()); editor.send(sendCommand); }else{ //オーナーを持ってない場合は、オーナーを持っているSessionManagerへSELECTコマンドを中継する Editor owner = session.getOwner(); owner.send(receivedCommand); } } break; case REP.SMCMD_SELECT_ACK: { String hostport = receivedCommand.string; Editor editor = getEditor(hostport); if(editor != null) { //host, port を見て、このコマンドが自分が送信したSelectコマンドのAckかどうかを判断する REPCommand command = new REPCommand(); command.setCMD(REP.SMCMD_JOIN_ACK); command.setSID(receivedCommand.sid); command.setEID(receivedCommand.eid); editor.send(command); }else{ //自分が送信したコマンドでなければ、次のSessionManagerへ中継する smList.sendExcept(channel, receivedCommand); } } break; case REP.SMCMD_SM_JOIN: { //SessionManagerのリストへ追加 smList.add(channel); //XMLからSessionListオブジェクトを生成する。 SessionXMLDecoder decoder = new SessionXMLDecoder(); SessionList receivedSessionList = decoder.decode(receivedCommand.string); //myHost を設定。 //立ち上げ時にやるとlocalhostしか取れない if(myHost == null) setMyHostName(getLocalHostName(channel)); //maxHost を設定。 if(setMaxHost(channel, receivedSessionList.getMaxHost())){ sendCommand = new REPCommand(); sendCommand.setCMD(REP.SMCMD_CH_MASTER); sendCommand.setString(maxHost); smList.sendExcept(channel, sendCommand); } //SessionListからXMLを生成。 //joinしてきたSessionManagerに対してACKを送信。 SessionXMLEncoder sessionlistEncoder = new SessionXMLEncoder(sessionList); sendCommand = new REPCommand(); sendCommand.setCMD(REP.SMCMD_SM_JOIN_ACK); sendCommand.setString(sessionlistEncoder.sessionListToXML()); send.write(sendCommand); //その他の SessionManager に対して SMCMD_UPDATEを 送信。 sendCommand = new REPCommand(); sendCommand.setCMD(REP.SMCMD_UPDATE); sendCommand.setString(receivedCommand.string); smList.sendExcept(channel, sendCommand); } break; case REP.SMCMD_SM_JOIN_ACK: //XMLからSessionListオブジェクトを生成。 SessionXMLDecoder decoder2 = new SessionXMLDecoder(); SessionList receivedSessionList2 = decoder2.decode(receivedCommand.string); //maxHostを決定。 if(setMaxHost(channel, receivedSessionList2.getMaxHost())){ sendCommand = new REPCommand(); sendCommand.setCMD(REP.SMCMD_CH_MASTER); sendCommand.setString(maxHost); smList.sendExcept(channel, sendCommand); } break; case REP.SMCMD_UPDATE: { SessionXMLDecoder decoder3 = new SessionXMLDecoder(); SessionList receivedSessionList3 = decoder3.decode(receivedCommand.string); //UPDATEコマンドにより送られてきたSessionの情報を追加する LinkedList<Session> list = receivedSessionList3.getList(); for(Session session : list){ session.getEditorList().get(0).setChannel(channel); sessionList.add(session); } //他のSessionManagerへ中継する smList.sendExcept(channel, receivedCommand); //リストのコピーをGUIに渡す LinkedList<Session> sList = new LinkedList<Session>(sessionList); LinkedList<Editor> eList = new LinkedList<Editor>(editorList); //GUIに反映 Runnable doRun = new DoGUIUpdate(sList, eList, gui); SwingUtilities.invokeLater(doRun); } break; case REP.SMCMD_UPDATE_ACK: { if(receivedCommand.sid > sessionList.size()){ Editor editor = new Editor(channel); editor.setName(receivedCommand.string); Session session = new Session(editor); session.addEditor(editor); sessionList.add(session); //リストのコピーをGUIに渡す LinkedList<Session> sList = new LinkedList<Session>(sessionList); LinkedList<Editor> eList = new LinkedList<Editor>(editorList); //GUIに反映 Runnable doRun = new DoGUIUpdate(sList, eList, gui); SwingUtilities.invokeLater(doRun); } smList.sendToSlave(receivedCommand); } break; case REP.SMCMD_CH_MASTER: { //maxHost を設定。 if(setMaxHost(channel, receivedCommand.string)){ sendCommand = new REPCommand(); sendCommand.setCMD(REP.SMCMD_CH_MASTER); sendCommand.setString(maxHost); smList.sendExcept(channel, sendCommand); } } break; case REP.REPCMD_DELETE: case REP.REPCMD_INSERT: { //sid から Session を取得 Session session = getSession(receivedCommand.sid); //マージの処理と次のエディタへコマンドを送信する処理 session.translate(channel, receivedCommand); Editor editor = session.getEditor(channel); Editor prevEditor = session.getPrevEditor(editor); //マージ中のエディタはコマンドを受け取らない if(editor.isMerging()){ //Handlerを切り替える setMergeState(prevEditor.getChannel(), session.getSID()); }else { setNormalState(prevEditor.getChannel(), session.getSID()); } } break; default: assert(false); break; } } private void guiUpdate() { //リストのコピーをGUIに渡す LinkedList<Session> sList = new LinkedList<Session>(sessionList); LinkedList<Editor> eList = new LinkedList<Editor>(editorList); //GUIに反映 Runnable doRun = new DoGUIUpdate(sList, eList, gui); SwingUtilities.invokeLater(doRun); } private void setNormalState(REPSocketChannel<REPCommand> channel, int sid) { SelectionKey key = channel.keyFor(selector); key.attach(new REPHandlerImpl(sid, this)); } private void setMergeState(REPSocketChannel<REPCommand> channel, int sid) { SelectionKey key = channel.keyFor(selector); key.attach(new REPHandlerInMerge(sid, this)); } private Editor getEditor(String hostport) { for(Editor editor : editorList){ if(editor.getHost() == hostport){ return editor; } } return null; } public Editor getEditor(REPSocketChannel<REPCommand> channel){ for(Editor editor : editorList){ if(editor.getChannel() == channel){ return editor; } } return null; } private Session getSession(int sid) { for(Session session : sessionList){ if(session.getSID() == sid) return session; } return null; } private boolean setMaxHost(REPSocketChannel<REPCommand> channel, String maxHost2) { if(maxHost.compareTo(maxHost2) > 0){ return false; }else{ maxHost = maxHost2; return true; } } private void setMyHostName(String localHostName) { myHost = localHostName + temp_port; if(maxHost == null) { maxHost = myHost; } setHostToEditor(myHost); } private void setHostToEditor(String myHost2) { for(Editor editor : editorList){ editor.setHost(myHost2); } } public static void main(String[] args) throws InterruptedException, IOException { int port = DEFAULT_PORT; int port_s = DEFAULT_PORT; //System.setProperty("file.encoding", "UTF-8"); if(args.length > 0){ port = Integer.parseInt(args[0]); port_s = Integer.parseInt(args[1]); } temp_port = port; send_port = port_s; SessionManager sm = new SessionManager(port); sm.openSelector(); sm.init(port); sm.startGUI(sm); sm.mainLoop(); } private void startGUI(SessionManager sm) { gui = new SessionManagerGUI(); Thread th = new Thread( gui ); th.start(); gui.addREPActionListener(this); } public void connectSession(String host) { int port = DEFAULT_PORT; port = send_port; InetSocketAddress addr = new InetSocketAddress(host, port); try { REPSocketChannel<REPCommand> sessionchannel = REPSocketChannel.<REPCommand>create(new REPCommandPacker()); sessionchannel.configureBlocking(true); sessionchannel.connect(addr); while(!sessionchannel.finishConnect()){ System.out.print("test afro"); } System.out.println(""); registerChannel(selector, sessionchannel, SelectionKey.OP_READ); sm_join(sessionchannel); }catch (IOException e) { e.printStackTrace(); } } private void sm_join(REPSocketChannel<REPCommand> channel){ //SM_JOINコマンドを生成。 REPCommand command = new REPCommand(); command.setCMD(REP.SMCMD_SM_JOIN); //hostnameをセット。 setMyHostName(getLocalHostName(channel)); //XMLを生成。送信コマンドにセット。 SessionXMLEncoder encoder = new SessionXMLEncoder(sessionList); String string = encoder.sessionListToXML(); command.setString(string); //SM_JOINコマンドを送信。 channel.write(command); //SessionManagerのListに追加。 smList.add(channel); } private String getLocalHostName(REPSocketChannel channel) { String host = null; host = channel.socket().getLocalAddress().getHostName(); return host; } // public void connectionOccured(SessionManagerEvent event) { // try { // waitingQueue.put(event); // } catch (InterruptedException e) { // } // selector.wakeup(); // } // // public void selectOccured(SessionManagerEvent event) { // try { // waitingQueue.put(event); // } catch (InterruptedException e) { // e.printStackTrace(); // } // selector.wakeup(); // } public void selectSession(SelectButtonEvent event) { REPSocketChannel<REPCommand> channel = event.getEditorChannel(); int sid = event.getSID(); Session session = getSession(sid); Editor editor = getEditor(channel); if(editor == null){ System.out.println("SessionManager.selectSession():editor = " + editor); return; } session.addEditor(editor); System.out.println(session.hasOwner()); if(session.hasOwner()){ REPCommand sendCommand = new REPCommand(); sendCommand.setCMD(REP.SMCMD_JOIN_ACK); sendCommand.setEID(editor.getEID()); sendCommand.setSID(sid); channel.write(sendCommand); }else { sid = event.getSID(); editor = new Editor(channel); editor.setHost(myHost); session = getSession(sid); session.addEditor(editor); Editor owner = session.getOwner(); REPCommand command = new REPCommand(); command.setCMD(REP.SMCMD_SELECT); command.setSID(sid); command.setString(editor.getHost()); owner.send(command); } } public void addWaitingCommand(PacketSet set) { waitingCommandInMerge.add(set); } public List<Session> getSessionList() { return sessionList; } public List<Editor> getEditorList() { return editorList; } public void buttonPressed(SessionManagerEvent event) { try { waitingQueue.put(event); } catch (InterruptedException e) {} selector.wakeup(); } }