module libasync.tcp;
import std.traits : isPointer;
import libasync.types;
import libasync.events;
import std.typecons : Tuple;

/// Wraps a TCP stream between 2 network adapters, using a custom handler to
/// signal related events. Many of these objects can be active concurrently 
/// in a thread if the event loop is running and the handlers do not block.
final class AsyncTCPConnection
{
package:
	EventLoop m_evLoop;

private:
	NetworkAddress m_peer;

nothrow:
	fd_t m_socket;
	bool m_noDelay;
	bool m_inbound;
	debug bool m_dataRemaining;
public:
	this(EventLoop evl)
	in { assert(evl !is null); }
	body { m_evLoop = evl; }

	mixin DefStatus;

	/// Returns false if the connection has gone.
	@property bool isConnected() const {
		return m_socket != fd_t.init;
	}

	/// Returns true if this connection was accepted by an AsyncTCPListener instance.
	@property bool inbound() const {
		return m_inbound;
	}

	/// Disables(true)/enables(false) nagle's algorithm (default:enabled). 
	@property void noDelay(bool b)
	{
		if (m_socket == fd_t.init)
			m_noDelay = b;
		else
			setOption(TCPOption.NODELAY, true);
	}

	/// Changes the default OS configurations for this underlying TCP Socket.
	bool setOption(T)(TCPOption op, in T val) 
	in { assert(isConnected, "No socket to operate on"); }
	body {
		return m_evLoop.setOption(m_socket, op, val);
	}

	/// Returns the OS-specific structure of the internet address 
	/// of the remote network adapter
	@property NetworkAddress peer() const 
	{
		return m_peer;
	}

	/// Returns the OS-specific structure of the internet address 
	/// for the local end of the connection.
	@property NetworkAddress local()
	in {
		assert(isConnected && m_peer != NetworkAddress.init, "Cannot get local address from a non-connected socket");
	}
	body {
		return m_evLoop.localAddr(m_socket, m_peer.ipv6);
	}

	/// Sets the remote address as an OS-specific structure (only usable before connecting).
	@property void peer(NetworkAddress addr)
	in { 
		assert(!isConnected, "Cannot change remote address on a connected socket"); 
		assert(addr != NetworkAddress.init);
	}
	body {
		m_peer = addr;
	}

	/// (Blocking) Resolves the specified host and resets the peer to this address.
	/// Use AsyncDNS for a non-blocking resolver. (only usable before connecting).
	typeof(this) host(string hostname, size_t port)
	in { 
		assert(!isConnected, "Cannot change remote address on a connected socket"); 
	}
	body {
		m_peer = m_evLoop.resolveHost(hostname, cast(ushort) port);
		return this;
	}

	/// Sets the peer to the specified IP address and port. (only usable before connecting).
	typeof(this) ip(string ip, size_t port)
	in { 
		assert(!isConnected, "Cannot change remote address on a connected socket"); 
	}
	body {
		m_peer = m_evLoop.resolveIP(ip, cast(ushort) port);
		return this;
	}

	/// Starts the connection by registering the associated callback handler in the
	/// underlying OS event loop.
	bool run(void delegate(TCPEvent) del) {
		TCPEventHandler handler;
		handler.del = del;
		handler.conn = this;
		return run(handler);
	}
	
	private bool run(TCPEventHandler del)
	in { assert(!isConnected); }
	body {
		m_socket = m_evLoop.run(this, del);
		if (m_socket == 0)
			return false;
		else
			return true;
		
	}

	/// Receive data from the underlying stream. To be used when TCPEvent.READ is received by the
	/// callback handler. IMPORTANT: This must be called until is returns a lower value than the buffer!
	uint recv(ref ubyte[] ub)
	in { assert(isConnected, "No socket to operate on"); }
	body {
		uint cnt = m_evLoop.recv(m_socket, ub);
		debug {
			if (ub.length > cnt)
			m_dataRemaining = false;
		}
		return cnt;
	}

	/// Send data through the underlying stream by moving it into the OS buffer.
	uint send(in ubyte[] ub)
	in { assert(isConnected, "No socket to operate on"); }
	body {
		version(Posix)
			scope(exit)
				if (m_evLoop.status.code == Status.ASYNC)
					this.writeBlocked = true;
		return m_evLoop.send(m_socket, ub);
	}

	/// Removes the connection from the event loop, closing it if necessary, and
	/// cleans up the underlying resources.
	bool kill(bool forced = false)
	in { assert(isConnected); }
	body {
		bool ret = m_evLoop.kill(this, forced);
		scope(exit) m_socket = 0;
		return ret;
	}

package:
	mixin TCPConnectionMixins;

	@property void inbound(bool b) {
		m_inbound = b;
	}

	@property bool noDelay() const
	{
		return m_noDelay;
	}

	@property fd_t socket() const {
		return m_socket;
	}

	@property void socket(fd_t sock) {
		m_socket = sock;
	}

}

