Mercurial > hg > Members > tatsuki > Alice
view src/alice/daemon/IncomingTcpConnection.java @ 51:919389066887
change protocol header: remove an integer representing message length (work?)
author | kazz <kazz@cr.ie.u-ryukyu.ac.jp> |
---|---|
date | Sun, 05 Feb 2012 06:26:26 +0900 |
parents | ff33af300567 |
children | f9c82d9a4936 |
line wrap: on
line source
package alice.daemon; import java.io.EOFException; import java.io.IOException; import java.nio.channels.ClosedChannelException; import org.msgpack.MessagePack; import org.msgpack.unpacker.Unpacker; import alice.datasegment.Command; import alice.datasegment.CommandType; import alice.datasegment.DataSegment; import alice.datasegment.DataSegmentKey; import alice.datasegment.DataSegmentManager; import alice.datasegment.LocalDataSegmentManager; public class IncomingTcpConnection extends Thread { public Connection connection; public DataSegmentManager manager; public String reverseKey; public IncomingTcpConnection(Connection connection, DataSegmentManager manager, String reverseKey) { this.manager = manager; this.connection = connection; this.reverseKey = reverseKey; } public void run() { MessagePack msgpack = new MessagePack(); Unpacker unpacker = null; try { unpacker = msgpack.createUnpacker(connection.socket.getInputStream()); } catch (IOException e2) { e2.printStackTrace(); } while (true) { try { CommandMessage msg = unpacker.read(CommandMessage.class); CommandType type = CommandType.getCommandTypeFromId(msg.type); LocalDataSegmentManager lmanager = (LocalDataSegmentManager)DataSegment.get("local"); DataSegmentKey dsKey = lmanager.getDataSegmentKey(msg.key); switch (type) { case UPDATE: dsKey.addCommand(new Command(type, null, null, msg.val, 0, 0, null, null, reverseKey)); break; case PUT: dsKey.addCommand(new Command(type, null, null, msg.val, 0, 0, null, null, reverseKey)); break; case PEEK: //Command(CommandType cmdType, String argKey, Value val, int index, int seq, BlockingQueue<Command> replyQueue, CodeSegment cs) { dsKey.addCommand(new Command(type, null, null, null, msg.index, msg.seq, connection.sendQueue, null, null)); break; case TAKE: dsKey.addCommand(new Command(type, null, null, null, msg.index, msg.seq, connection.sendQueue, null, null)); break; case REMOVE: dsKey.addCommand(new Command(type, null, null, null, 0, 0, null, null, null)); break; case REPLY: try { manager.replyQueue.put(new Command(type, null, null, msg.val, msg.index, msg.seq, null, null, null)); } catch (InterruptedException e) { e.printStackTrace(); } break; default: break; } } catch (ClosedChannelException e) { try { connection.sendQueue.put(new Command(CommandType.CLOSE, null, null, null, 0, 0, null, null, null)); } catch (InterruptedException e1) { e1.printStackTrace(); } return; } catch (EOFException e) { try { connection.sendQueue.put(new Command(CommandType.CLOSE, null, null, null, 0, 0, null, null, null)); } catch (InterruptedException e1) { e1.printStackTrace(); } return; } catch (IOException e) { e.printStackTrace(); } } } }