Mercurial > hg > Database > Alice
view src/main/java/alice/datasegment/Command.java @ 452:f68d103498e0 dispose
refactor (InputDataSegment holder class changed)
author | sugi |
---|---|
date | Tue, 28 Oct 2014 17:24:16 +0900 |
parents | 5b14d0b60201 |
children | b004f62b83e5 |
line wrap: on
line source
package alice.datasegment; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.ByteBuffer; import java.util.concurrent.BlockingQueue; import java.util.zip.Deflater; import java.util.zip.DeflaterOutputStream; import org.msgpack.MessagePack; import alice.codesegment.CodeSegment; import alice.codesegment.SingletonMessage; import alice.daemon.CommandMessage; import alice.daemon.Connection; public class Command { public CommandType type; public String key; public Receiver receiver; public Object val; public int index; public int seq; public Connection connection; // for remote public BlockingQueue<Command> replyQueue; public CodeSegment cs; public String reverseKey; private boolean quickFlag = false; private boolean serializeFlag = false; private boolean compressFlag = false; public Command(CommandType cmdType, Receiver receiver, String key, Object val, int index, int seq, BlockingQueue<Command> replyQueue, CodeSegment cs, String reverseKey) { this.type = cmdType; this.receiver = receiver; this.key = key; this.val = val; this.index = index; this.seq = seq; this.replyQueue = replyQueue; this.cs = cs; this.reverseKey = reverseKey; } public Command(CommandType cmdType, Receiver receiver, String key, Object val, int index, int seq, CodeSegment cs, String reverseKey, Connection connection) { this.type = cmdType; this.receiver = receiver; this.key = key; this.val = val; this.index = index; this.seq = seq; this.connection = connection; this.cs = cs; this.reverseKey = reverseKey; } public String getCommandString() { String csName = "null"; if (cs != null) { csName = cs.toString(); } return this.type + "\t" + key + "\t" + val + "\tindex=" + index + "\tcs=" + csName; } /** * @return serialized ByteBuffer */ public ByteBuffer convert() { ByteBuffer buf = null; MessagePack msg = SingletonMessage.getInstance(); try { byte[] header = null; byte[] data = null; byte[] dataSize = null; switch (type) { /* * UPDATE, PUT, REPLY need send DataSegment to RemoteDataSegment * case UPDATE and PUT * compress and serialize flag are selected by user, so if true, need convert. * case REPLY * these flags represent DataSegment status. * for example, serializeFlag is true. DataSegment had already converted, so no need convert. */ case UPDATE: case PUT: if (!serializeFlag) { data = (byte[]) val; } else { long start = System.currentTimeMillis(); data = msg.write(val); long end = System.currentTimeMillis(); System.out.println("convert DataSegment" +(end - start)); } if (compressFlag) { data = zip(data); } header = msg.write(new CommandMessage(type.id, index, seq, key, quickFlag, serializeFlag, compressFlag)); dataSize = msg.write(data.length); buf = ByteBuffer.allocate(header.length+dataSize.length+data.length); buf.put(header); buf.put(dataSize); long start = System.currentTimeMillis(); buf.put(data); long end = System.currentTimeMillis(); System.out.println("put DataSegment" +(end - start)); break; case REPLY: // only serialize if (serializeFlag) { data = (byte[]) val; } else { data = msg.write(val); this.serializeFlag = true; } header = msg.write(new CommandMessage(type.id, index, seq, key, quickFlag, serializeFlag, compressFlag)); dataSize = msg.write(data.length); buf = ByteBuffer.allocate(header.length+dataSize.length+data.length); buf.put(header); buf.put(dataSize); buf.put(data); break; default: header = msg.write(new CommandMessage(type.id, index, seq, key, quickFlag, serializeFlag, compressFlag)); buf = ByteBuffer.allocate(header.length); buf.put(header); break; } buf.flip(); } catch (IOException e) { e.printStackTrace(); } return buf; } public void setQuickFlag(boolean flag){ quickFlag = flag; } public boolean getQuickFlag(){ return quickFlag; } public void setSerializeFlag(boolean flag){ serializeFlag = flag; } public boolean getSerializeFlag(){ return serializeFlag; } public void setCompressFlag(boolean flag){ compressFlag = flag; } public boolean getCompressFlag(){ return compressFlag; } public byte[] zip(byte[] input) throws IOException{ Deflater deflater = new Deflater(); ByteArrayOutputStream os = new ByteArrayOutputStream(); DeflaterOutputStream dos = new DeflaterOutputStream(os, deflater); dos.write(input); dos.finish(); return os.toByteArray(); } }