package org.kne.cloud.network.ntp;

import java.io.Closeable;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.kne.cloud.clock.HighAccuracyClock;
import org.kne.cloud.clock.NTPTimestamps;
import org.kne.cloud.network.MultiProtocolSocketAddress;
import org.kne.cloud.network.ThreadTool;
import org.kne.io.KNEChannels;
import org.kne.math.Long128;

public class NTPv4Protocol implements Closeable, AutoCloseable {
	private static final boolean showPacket = false;
	public static final int NTP_DEFAULT_PORT = 123;
	private static final long DEFAULT_POLL_INTERVAL = 1000000000L;
	private static final long REQUEST_TIMEOUT = 10000000000L;
	private static final long EPHEMERAL_TIMEOUT = 60000000000L;

	public static class NTPPeer {
		private MultiProtocolSocketAddress address;
		private boolean isEphemeral;// ephemeral
		private volatile long ephemeralUpdateTime = System.nanoTime();
		private volatile long pollInterval = DEFAULT_POLL_INTERVAL;
		private volatile long pollUpdateTime = System.nanoTime() - pollInterval;
		private volatile int requestMode;
		private volatile long peerPollInterval = pollInterval;

		private boolean checkPollInterval() {
			long curr = System.nanoTime();
			if (curr - pollUpdateTime > Math.min(peerPollInterval, pollInterval)) {
				pollUpdateTime = curr;
				return true;
			} else {
				return false;
			}
		}

		private boolean checkEphemeralTimeout() {
			if (!isEphemeral) {
				return false;
			}
			long curr = System.nanoTime();
			return curr - ephemeralUpdateTime > EPHEMERAL_TIMEOUT;
		}

		private void resetEphemeralTimeout() {
			ephemeralUpdateTime = System.nanoTime();
		}

		public long getPollInterval() {
			return pollInterval;
		}

		public void setPollInterval(long pollInterval) {
			this.pollInterval = pollInterval;
		}

		public NTPPeer(MultiProtocolSocketAddress address, int requestMode) {
			this(address, requestMode, false);
		}

		protected NTPPeer(MultiProtocolSocketAddress address, int requestMode, boolean isEphemeral) {
			super();
			checkRequestMode(requestMode);
			this.address = address;
			this.isEphemeral = isEphemeral;
			this.requestMode = requestMode;
		}

		private void checkRequestMode(int requestMode2) {
			if (requestMode2 != NTPv4Packet.NTP_CLIENT && requestMode2 != NTPv4Packet.NTP_SYMMETRIC_ACTIVE) {
				throw new IllegalArgumentException(
						"request mode must be NTPv4Packet.NTP_CLIEN or NTPv4Packet.NTP_SYMMETRIC_ACTIVE");
			}
		}

		public NTPPeer(String hostport, int requestMode) {
			this(new MultiProtocolSocketAddress(hostport), requestMode);
		}

		public MultiProtocolSocketAddress getAddress() {
			return address;
		}

		private boolean isEphemeral() {
			return isEphemeral;
		}

		@Override
		public String toString() {
			return "NTPPeer [address=" + address + ", isEphemeral=" + isEphemeral + ", ephemeralUpdateTime="
					+ ephemeralUpdateTime + "]";
		}

		@Override
		public int hashCode() {
			return Objects.hash(address);
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			NTPPeer other = (NTPPeer) obj;
			return Objects.equals(address, other.address);
		}

		private void setPeerPollInterval(long peerPollInterval) {
			this.peerPollInterval = peerPollInterval;
		}

	}

	private static class CheckListItem {
		private NTPPeer peer;
		private Long128 timeStamp;
		private long createTime = System.nanoTime();

		private CheckListItem(Long128 timeStamp, NTPPeer peer) {
			this.timeStamp = timeStamp;
			this.peer = peer;
		}

		private boolean checkTimeout() {
			return System.nanoTime() - createTime > REQUEST_TIMEOUT;
		}
	}

	private Set<NTPPeer> peers = Collections.synchronizedSet(new HashSet<>());
	private Set<NTPPeer> ephemeralPeers = Collections.synchronizedSet(new HashSet<>());

	private ReentrantLock checkLock = new ReentrantLock();
	private List<CheckListItem> checkList = new ArrayList<>();

	private NTPContext context;
	private DatagramSocket dgs;
	private MultiProtocolSocketAddress bind;

	public NTPv4Protocol(NTPContext context) throws UnknownHostException, IOException {
		this(context, new MultiProtocolSocketAddress("UDP", "::0", NTP_DEFAULT_PORT));
	}

