public abstract class Actor {
	
	private Node node;
	
	private String name;
	
	private final BlockingQueue<Message> mailbox = new LinkedBlockingQueue<>();

	private Thread actorThread;
	
	public Node getNode() {
		return node;
	}
	
	public void setNode(Node node) {
		this.node = node;
	}

	public void setName(String name) {
		this.name = name;
	}
	
	public String getName() {
		return name;
	}

    public void start() {
        actorThread = new Thread(() -> {
        	ActorSystem.setThreadLocalActor(this);
            for(;;) {
                try {
                    Message message = mailbox.take();
                    try {
                    	handleMessage(message);
                    } catch (Exception e) {
                    	e.printStackTrace();
                    }
                } catch (InterruptedException ignore) {
                    // ignore
                }
            }
        });

        actorThread.start();
    }

    public void act(Message msg) {
        mailbox.offer(msg);
    }
    
    protected abstract void handleMessage(Message message);
}

public abstract class Node {
	
	/**
	 * 名字
	 * 需要是唯一的，按名字查找
	 */
	private String name;
	
	private InetSocketAddress address;
	
	public String getName() {
		return name;
	}

	public void setName(String nodeName) {
		name = nodeName;
	}

	public void setAddress(InetSocketAddress address) {
		this.address = address;
	}
}

public class ActorSystem {
	
	private static Map<String, InetSocketAddress> clusterConfig;
	
	/**
	 * 當(dāng)前綁定到的節(jié)點(diǎn)
	 */
	private static Node currNode;
	
	private final static Map<String, Actor> actors = new HashMap<>();
	
	/**
	 * 維護(hù)線程與Actor的對應(yīng)關(guān)系
	 */
	private final static ThreadLocal<Actor> currThreadActor = new ThreadLocal<>();
	
	/**
	 * 客戶端Netty bootstrap
	 */
	private static Bootstrap clientBootstrap;
	
	/**
	 * 維護(hù)節(jié)點(diǎn)與通道的對應(yīng)關(guān)系
	 */
	private final static Map<String, Channel> channels = new ConcurrentHashMap<>();
	
	private static void startNettyBootstrap() {
        try {
        	// 先啟動(dòng)服務(wù)端bootstrap
    		EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)))
                     .addLast(new ObjectEncoder())
                     .addLast(new ServerHandler());
                 }
             });
            InetSocketAddress address = clusterConfig.get(currNode.getName());
            b.bind(address).sync();
            
            // 再啟動(dòng)客戶端bootstrap
            EventLoopGroup group = new NioEventLoopGroup();
            clientBootstrap = new Bootstrap();
            clientBootstrap.group(group)
             .channel(NioSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .option(ChannelOption.TCP_NODELAY, true)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)))
                     .addLast(new ObjectEncoder())
                     .addLast(new ClientHandler());
                 }
             });
        } catch (Exception e) {
        	throw new RuntimeException("actor system start fail", e);
        }
	}
	
	public static void start() {
		// 啟動(dòng)定時(shí)器
		Timer.start();
		// 啟動(dòng)Netty bootstrap
		startNettyBootstrap();
	}
	
	public static void conf(Map<String, InetSocketAddress> config) {
		clusterConfig = config;
	}

	/**
	 * 將當(dāng)前系統(tǒng)綁定到某個(gè)節(jié)點(diǎn)
	 */
	public static void bindNode(Class<? extends Node> nodeClass, String nodeName) {
		InetSocketAddress address = clusterConfig.get(nodeName);
		try {
			Constructor<? extends Node> constructor =  nodeClass.getDeclaredConstructor();
			Node node = constructor.newInstance();
			node.setName(nodeName);
			currNode = node;
		} catch (Exception e) {
			throw new RuntimeException("create node fail", e);
		}
	}

	/**
	 * 啟動(dòng)新的Actor
	 */
	public static void newActor(Class<? extends Actor> actorClass, String name) {
		try {
			Constructor<? extends Actor> constructor =  actorClass.getDeclaredConstructor();
			Actor actor = constructor.newInstance();
			actor.setName(name);
			actor.setNode(currNode);
			actor.start();
			actors.put(name, actor);
		} catch (Exception e) {
			throw new RuntimeException("create actor fail", e);
		}
	}
}

	public static void send(Message msg) {
		String destNodeName = msg.getDestNode();
		String destActorName = msg.getDestActor();
		if (destNodeName.equals(currNode.getName())) {
			Actor destActor = actors.get(destActorName);
			destActor.act(msg);
		} else {
	        sendToAnotherNode(msg);
		}
	}
	
	private static void sendToAnotherNode(Message msg) {
		try {
			String destNodeName = msg.getDestNode();
        	// 如果沒有連接，那么先建立連接
			Channel channel = getChannel(destNodeName);
        	if (!isChannelValid(channel)) {
        		InetSocketAddress address = clusterConfig.get(destNodeName);
        		// TODO 有可能出現(xiàn)多線程同時(shí)嘗試建立連接的情況，這里會保留最后一個(gè)
        		// 優(yōu)化方法有兩種：
        		// 1. 允許多次嘗試，當(dāng)建立連接成功后，如果已有成功連接的引用，那么不保留這次創(chuàng)建的連接
        		// 2. 嘗試時(shí)阻塞其他嘗試
        		clientBootstrap.connect(address).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                    	setChannel(destNodeName, future.channel());
                    	future.channel().writeAndFlush(msg);
                    }
        		});
        	} else {
        		// 否則直接發(fā)送消息
        		channel.writeAndFlush(msg);
        	}
        } catch (Exception e) {
        	throw new RuntimeException("send to another node fail");
        }
	}
	
	public static void send(String destNodeName, String destActorName, String command, Object... params) {
		Actor srcActor = currThreadActor.get();
		String srcActorName = srcActor == null ? null : srcActor.getName();
		String srcNodeName = srcActor == null ? null : srcActor.getNode().getName();
		Message msg = new Message(command, srcNodeName, srcActorName, destNodeName, destActorName, params);
		send(msg);
	}
	
	public static boolean isChannelValid(Channel channel) {
		return channel != null && channel.isActive() && channel.isWritable();
	}
	
	public static Channel getChannel(String destNodeName) {
		return channels.get(destNodeName);
	}
	
	public static void setChannel(String destNodeName, Channel channel) {
		channels.put(destNodeName, channel);
	}

	/**
	 * Actor發(fā)送給自己
	 */
	public static void sendSelf(String command, Object... params) {
		Actor selfActor = currThreadActor.get();
		if (selfActor == null) {
			throw new RuntimeException("not in an actor, send fail");
		}
		send(selfActor.getNode().getName(), selfActor.getName(), command, params);
	}
	
	public static void setThreadLocalActor(Actor actor) {
		currThreadActor.set(actor);
	}

	public static void sleep(long millis, String command, Object... params) {
		String destActorName = currThreadActor.get().getName();
		Timer.addTimeTask(new TimerTask(System.currentTimeMillis() + millis, () -> {
			ActorSystem.send(currNode.getName(), destActorName, command, params);
		}));
	}