/// Accepts connections on a single IP:PORT tuple by sending a new inbound AsyncTCPConnection 
/// object to the handler for every newly completed handshake.
/// 
/// Note: If multiple threads are listening to the same IP:PORT tuple, the connections will
/// be distributed evenly between them. However, this behavior on Windows is not implemented yet.
final class AsyncTCPListener
{
private:
nothrow:
	EventLoop m_evLoop;
	fd_t m_socket;
	NetworkAddress m_local;
	bool m_noDelay;
	bool m_started;

public:

	this(EventLoop evl, fd_t sock = fd_t.init) { m_evLoop = evl; m_socket = sock; }

	mixin DefStatus;

	/// Sets the default value for nagle's algorithm on new connections.
	@property void noDelay(bool b)
	in { assert(!m_started, "Cannot set noDelay on a running object."); }
	body {
		m_noDelay = b;
	}

	/// Returns the local internet address as an OS-specific structure.
	@property NetworkAddress local() const
	{
		return m_local;
	}

	/// Sets the local internet address as an OS-specific structure.
	@property void local(NetworkAddress addr)
	in { assert(!m_started, "Cannot rebind a listening socket"); }
	body {
		m_local = addr;
	}

	/// Sets the local listening interface to the specified hostname/port.
	typeof(this) host(string hostname, size_t port)
	in { assert(!m_started, "Cannot rebind a listening socket"); }
	body {
		m_local = m_evLoop.resolveHost(hostname, cast(ushort) port);
		return this;
	}

	/// Sets the local listening interface to the specified ip/port.
	typeof(this) ip(string ip, size_t port)
	in { assert(!m_started, "Cannot rebind a listening socket"); }
	body {
		m_local = m_evLoop.resolveIP(ip, cast(ushort) port);
		return this;
	}

	/// Starts accepting connections by registering the given handler with the underlying OS event.
	bool run(void delegate(TCPEvent) delegate(AsyncTCPConnection) del) {
		TCPAcceptHandler handler;
		handler.ctxt = this;
		handler.del = del;
		return run(handler);
	}

	private bool run(TCPAcceptHandler del)
	in { 
		assert(m_local != NetworkAddress.init, "Cannot bind without an address. Please run .host() or .ip()");
	}
	body {
		m_socket = m_evLoop.run(this, del);
		if (m_socket == fd_t.init)
			return false;
		else {
			m_started = true;
			return true;
		}
	}

	/// Use to implement distributed servicing of connections
	@property fd_t socket() const {
		return m_socket;
	}

	/// Stops accepting connections and cleans up the underlying OS resources.
	bool kill()
	in { assert(m_socket != 0); }
	body {
		bool ret = m_evLoop.kill(this);
		if (ret)
			m_started = false;
		return ret;
	}

package:
	version(Posix) mixin EvInfoMixins;
	version(Windows) mixin TCPListenerDistMixins;
	@property bool noDelay() const
	{
		return m_noDelay;
	}
}

package struct TCPEventHandler {
	AsyncTCPConnection conn;

	/// Use getContext/setContext to persist the context in each activity. Using AsyncTCPConnection in args 
	/// allows the EventLoop implementation to create and pass a new object, which is necessary for listeners.
	void delegate(TCPEvent) del;

	void opCall(TCPEvent ev){
		if (conn is null) return; //, "Connection was disposed before shutdown could be completed");
		if (!conn.isConnected)
			return;
		debug conn.m_dataRemaining = true;
		del(ev);
		//debug assert(!conn.m_dataRemaining, "You must recv the whole buffer, because TCP events are edge triggered!");
		return;
	}
}

package struct TCPAcceptHandler {
	AsyncTCPListener ctxt;
	void delegate(TCPEvent) delegate(AsyncTCPConnection) del;

	TCPEventHandler opCall(AsyncTCPConnection conn){ // conn is null = error!
		assert(ctxt !is null);

		void delegate(TCPEvent) ev_handler = del(conn);
		TCPEventHandler handler;
		handler.del = ev_handler;
		handler.conn = conn;
		return handler;
	}
}

enum TCPEvent : char {
	ERROR = 0, // The connection will be forcefully closed, this is debugging information
	CONNECT, // indicates write will not block, although recv may or may not have data
	READ, // called once when new bytes are in the buffer
	WRITE, // only called when send returned Status.ASYNC
	CLOSE // The connection is being shutdown
}

enum TCPOption : char {
	NODELAY = 0,		// Don't delay send to coalesce packets
	REUSEADDR = 1,
	CORK,
	LINGER,
	BUFFER_RECV,
	BUFFER_SEND,
	TIMEOUT_RECV,
	TIMEOUT_SEND,
	TIMEOUT_HALFOPEN,
	KEEPALIVE_ENABLE,
	KEEPALIVE_DEFER,	// Start keeplives after this period
	KEEPALIVE_COUNT,	// Number of keepalives before death
	KEEPALIVE_INTERVAL,	// Interval between keepalives
	DEFER_ACCEPT,
	QUICK_ACK,			// Bock/reenable quick ACKs.
	CONGESTION
}