Mercurial > hg > RemoteEditor > REPSessionManager
view rep/SessionManager.java @ 167:3dc194f5e28f
*** empty log message ***
author | pin |
---|---|
date | Thu, 28 Aug 2008 18:44:05 +0900 |
parents | 3841cc75b808 |
children | 2974db051dd8 |
line wrap: on
line source
package rep; import java.io.IOException; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.channels.SelectableChannel; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.nio.charset.CharacterCodingException; import java.nio.charset.Charset; import java.nio.charset.CharsetEncoder; import java.util.LinkedList; import java.util.List; import java.util.Set; import java.util.StringTokenizer; import rep.channel.REPServerSocketChannel; import rep.channel.REPSocketChannel; import rep.channel.SelectionKeySimulator; import rep.channel.SelectorSimulator; 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; //+-------+--------+--------+-------+--------+---------+------+ //| 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 ConnectionListener, REPActionListener{ //private SessionList sessionlist; private LinkedList<Session> sessionList; private SessionManagerGUI gui; private Selector selector; private SessionManagerList smList; private String myHost; private boolean isMaster = true; //private EditorList ownEditorList; private List<Editor> editorList; private String maxHost; private boolean isSimulation; private List<PacketSet> packetSetList; //private List<SessionManagerNode> managerList; private static int temp_port; private static int send_port; static final int DEFAULT_PORT = 8766; public SessionManager(int port) { gui = new SessionManagerGUI(); } public void openSelector() throws IOException{ selector = REPSelector.open(); } public void init(int port) throws InterruptedException, IOException { REPServerSocketChannel<REPCommand> ssc = REPServerSocketChannel.<REPCommand>open(); ssc.configureBlocking(false); //reuse address 必須 ssc.socket().setReuseAddress(true); ssc.socket().bind(new InetSocketAddress(port)); ssc.register(selector, SelectionKey.OP_ACCEPT); //sessionlist = new SessionList(); sessionList = new LinkedList<Session>(); smList = new SessionManagerList(); //ownEditorList = new EditorList(); editorList = new LinkedList<Editor>(); packetSetList = new LinkedList<PacketSet>(); // main loop mainLoop(); } private void mainLoop() throws IOException { while(true){ selector.select(); select(); } } private void select() throws IOException { for(SelectionKey key : selector.selectedKeys()){ if(key.isAcceptable()){ /*** serverChannelはenableになったSelectionKeyのchannel ***/ REPServerSocketChannel serverChannel = (REPServerSocketChannel)key.channel(); REPSocketChannel channel = serverChannel.accept(); //keyからchannelを取って、accept registerChannel (selector, channel, SelectionKey.OP_READ); channel = null; }else if(key.isReadable()){ // REPSocketChannel<REPCommand> channel = (REPSocketChannel<REPCommand>)key.channel(); // REPPacketReceive receive = new REPPacketReceive(channel); // receive.setkey(key); // REPCommand receivedCommand = receive.unpackUConv(); // manage(channel, receivedCommand); REPHandler handler = (REPHandler)key.attachment(); handler.handle(key); }else if(key.isConnectable()){ System.out.println("Connectable"); } } } private synchronized void registerChannel(Selector selector, SelectableChannel channel, int ops) throws IOException { if(channel == null) { return; } channel.configureBlocking(false); selector.wakeup(); REPHandler handler = new REPHandlerImpl(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); REPPacketSend send = new REPPacketSend(channel); switch(receivedCommand.cmd){ case REP.SMCMD_JOIN: { //どのSessionにも属さないエディタをリストに追加 Editor editor = new Editor(editorList.size(), channel); editor.setHost(myHost); editorList.add(editor); //GUIに反映 gui.setComboEditor(editor.getEID(), channel); } break; case REP.SMCMD_JOIN_ACK: 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, editor); session.hasOwner(true); sessionList.add(new Session(sid, editor)); //GUIに反映 gui.setComboSession(session.getSID(), session.getName()); gui.setComboEditor(editor.getEID(), editor.getChannel()); //エディタに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: { //エディタを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); //SessionListへ追加し変換テーブルを生成する。 //sessionlist.update(channel, receivedSessionList); //myHost を設定。 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.send(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); //SessionListへ追加し変換テーブルを生成する。 //sessionlist.update(channel, receivedSessionList3); smList.sendExcept(channel, receivedCommand); for(Session session3 : receivedSessionList3.getList()){ gui.setComboSession(session3.getSID(), session3.getName()); } 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.setComboSession(session.getSID(), session.getName()); } smList.sendToSlave(receivedCommand); break; // case REP.REPCMD_READ: // //sessionlist.sendCmd(channel, repCmd); // 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; default: { //sid から Session を取得 Session session = getSession(receivedCommand.sid); //マージの処理と次のエディタへコマンドを送信する処理 session.translate(channel, receivedCommand); //マージ中は前のエディタからのコマンドを受信しない Editor editor = session.getEditor(channel); Editor prevEditor = session.getPrevEditor(editor); if(editor.isFinished){ set } if(editor.isMerging()){ //Handlerを切り替える setMergeState(prevEditor.getChannel(), session.getSID()); } } break; } } private void setMergeState(REPSocketChannel<REPCommand> channel, int sid) { SelectionKey key = channel.keyFor(selector); key.attach(new REPHandlerInMerge(sid, this)); } private Editor getEditor(String hostport) { return null; } private Session getSession(int sid) { for(Session session : sessionList){ if(session.getSID() == sid) return session; } return null; } private boolean setMaxHost(REPSocketChannel channel, String maxHost2) { // TODO Auto-generated method stub return false; } 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.openWindow(); sm.init(port); sm.mainLoop(); } private void openWindow() { Thread th = new Thread( gui ); th.start(); //System.out.println(sessionmanagerGUI.toString()); gui.addConnectionListener(this); gui.addREPActionListener(this); } private void connectSession(String host) { int port = DEFAULT_PORT; port = send_port; InetSocketAddress addr = new InetSocketAddress(host, port); try { REPSocketChannel<REPCommand> sessionchannel = REPSocketChannel.<REPCommand>create(); 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コマンドを送信。 REPPacketSend send = new REPPacketSend(channel); send.send(command); //SessionManagerのListに追加。 smList.add(channel); } private String getLocalHostName(REPSocketChannel channel) { String host = null; host = channel.socket().getLocalAddress().getHostName(); return host; } public void connectionOccured(ConnectionEvent event) { connectSession(event.getHost()); } public void ActionOccured(REPActionEvent event) { REPSocketChannel<REPCommand> channel = event.getEditorChannel(); int sid = event.getSID(); Session session = getSession(sid); if(session.hasOwner()){ Editor editor = new Editor(channel); session.addEditor(new Editor(channel)); REPCommand sendCommand = new REPCommand(); sendCommand.setCMD(REP.SMCMD_JOIN_ACK); sendCommand.setEID(editor.getEID()); sendCommand.setSID(sid); REPPacketSend sender = new REPPacketSend(channel); sender.send(sendCommand); }else { REPSocketChannel<REPCommand> editorChannel = event.getEditorChannel(); sid = event.getSID(); Editor editor = new Editor(editorChannel); 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() + ":" + editor.getPort()); owner.send(command); } } public void addWaitingCommand(PacketSet set) { // TODO Auto-generated method stub packetSetList.add(set); } public void undo() { // TODO Auto-generated method stub } }