package alice.codesegment;

import java.io.IOException;


import org.msgpack.type.Value;
import org.msgpack.type.ValueFactory;

import alice.datasegment.Command;
import alice.datasegment.CommandType;
import alice.datasegment.DataSegment;
import alice.datasegment.DataSegmentKey;
import alice.datasegment.Receiver;
import alice.test.codesegment.local.bitonicsort.DataList;

public class OutputDataSegment {
	
	/**
	 * for local
	 */
	public <T> void put(String key, T val,Boolean covertFlag) {
		if (covertFlag){ 
			put(key, val);
		} else {
			DataSegment.getLocal().put(key, val);
		}
	}
	
	public <T> void update(String key, T val,Boolean covertFlag) {
		if (covertFlag){ 
			update(key, val);
		} else {
			DataSegment.getLocal().update(key, val);
		}
	}
	
	public void flip(Receiver receiver) {
		//DataSegment.getLocal().put(receiver.key, receiver.val,receiver.obj);
		DataSegmentKey key = DataSegment.getLocal().getDataSegmentKey(receiver.key);
		key.runCommand(new Command(CommandType.PUT,null,null, receiver.val,receiver.obj, 0,0,null,null, "local"));
	}
	
	public void put(String key, Value val) {
		DataSegment.getLocal().put(key, val);
	}
	
	public void put(String key, String val) {
		DataSegment.getLocal().put(key, ValueFactory.createRawValue(val));
	}
	
	public void put(String key, byte[] val) {
		DataSegment.getLocal().put(key, ValueFactory.createRawValue(val, true));
	}
	
	public void put(String key, int val) {
		DataSegment.getLocal().put(key, ValueFactory.createIntegerValue(val));
	}
	
	public <T> void put(String key, T val) {
		try {
			DataSegment.getLocal().put(key, SingletonMessage.getInstance().unconvert(val));
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public void put(Receiver receiver, String key, Object obj, boolean flag) {
		DataSegment.getLocal().put(receiver, key, obj);
	}
	
	public void update(Receiver receiver, String key, Object obj, boolean flag) {
		DataSegment.getLocal().update(receiver, key, obj);
	}
	
	
	public void update(String key, Value val) {
		DataSegment.getLocal().update(key, val);
	}
	
	public void update(String key, String val) {
		DataSegment.getLocal().update(key, ValueFactory.createRawValue(val));
	}
	
	public void update(String key, byte[] val) {
		DataSegment.getLocal().update(key, ValueFactory.createRawValue(val, true));
	}
	
	public void update(String key, int val) {
		DataSegment.getLocal().update(key, ValueFactory.createIntegerValue(val));
	}
	
	public <T> void update(String key, T val) {
		try {
			long t = System.currentTimeMillis();
			DataSegment.getLocal().update(key, SingletonMessage.getInstance().unconvert(val));
			System.out.println(System.currentTimeMillis() - t);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	
	/**
	 * for remote
	 */
	
	public void put(String managerKey, String key, Value val) {
		DataSegment.get(managerKey).put(key, val);
	}
	
	public void put(String managerKey, String key, String val) {
		DataSegment.get(managerKey).put(key, ValueFactory.createRawValue(val));
	}
	
	public void put(String managerKey, String key, byte[] val) {
		DataSegment.get(managerKey).put(key, ValueFactory.createRawValue(val, true));
	}
	
	public void put(String managerKey, String key, int val) {
		DataSegment.get(managerKey).put(key, ValueFactory.createIntegerValue(val));
	}
	
	public <T> void put(String managerKey, String key, T val) {
		try {
			DataSegment.get(managerKey).put(key, SingletonMessage.getInstance().unconvert(val));
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public void update(String managerKey, String key, Value val) {
		DataSegment.get(managerKey).update(key, val);
	}
	
	public void update(String managerKey, String key, String val) {
		DataSegment.get(managerKey).update(key, ValueFactory.createRawValue(val));
	}
	
	public void update(String managerKey, String key, byte[] val) {
		DataSegment.get(managerKey).update(key, ValueFactory.createRawValue(val, true));
	}
	
	public void update(String managerKey, String key, int val) {
		DataSegment.get(managerKey).update(key, ValueFactory.createIntegerValue(val));
	}
	
	public <T> void update(String managerKey, String key, T val) {
		try {
			DataSegment.get(managerKey).update(key, SingletonMessage.getInstance().unconvert(val));
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * kill the Alice process after send other messages.
	 * 
	 * @param managerKey
	 */
	
	public void finish(String managerKey) {
		DataSegment.get(managerKey).finish();
	}

	/**
	 * close socket for RemoteDataSegment after send other messages.
	 * 
	 * @param managerKey
	 */
	
	public void close(String managerKey) {
		DataSegment.get(managerKey).close();
	}

	

	

}