	public NTPv4Protocol(NTPContext context, MultiProtocolSocketAddress bind) throws UnknownHostException, IOException {
		this.bind = bind;
		this.context = context;
		dgs = bind.listenDatagramSocket();
		context.registerIO(this);
		Thread t4= ThreadTool.makeVThread("NTPv4 Packet Cleaner",()->{
			while(isClosed()){
			removeTimeoutTimestamp();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
					e.printStackTrace();
                }
            }
		});
		t4.start();
		Thread tr = new Thread(recv);
		tr.setName("NTPv4 Receive Thread");
		tr.start();
	}

	private void setNTPInformation(NTPv4Packet nv4, NTPPeer peer) {
		if (peer != null) {
			long interval = peer.getPollInterval();
			int pi = (int) log(interval / 1000000000L, 2);
			nv4.setPollInterval(pi);
		} else {
			long interval = DEFAULT_POLL_INTERVAL;
			int pi = (int) log(interval / 1000000000L, 2);
			nv4.setPollInterval(pi);
		}
		nv4.setVersionNumber(NTPv4Packet.NTP_VERSION_NUMBER);
		nv4.setLeapIndicator(context.getLeapIndicator());
		int stratum = context.getStratum();
		if (stratum == 16) {
			stratum = 0;
		}
		nv4.setStratum(stratum);
		Long128 precision = context.getLocalPrecision();
		byte b = (byte) Math.ceil(log(precision.doubleValue() / (Math.pow(2, 64)), 2));
		nv4.setPrecision(b);
		nv4.setRootDelay((int) context.getRootDelay());
		nv4.setRootDispersion((int) context.getCurrentRootDispersion());
		nv4.setReferenceTimestamp64(context.getReferenceTimestamp64());
		nv4.setTransmitTimestamp64(context.getClock().getCurrentTimeNTP64());
	}

	private void setNTPInformation(NTPv4Packet nv4, NTPv4Packet request, NTPPeer peer) {
		nv4.setOriginateTimestamp(request.getTransmitTimestamp());
		nv4.setReceiveTimestamp64(NTPTimestamps.ntp128To64(request.getDestinationTimestamp128()));
		setNTPInformation(nv4, peer);
	}

	private static double log(double value, double base) {

		return Math.log(value) / Math.log(base);

	}

	public void request(NTPPeer peer) throws IOException {
		NTPv4Packet nv4 = new NTPv4Packet(context.getClock());
		nv4.setMode(peer.requestMode);
		setNTPInformation(nv4, peer);
		putOriginTimestamp(nv4, peer);
		peer.address.getInetAddress();
		sendNTPPacket(nv4, peer.address.getSocketAddress());
	}

	private void putOriginTimestamp(NTPv4Packet nv4, NTPPeer peer) {
		checkLock.lock();
		try {
			checkList.add(new CheckListItem(nv4.getTransmitTimestamp128(), peer));
		} finally {
			checkLock.unlock();
		}
	}

	private void removeTimeoutTimestamp() {
		checkLock.lock();
		try {
			for (Iterator<CheckListItem> iterator = checkList.iterator(); iterator.hasNext(); ) {
				CheckListItem bigInteger = (CheckListItem) iterator.next();
				if (bigInteger.checkTimeout()) {
					if (showPacket)
						System.out.println("timeout:" + bigInteger.timeStamp);
					iterator.remove();
				}
			}
		}finally{
			checkLock.unlock();
		}
	}

	private void sendNTPPacket(NTPv4Packet nv4, SocketAddress target) throws IOException {
		ByteBuffer bbf = ByteBuffer.allocate(65535);
		nv4.writeToChannel(KNEChannels.newWritableChannel(bbf));
		bbf.flip();
		DatagramPacket dp = new DatagramPacket(bbf.array(), bbf.limit());
		dp.setSocketAddress(target);
		dgs.send(dp);
		if (showPacket)
			System.out.println("NTPv4 TX:" + target + " " + nv4);
	}

	private NTPPeer checkOriginTimestamp(NTPv4Packet nv4) {
		checkLock.lock();
		try {
			Long128 ori = nv4.getOriginateTimestamp128();
			removeTimeoutTimestamp();
			for (Iterator<CheckListItem> iterator = checkList.iterator(); iterator.hasNext();) {
				CheckListItem item = (CheckListItem) iterator.next();
				if (item.timeStamp.equals(ori)) {
					iterator.remove();
					return item.peer;
				}
			}
		} finally {
			checkLock.unlock();
		}
		return null;
	}

	NTPPeer findPeer(MultiProtocolSocketAddress addr) {
		Object[] objs = peers.toArray();
		for (int i = 0; i < objs.length; i++) {
			NTPPeer np = (NTPPeer) objs[i];
			if (np.getAddress().equals2(addr)) {
				return np;
			}
		}
		objs = ephemeralPeers.toArray();
		for (int i = 0; i < objs.length; i++) {
			NTPPeer np = (NTPPeer) objs[i];
			if (np.getAddress().equals2(addr)) {
				return np;
			}
		}
		return null;
	}

	private Runnable recv = new Runnable() {
		@Override
		public void run() {
			try {
				while (!dgs.isClosed()) {
					AtomicReference<InetSocketAddress> isa = new AtomicReference<>();
					NTPv4Packet nv4 = receiveNTPPacket(isa);
					MultiProtocolSocketAddress mpsafrom = new MultiProtocolSocketAddress(bind.getProtocol(), isa.get());
					switch (nv4.getMode()) {
					case NTPv4Packet.NTP_SYMMETRIC_ACTIVE:
						NTPv4Packet nv4r = new NTPv4Packet(context.getClock());
						nv4r.setMode(NTPv4Packet.NTP_SYMMETRIC_PASSIVE);
						setNTPInformation(nv4r, nv4, findPeer(mpsafrom));
						sendNTPPacket(nv4r, isa.get());
						ephemeralPeers.add(new NTPPeer(mpsafrom, NTPv4Packet.NTP_SYMMETRIC_ACTIVE, true));
						break;
					case NTPv4Packet.NTP_SYMMETRIC_PASSIVE:
						NTPPeer npr = checkOriginTimestamp(nv4);
						if (npr != null) {
							npr.resetEphemeralTimeout();
							npr.setPeerPollInterval((1L << nv4.getPollInterval()) * 1000000000);
							context.putPacket(nv4, mpsafrom);
						} else {
						}
						break;
					case NTPv4Packet.NTP_CLIENT:
						NTPv4Packet nv4s = new NTPv4Packet(context.getClock());
						nv4s.setMode(NTPv4Packet.NTP_SERVER);
						setNTPInformation(nv4s, nv4, findPeer(mpsafrom));
						sendNTPPacket(nv4s, isa.get());
						break;
					case NTPv4Packet.NTP_SERVER:
						NTPPeer nprx = checkOriginTimestamp(nv4);
						if (nprx != null) {
							nprx.resetEphemeralTimeout();
							nprx.setPeerPollInterval((1L << nv4.getPollInterval()) * 1000000000);
							context.putPacket(nv4, mpsafrom);
						} else {
						}
						break;
					}

				}
			} catch (IOException e) {
				e.printStackTrace();
			}

		}

	};

	protected void sendReqPackets() throws IOException {
		List<NTPPeer> sends = getPeersWillSend();
		for (NTPPeer npr : sends) {
			request(npr);
		}

	}

	protected List<NTPPeer> getPeersWillSend() {
		List<NTPPeer> sends = new ArrayList<>();
		peers.forEach((pr) -> {
			if (pr.checkPollInterval()) {
				sends.add(pr);
			}
		});
		ephemeralPeers.removeIf((pr) -> {
			return pr.checkEphemeralTimeout();
		});
		ephemeralPeers.forEach((pr) -> {
			if (pr.checkPollInterval()) {
				if (!peers.contains(pr)) {
					sends.add(pr);
				}
			}
		});
		return sends;
	}

	private NTPv4Packet receiveNTPPacket(AtomicReference<InetSocketAddress> isa) throws IOException {
		ByteBuffer bbf = ByteBuffer.allocate(65535);
		DatagramPacket dp = new DatagramPacket(bbf.array(), bbf.limit());
		dgs.receive(dp);
		isa.set((InetSocketAddress) dp.getSocketAddress());
		NTPv4Packet nv4 = new NTPv4Packet(context.getClock());
		nv4.readFromChannel(KNEChannels.newReadableChannel(bbf), dp.getLength());
		if (showPacket)
			System.out.println("NTPv4 RX:" + isa.get() + " " + nv4);
		return nv4;
	}

	public Set<NTPPeer> getPeers() {
		return peers;
	}

	@Override
	public void close() throws IOException {
		dgs.close();
		context.unregisterIO(this);
	}

	public boolean isClosed() {
		return dgs.isClosed();
	}

	public static void main(String[] args) throws IOException, InterruptedException {
		HighAccuracyClock hac = new HighAccuracyClock();
		NTPContext context = new NTPContext(hac);
		System.out.println(context);
		NTPv4Protocol nvc = new NTPv4Protocol(context, new MultiProtocolSocketAddress("{UDP}0.0.0.0:123"));// 106.55.184.199
		nvc.getPeers().add(new NTPPeer("{UDP}106.55.184.199:123", NTPv4Packet.NTP_CLIENT));
		nvc.getPeers().add(new NTPPeer("{UDP}time.windows.com:123", NTPv4Packet.NTP_CLIENT));
		nvc.getPeers().add(new NTPPeer("{UDP}127.0.0.1:123", NTPv4Packet.NTP_SYMMETRIC_ACTIVE));
		/*
		 * for (int i = 0; i < 10000; i++) { Thread.sleep(1000);
		 * System.out.println(context.getCurrentSelfDispersion128());
		 * System.out.println(context); }
		 */
	}
}