public class Timer {
	
	/**
	 * 基于優(yōu)先級隊(duì)列實(shí)現(xiàn)的定時(shí)任務(wù)隊(duì)列
	 */
	private static final PriorityQueue<TimerTask> timerTasks = new PriorityQueue<>();
	
	private static final ReentrantLock lock = new ReentrantLock();
	
	/**
	 * 喚醒阻塞條件一：隊(duì)列非空
	 */
	private static final Condition notEmpty = lock.newCondition();
	
	/**
	 * 喚醒阻塞條件二：當(dāng)前時(shí)刻有任務(wù)需要執(zhí)行
	 */
	private static final Condition hasCurrTask = lock.newCondition();
	
	/**
	 * 添加新的定時(shí)任務(wù)
	 */
	public static void addTimeTask(TimerTask task) {
		lock.lock();
		if (timerTasks.isEmpty()) {
			notEmpty.signal();
		}
		TimerTask firstTask = timerTasks.peek();
		timerTasks.offer(task);
		if (firstTask != null && task.getExecTime() < firstTask.getExecTime()) {
			hasCurrTask.signal();
		}
		lock.unlock();
	}
	
	/**
	 * 啟動(dòng)定時(shí)器
	 */
	public static void start() {
		Executor executor = Executors.newSingleThreadExecutor();
		executor.execute(() -> {
			while (true) {
				TimerTask firstTask;
				lock.lock();
				if (timerTasks.isEmpty()) {
					try {
						notEmpty.await();
					} catch (InterruptedException ignore) {
						// ignore
					}
				}
	    	   	firstTask = timerTasks.peek();
	    	   	long currDeadlineMillis = firstTask.getExecTime();
	    	   	long currTime = System.currentTimeMillis();
	    	   	long delay = currDeadlineMillis - currTime;
	    	   	if (delay > 0) {
	    	   		try {
						hasCurrTask.await(delay, TimeUnit.MILLISECONDS);
					} catch (InterruptedException ignore) {
						// ignore
					}
	    	   	} else {
	    	   		firstTask = timerTasks.poll();
	    	   	}
	    	   	lock.unlock();
	    	   	if (firstTask != null) {
	    	   		firstTask.run();
	    	   	}
			}
		});
	}

}