use errors;
use ev;
use ev::server;
use fmt;
use io;
use log;
use memio;
use net::http;
use net::ip;
use net::tcp;
use net;
use os;
use unix::signal;

type state = struct {
	loop: *ev::loop,
	exit: int,
};

export fn main() void = {
	const loop = ev::newloop()!;
	defer ev::finish(&loop);

	const sock = match (ev::listen_tcp(&loop, ip::LOCAL_V4, 8080, tcp::reuseaddr)) {
	case let err: net::error =>
		log::fatalf("Error: listen: {}", net::strerror(err));
	case let err: errors::error =>
		log::fatalf("Error: listen: {}", errors::strerror(err));
	case let sock: *ev::file =>
		yield sock;
	};

	defer ev::close(sock);

	defer ev::close(sock)!;

	let state = state {
		loop = &loop,
		...
	};

	const server = server::http_serve(sock, &http_serve, &state);
	defer server::server_finish(server);

	const server = server::http_serve(sock, &http_serve, &state)!;
	defer server::server_finish(server)!;

	const sig = ev::signal(&loop, &signal, signal::sig::INT, signal::sig::TERM)!;

	defer ev::close(sig);

	defer ev::close(sig)!;

	ev::setuser(sig, &state);

	log::println("Listening on 127.0.0.1:8080");
	for (ev::dispatch(&loop, -1)!) void;
	os::exit(state.exit);
};

fn signal(file: *ev::file, sig: signal::sig) void = {
	log::printfln("Exiting due to {}", signal::signame(sig));
	const state = ev::getuser(file): *state;
	ev::stop(state.loop);
};

fn http_serve(
	user: nullable *opaque,
	req: http::request,
	rw: http::response_writer
) void = {
	const state = user: *state;

	const (ip, port) = http::peeraddr(&rw);
	log::printfln("Serving from {}:{}", ip::string(ip), port);

	const buf = memio::dynamic();
	defer io::close(&buf)!;

	fmt::fprintfln(&buf, "Method: {}", req.method)!;
	fmt::fprintfln(&buf, "Path: {}", req.target.path)!;
	fmt::fprintfln(&buf, "Fragment: {}", req.target.fragment)!;
	fmt::fprintfln(&buf, "Query: {}", req.target.query)!;
	fmt::fprintfln(&buf, "Headers:")!;
	http::write_header(&buf, &req.header)!;
	fmt::fprintln(&buf)!;
	io::copy(&buf, req.body)!;

	io::seek(&buf, 0, io::whence::SET)!;

	http::response_add_header(&rw, "Content-Type", "text/plain");

	http::response_add_header(&rw, "Content-Type", "text/plain")!;

	http::response_set_body(&rw, &buf);
	http::response_write(&rw)!;
};

use errors;
use io;
use net;
use net::ip;
use rt;
use unix::signal;

export type op = enum u64 {
	NONE     = 0,
	READV    = 1 << 0,
	WRITEV   = 1 << 1,

	READABLE     = 1 << 16,
	WRITABLE     = 2 << 16,
	ACCEPT       = 3 << 16,
	CONNECT_TCP  = 4 << 16,
	CONNECT_UNIX = 5 << 16,
	SIGNAL       = 6 << 16,
	TIMER        = 7 << 16,
	SENDTO       = 8 << 16,
	RECVFROM     = 9 << 16,
	SEND         = 10 << 16,
	RECV         = 11 << 16,
};

export type fflags = enum uint {
	NONE = 0,
	BLOCKING = 1 << 31,
};

export type file = struct {
	fd: io::file,
	ev: *loop,

	flags: fflags,
	op: op,
	cb: nullable *opaque,
	cb2: nullable *opaque,
	user: nullable *opaque,

	prio: int,

	// Operation-specific data
	union {
		struct {
			rvbuf: io::vector,
			rvec: []io::vector,
			wvbuf: io::vector,
			wvec: []io::vector,
		},
		sockflag: net::sockflag,
		sigmask: signal::sigset,
		sendrecv: struct {
			sbuf: []u8,
			rbuf: []u8,
			dest: ip::addr,
			port: u16,
		},
	},
};

// Registers a file descriptor with an event loop.
export fn register(
	loop: *loop,
	fd: io::file,
	user: nullable *opaque = null,

) (*file | errors::error) = {

) (*file | errors::error | nomem) = {

	const file = alloc(file {
		flags = fflags::NONE,
		fd = fd,
		ev = loop,
		op = op::NONE,
		user = user,
		...

	});

	})?;

	let ev = rt::epoll_event {
		events = 0,
		data = rt::epoll_data {
			fd = 0,
		}
	};
	ev.data.ptr = file;
	match (rt::epoll_ctl(loop.fd, rt::EPOLL_CTL_ADD, fd, &ev)) {
	case void =>
		yield;
	case let err: rt::errno =>
		if (err == rt::EPERM) {
			// epoll(2) does not support regular files, use blocking
			// I/O instead
			file.flags = fflags::BLOCKING;
			return file;
		};
		return errors::errno(err);
	};

	return file;
};

// Unregisters a file object with an event loop and frees resources associated
// with it. Does not close the underlying file descriptor.

export fn unregister(file: *file) void = {

export fn unregister(file: *file) (void | nomem) = {

	const loop = file.ev;
	if (file.flags & fflags::BLOCKING == 0) {
		// The only way that this could fail is in the event of a
		// use-after-free or if the user fucks around and constructs a
		// custom [[file]] which was never registered, so assert on
		// error.
		rt::epoll_ctl(loop.fd, rt::EPOLL_CTL_DEL, file.fd, null)!;
	};
	if (file.op == op::SIGNAL) {

		signal_restore(file);

		signal_restore(file)?;

	};

	for (let ev &.. loop.events) {
		if (file != ev.data.ptr) {
			continue;
		};
		ev.events = 0;
		ev.data.ptr = null: *opaque;
		ev.data.fd = 0;
		break;
	};
	free(file);
};

// Unregisters a file object with an event loop, frees resources associated with
// it, and closes the underlying file descriptor.

export fn close(file: *file) void = {

export fn close(file: *file) (void | nomem) = {

	const fd = file.fd;

	unregister(file);

	unregister(file)?;

	io::close(fd)!;
};

// Sets the user data field on this file object to the provided object.
export fn setuser(file: *file, user: nullable *opaque) void = {
	file.user = user;
};

// Returns the user data field from this file object. If the field was null, an
// assertion is raised.
export fn getuser(file: *file) *opaque = {
	return file.user as *opaque;
};

// Returns the file descriptor for a given file. Note that ev assumes that it
// will be responsible for all I/O on the file and any user modifications may
// cause the event loop to enter an invalid state.
export fn getfd(file: *file) io::file = {
	return file.fd;
};

// Returns the event loop for a given file.
export fn getloop(file: *file) *loop = {
	return file.ev;
};

// Sets the priority on this file object. Used to priorize concurrent events,
// starting on the next dispatch. The lowest run first.
export fn setprio(file: *file, prio: int) void = {
	file.prio = prio;
};

fn sort_events(a: const *opaque, b: const *opaque) int = {
	const a = a: *rt::epoll_event;
	const b = b: *rt::epoll_event;

	if (a.data.fd == 0 || b.data.fd == 0) {
		return 0;
	};

	const a = a.data.ptr: *file;
	const b = b.data.ptr: *file;

	return a.prio - b.prio;
};

// Updates epoll events for a given file. For internal use.
fn file_epoll_ctl(file: *file) void = {
	let events = rt::EPOLLONESHOT;
	if (file.op & op::READV != 0 || file.op == op::READABLE) {
		events |= rt::EPOLLIN | rt::EPOLLHUP;
	};
	if (file.op & op::WRITEV != 0 || file.op == op::WRITABLE) {
		events |= rt::EPOLLOUT | rt::EPOLLHUP;
	};
	switch (file.op) {
	case op::ACCEPT =>
		events |= rt::EPOLLIN;
	case op::CONNECT_TCP, op::CONNECT_UNIX =>
		events |= rt::EPOLLOUT;
	case op::SIGNAL =>
		events |= rt::EPOLLIN;
	case op::TIMER =>
		events &= ~rt::EPOLLONESHOT;
		events |= rt::EPOLLIN;
	case op::SEND, op::SENDTO =>
		events |= rt::EPOLLOUT;
	case op::RECV, op::RECVFROM =>
		events |= rt::EPOLLIN;
	case =>
		yield;
	};

	let ev = rt::epoll_event {
		events = events,
		data = rt::epoll_data {
			fd = 0,
		},
	};
	ev.data.ptr = file;
	// This can only fail under conditions associated with EPOLLEXCLUSIVE,
	// which we do not support.
	rt::epoll_ctl(file.ev.fd, rt::EPOLL_CTL_MOD, file.fd, &ev)!;
};

use io;
use rt;

// A callback for a [[read]] or [[readv]] operation.

export type readcb = fn(file: *file, result: (size | io::EOF | io::error)) void;

export type readcb = fn(file: *file, result: (size | io::EOF | io::error)) (void | nomem);

// Schedules a read operation on a file object. The provided buffer must be
// valid for the duration of the read operation.
export fn read(
	file: *file,
	cb: *readcb,
	buf: []u8,
) req = {
	file.rvbuf = io::mkvector(buf);
	// XXX: Bit of a hack to avoid allocating a slice
	const vec = (&file.rvbuf: *[*]io::vector)[..1];
	return readv(file, cb, vec...);
};

// Schedules a vectored read operation on a file object. The provided vectors
// must be valid for the duration of the read operation.
export fn readv(
	file: *file,
	cb: *readcb,
	vec: io::vector...
) req = {
	assert(file.op & op::READV == 0);
	if (file.flags & fflags::BLOCKING != 0) {
		const r = io::readv(file.fd, vec...);

		cb(file, r);

		cb(file, r)!;

		return req { ... };
	};

	file.op |= op::READV;
	file.cb = cb;
	file.rvec = vec;
	file_epoll_ctl(file);
	return mkreq(&readv_cancel, file);
};

fn readv_ready(file: *file, ev: *rt::epoll_event) void = {
	assert(file.op & op::READV != 0);
	assert(file.cb != null);
	const cb = file.cb: *readcb;
	file.op &= ~op::READV;
	file_epoll_ctl(file);

	if (ev.events & rt::EPOLLHUP != 0) {

		cb(file, io::EOF);

		cb(file, io::EOF)!;

	} else {
		const vec = file.rvec: []io::vector;
		const r = io::readv(file.fd, vec...);

		cb(file, r);

		cb(file, r)!;

	};
};

fn readv_cancel(req: *req) void = {
	const file = req.user: *file;
	assert(file.op & op::READV != 0);
	file.op &= ~op::READV;
	file_epoll_ctl(file);
};

// A callback for a [[write]] or [[writev]] operation.

export type writecb = fn(file: *file, result: (size | io::error)) void;

export type writecb = fn(file: *file, result: (size | io::error)) (void | nomem);

// Schedules a write operation on a file object. The provided buffer must be
// valid for the duration of the write operation.
export fn write(
	file: *file,
	cb: *writecb,
	buf: []u8,
) req = {
	file.wvbuf = io::mkvector(buf);
	// XXX: Bit of a hack to avoid allocating a slice
	const vec = (&file.wvbuf: *[*]io::vector)[..1];
	return writev(file, cb, vec...);
};

// Schedules a vectored read operation on a file object. The provided buffer
// must be valid for the duration of the write operation.
export fn writev(
	file: *file,
	cb: *writecb,
	vec: io::vector...
) req = {
	assert(file.op & op::WRITEV == 0);
	if (file.flags & fflags::BLOCKING != 0) {
		const r = io::writev(file.fd, vec...);

		cb(file, r);

		cb(file, r)!;

		return req { ... };
	};

	file.op |= op::WRITEV;
	file.cb2 = cb;
	file.wvec = vec;
	file_epoll_ctl(file);
	return mkreq(&writev_cancel, file);
};

fn writev_ready(file: *file, ev: *rt::epoll_event) void = {
	assert(file.op & op::WRITEV != 0);
	assert(file.cb != null);
	const vec = file.wvec: []io::vector;
	const r = io::writev(file.fd, vec...);
	const cb = file.cb2: *writecb;
	file.op &= ~op::WRITEV;
	file_epoll_ctl(file);

	cb(file, r);

	cb(file, r)!;

};

fn writev_cancel(req: *req) void = {
	const file = req.user: *file;
	assert(file.op & op::WRITEV != 0);
	file.op &= ~op::WRITEV;
	file_epoll_ctl(file);
};

use errors;
use io;
use rt;
use sort;
use time;
use types;
use unix::signal;

// Dispatch callback. See [[ondispatch]].
export type dispatchcb = fn(loop: *loop, user: nullable *opaque) void;

export type ondispatch = struct {
	cb: *dispatchcb,
	user: nullable *opaque,
	loop: *loop,
};

export type loop = struct {
	fd: io::file,
	events: []rt::epoll_event,
	dispatch: []*ondispatch,
	stop: bool,
};

// Creates a new event loop. The user must pass the return value to [[finish]]
// to free associated resources when done using the loop.

export fn newloop() (loop | errors::error) = {

export fn newloop() (loop | errors::error | nomem) = {

	const fd = match (rt::epoll_create1(rt::EPOLL_CLOEXEC)) {
	case let fd: int =>
		yield fd: io::file;
	case let err: rt::errno =>
		return errors::errno(err);
	};

	return loop {
		fd = fd,
		// XXX: Should the number of events be customizable?
		events = alloc([rt::epoll_event {
			events = 0,
			data = rt::epoll_data {
				fd = 0,
			}

		}...], 256),

		}...], 256)?,

		dispatch = [],
		stop = false,
	};
};

// Frees resources associated with an event loop. Must only be called once per
// event loop object. Calling finish invalidates all I/O objects associated with
// the event loop.
export fn finish(loop: *loop) void = {
	free(loop.events);
	io::close(loop.fd)!;
};

// Returns an [[io::file]] for this event loop which can be polled on when
// events are available for processing, for chaining together different event
// loops. The exact semantics of this function are platform-specific, and it may
// not be available for all implementations.
export fn loop_file(loop: *loop) io::file = {
	return loop.fd;
};

// Registers a callback to be invoked before the next call to [[dispatch]]
// processes requests. The callback may schedule additional I/O requests to be
// processed in this batch.
//
// Dispatch callbacks are only called once. If you wish to schedule another
// dispatch callback after the first, call [[do]] again.
export fn do(
	loop: *loop,
	cb: *dispatchcb,
	user: nullable *opaque = null,

) req = {

) (req | nomem) = {

	const dispatch = alloc(ondispatch {
		cb = cb,
		user = user,
		loop = loop,

	});
	append(loop.dispatch, dispatch);

	})?;
	append(loop.dispatch, dispatch)?;

	return mkreq(&do_cancel, dispatch);
};

fn do_cancel(req: *req) void = {
	const dispatch = req.user: *ondispatch;
	const loop = dispatch.loop;
	for (let i = 0z; i < len(loop.dispatch); i += 1) {
		if (loop.dispatch[i] == dispatch) {
			delete(loop.dispatch[i]);
			break;
		};
	};
	free(dispatch);
};

// Dispatches the event loop, waiting for new events and calling their callbacks
// as appropriate.
//
// A timeout of -1 will block indefinitely until the next event occurs. A
// timeout of 0 will cause dispatch to return immediately if no events are
// available to process. Portable use of the timeout argument supports only
// millisecond granularity of up to 24 days ([[types::INT_MAX]] milliseconds).
// Negative values other than -1 will cause the program to abort.
//
// Returns false if the loop has been stopped via [[stop]], or true otherwise.
export fn dispatch(
	loop: *loop,
	timeout: time::duration,

) (bool | errors::error) = {

) (bool | errors::error | nomem) = {

	const millis: int = if (timeout == -1) {
		yield -1;
	} else if (timeout < 0) {
		abort("ev::dispatch: invalid timeout");
	} else {
		yield (timeout / time::MILLISECOND): int;
	};
	if (loop.stop) {
		return false;
	};

	let todo = loop.dispatch;
	loop.dispatch = [];
	for (let dispatch .. todo) {
		dispatch.cb(loop, dispatch.user);
		free(dispatch);
	};
	free(todo);

	if (len(loop.events) == 0) {
		return true;
	};

	// TODO: Deal with signals
	const maxev = len(loop.events);
	assert(maxev <= types::INT_MAX: size, "ev::dispatch: too many events");
	const nevent = match(rt::epoll_pwait(
		loop.fd, &loop.events[0],
		maxev: int, millis, null)) {
	case let nevent: int =>
		yield nevent;
	case let err: rt::errno =>
		switch (err) {
		case rt::EINTR =>
			// We shallow system suspension error code
			return true;
		case =>
			abort("ev::dispatch: epoll_pwait failure");
		};
	};

	if (nevent == 0) {
		return true;
	};

	sort::sort(loop.events[..nevent], size(rt::epoll_event), &sort_events);

	for (let ev &.. loop.events[..nevent]) {
		const file = ev.data.ptr: *file;
		if (ev.events == 0) {
			continue;
		};
		const pending = file.op;
		if (ev.events & (rt::EPOLLIN | rt::EPOLLHUP) != 0
				&& file.op & op::READV != 0) {
			readv_ready(file, ev);
		};
		if (ev.events & (rt::EPOLLOUT | rt::EPOLLHUP) != 0
				&& file.op & op::WRITEV != 0) {
			writev_ready(file, ev);
		};
		switch (pending) {
		case op::NONE =>
			abort("No operation pending for ready object");
		case op::READABLE =>
			readable_ready(file, ev);
		case op::WRITABLE =>
			writable_ready(file, ev);
		case op::ACCEPT =>

			accept_ready(file, ev);

			accept_ready(file, ev)?;

		case op::CONNECT_TCP =>

			connect_tcp_ready(file, ev);

			connect_tcp_ready(file, ev)?;

		case op::CONNECT_UNIX =>

			connect_unix_ready(file, ev);

			connect_unix_ready(file, ev)?;

		case op::SIGNAL =>
			signal_ready(file, ev);
		case op::TIMER =>
			timer_ready(file, ev);
		case op::SENDTO =>
			sendto_ready(file, ev);
		case op::RECVFROM =>
			recvfrom_ready(file, ev);
		case op::SEND =>
			send_ready(file, ev);
		case op::RECV =>
			recv_ready(file, ev);
		case =>
			assert(pending & ~(op::READV | op::WRITEV) == 0);
		};
	};

	return !loop.stop;
};

// Signals the loop to stop processing events. If called during a callback, it
// will cause that invocation of [[dispatch]] to return false. Otherwise, false
// will be returned only upon the next call to [[dispatch]].
export fn stop(loop: *loop) void = {
	loop.stop = true;
};

// TODO: Expose full siginfo data for non-portable use
use errors;
use rt;
use unix::signal;

// Callback function for [[signal]] operations.
export type signalcb = fn(file: *file, sig: signal::sig) void;

// Registers a signal handler with this event loop. The signals specified will
// be masked so they are only raised via the provided callback. Closing this
// file will unmask the signals.
//
// It is not necessary to call [[signal]] again after the callback has
// processed; it will automatically re-register the operation for subsequent
// signals.
export fn signal(
	loop: *loop,
	cb: *signalcb,
	signals: signal::sig...

) (*file | errors::error) = {

) (*file | errors::error | nomem) = {

	const fd = signal::signalfd(signals...)?;
	const file = register(loop, fd)?;
	file.op = op::SIGNAL;
	file.cb = cb;
	file_epoll_ctl(file);
	signal::sigset_empty(&file.sigmask);
	signal::sigset_add(&file.sigmask, signals...);
	signal::block(signals...);
	return file;
};

fn signal_restore(file: *file) void = {

fn signal_restore(file: *file) (void | nomem) = {

	assert(file.op == op::SIGNAL);
	let buf: [rt::NSIG]signal::sig = [0...];
	let signals = buf[..0];
	for (let i = 1; i < rt::NSIG; i += 1) {
		const sig = i: signal::sig;
		if (signal::sigset_member(&file.sigmask, sig)) {

			static append(signals, sig);

			static append(signals, sig)?;

		};
	};
	signal::unblock(signals...);
};

fn signal_ready(file: *file, ev: *rt::epoll_event) void = {
	assert(file.op == op::SIGNAL);
	assert(file.cb != null);
	const cb = file.cb: *signalcb;
	const info = signal::read(file.fd)!;
	cb(file, info.signo);
};

use errors;
use net;
use net::ip;
use net::tcp;
use net::udp;
use net::unix;
use rt;

// Creates a socket which listens for incoming TCP connections on the given
// IP address and port.
export fn listen_tcp(
	loop: *loop,
	addr: ip::addr,
	port: u16,
	opts: tcp::listen_option...

) (*file | net::error | errors::error) = {

) (*file | net::error | errors::error | nomem) = {

	const sock = tcp::listen(addr, port, opts...)?;
	return register(loop, sock)?;
};

// Creates a socket which listens for incoming UDP packets on the given IP
// address and port.
export fn listen_udp(
	loop: *loop,
	addr: ip::addr,
	port: u16,
	opts: udp::listen_option...

) (*file | net::error | errors::error) = {

) (*file | net::error | errors::error | nomem) = {

	const sock = udp::listen(addr, port, opts...)?;
	return register(loop, sock)?;
};

// Creates a UNIX domain socket at the given path.
export fn listen_unix(
	loop: *loop,
	addr: unix::addr,
	opts: unix::listen_option...

) (*file | net::error | errors::error) = {

) (*file | net::error | errors::error | nomem) = {

	const sock = unix::listen(addr, opts...)?;
	return register(loop, sock)?;
};

// Creates a UDP socket on this event loop and sets the default destination to
// the given address.
export fn connect_udp(
	loop: *loop,
	dest: ip::addr,
	port: u16,
	opts: udp::connect_option...

) (*file | net::error | errors::error) = {

) (*file | net::error | errors::error | nomem) = {

	const sock = udp::connect(dest, port, opts...)?;
	const file = register(loop, sock)?;
	return file;
};

export type connectcb = fn(result: (*file | net::error), user: nullable *opaque) void;

// Creates a socket and connects to a given IP address and port over TCP.
//
// The variadic arguments accept [[net::sockflag]] and/or no more than one user
// data pointer. If the user data pointer is provided, it will be passed to the
// callback. This allows the user to pass a state object through the connection
// process:
//
//	let user: state = // ...
// 	ev::connect_tcp(&loop, &connected, addr, port, &user);
//
// 	fn connected(result: (*ev::file | net::error), user: nullable *opaque) void = {
// 		let user = user: *state;
// 	};
//
// The user data object provided will be assigned to the [[file]] which is
// provided to the callback after the connection is established.
//
// If you don't need a user data object you can just omit it:
//
// 	ev::connect_tcp(&loop, &connected, addr, port);
export fn connect_tcp(
	loop: *loop,
	cb: *connectcb,
	addr: ip::addr,
	port: u16,
	opts: (net::sockflag | *opaque)...

) (req | net::error | errors::error) = {

) (req | net::error | errors::error | nomem) = {

	// XXX: This doesn't let us set keepalive
	let opt: net::sockflag = 0;
	let user: nullable *opaque = null;
	for (let i = 0z; i < len(opts); i += 1) {
		match (opts[i]) {
		case let o: net::sockflag =>
			opt |= o;
		case let u: *opaque =>
			assert(user == null);
			user = u;
		};
	};
	const sock = tcp::connect(addr, port, opt | net::sockflag::NONBLOCK)?;
	let file = register(loop, sock)?;
	file.user = user;
	file.cb = cb;
	file.op = op::CONNECT_TCP;
	file_epoll_ctl(file);
	return mkreq(&connect_tcp_cancel, file);
};

// Connects to a UNIX domain socket.
//
// The variadic arguments accept [[net::sockflag]] and/or no more than one user
// data pointer. If the user data pointer is provided, it will be passed to the
// callback. This allows the user to pass a state object through the connection
// process:
//
//	let user: state = // ...
// 	ev::connect_user(&loop, &connected, addr, &user);
//
// 	fn connected(result: (*ev::file | net::error), user: nullable *opaque) void = {
// 		let user = user: *state;
// 	};
//
// The user data object provided will be assigned to the [[file]] which is
// provided to the callback after the connection is established.
//
// If you don't need a user data object you can just omit it:
//
// 	ev::connect_unix(&loop, &connected, addr);
export fn connect_unix(
	loop: *loop,
	cb: *connectcb,
	addr: unix::addr,
	opts: (net::sockflag | *opaque)...

) (req | net::error | errors::error) = {

) (req | net::error | errors::error | nomem) = {

	let opt: net::sockflag = 0;
	let user: nullable *opaque = null;
	for (let i = 0z; i < len(opts); i += 1) {
		match (opts[i]) {
		case let o: net::sockflag =>
			opt |= o;
		case let u: *opaque =>
			assert(user == null);
			user = u;
		};
	};
	const sock = unix::connect(addr, opt | net::sockflag::NONBLOCK)?;
	let file = register(loop, sock)?;
	file.user = user;
	file.cb = cb;
	file.op = op::CONNECT_UNIX;
	file_epoll_ctl(file);
	return mkreq(&connect_unix_cancel, file);
};

fn connect_tcp_ready(
	sock: *file,
	ev: *rt::epoll_event,

) void = {

) (void | nomem) = {

	assert(sock.op == op::CONNECT_TCP);
	assert(ev.events & rt::EPOLLOUT != 0);
	assert(sock.cb != null);
	const cb = sock.cb: *connectcb;
	sock.op &= ~op::CONNECT_TCP;
	file_epoll_ctl(sock);

	let errno = 0i, optsz = size(int): u32;
	rt::getsockopt(sock.fd, rt::SOL_SOCKET, rt::SO_ERROR, &errno, &optsz)!;
	if (errno != 0) {
		cb(errors::errno(errno), sock.user);

		close(sock);

		close(sock)?;

	} else {
		// XXX: If the user puts NONBLOCK into the opts provided at
		// [[connect_tcp]] we could try to preserve that here
		const fl = rt::fcntl(sock.fd, rt::F_GETFL, void)!;
		rt::fcntl(sock.fd, rt::F_SETFL, fl & ~rt::O_NONBLOCK)!;
		cb(sock, sock.user);
	};
};

fn connect_unix_ready(
	sock: *file,
	ev: *rt::epoll_event,

) void = {

) (void | nomem) = {

	assert(sock.op == op::CONNECT_UNIX);
	assert(ev.events & rt::EPOLLOUT != 0);
	assert(sock.cb != null);
	const cb = sock.cb: *connectcb;
	sock.op &= ~op::CONNECT_UNIX;
	file_epoll_ctl(sock);

	let errno = 0i, optsz = size(int): u32;
	rt::getsockopt(sock.fd, rt::SOL_SOCKET, rt::SO_ERROR, &errno, &optsz)!;
	if (errno != 0) {
		cb(errors::errno(errno), sock.user);

		close(sock);

		close(sock)?;

	} else {
		// XXX: If the user puts NONBLOCK into the opts provided at
		// [[connect_unix]] we could try to preserve that here
		const fl = rt::fcntl(sock.fd, rt::F_GETFL, void)!;
		rt::fcntl(sock.fd, rt::F_SETFL, fl & ~rt::O_NONBLOCK)!;
		cb(sock, sock.user);
	};
};

fn connect_tcp_cancel(req: *req) void = {
	const sock = req.user: *file;
	assert(sock.op == op::CONNECT_TCP);
	sock.op = op::NONE;
	file_epoll_ctl(sock);
};

fn connect_unix_cancel(req: *req) void = {
	const sock = req.user: *file;
	assert(sock.op == op::CONNECT_UNIX);
	sock.op = op::NONE;
	file_epoll_ctl(sock);
};

// A callback for an [[accept]] operation.

export type acceptcb = fn(file: *file, result: (*file | net::error)) void;

export type acceptcb = fn(file: *file, result: (*file | net::error)) (void | nomem);

// Schedules an accept operation on a socket.
export fn accept(
	sock: *file,
	cb: *acceptcb,
	flags: net::sockflag...
) req = {
	assert(sock.op == op::NONE);
	let fl: net::sockflag = 0;
	for (let i = 0z; i < len(flags); i += 1) {
		fl |= flags[i];
	};
	sock.op = op::ACCEPT;
	sock.cb = cb;
	sock.sockflag = fl;
	file_epoll_ctl(sock);
	return mkreq(&accept_cancel, sock);
};

fn accept_ready(
	sock: *file,
	ev: *rt::epoll_event,

) void = {

) (void | nomem) = {

	assert(sock.op == op::ACCEPT);
	assert(ev.events & rt::EPOLLIN != 0);
	assert(sock.cb != null);
	const cb = sock.cb: *acceptcb;
	sock.op = op::NONE;
	file_epoll_ctl(sock);

	const r = tcp::accept(sock.fd, sock.sockflag);
	match (r) {
	case let fd: net::socket =>
		// TODO: Bubble up errors from here?
		const file = register(sock.ev, fd)!;

		cb(sock, file);

		cb(sock, file)?;

	case let err: net::error =>

		cb(sock, err);

		cb(sock, err)?;

	};
};

fn accept_cancel(req: *req) void = {
	const sock = req.user: *file;
	assert(sock.op == op::ACCEPT);
	sock.op = op::NONE;
	file_epoll_ctl(sock);
};

// TODO: Support recv & send in parallel

// Callback for a [[recvfrom]] operation. The second parameter is either an
// error or a tuple of the number of bytes received and the IP address and port
// of the sender.
export type recvfromcb = fn(
	file: *file,
	r: ((size, ip::addr, u16) | net::error),
) void;

// Schedules a receive operation on a socket.
export fn recvfrom(
	sock: *file,
	cb: *recvfromcb,
	buf: []u8,
) req = {
	assert(sock.op == op::NONE);
	sock.op = op::RECVFROM;
	sock.cb = cb;
	sock.sendrecv.rbuf = buf;
	file_epoll_ctl(sock);
	return mkreq(&recvfrom_cancel, sock);
};

fn recvfrom_ready(
	sock: *file,
	ev: *rt::epoll_event,
) void = {
	assert(sock.op == op::RECVFROM);
	assert(sock.cb != null);
	const cb = sock.cb: *recvfromcb;
	sock.op = op::NONE;
	file_epoll_ctl(sock);

	let src: ip::addr = ip::ANY_V4, port = 0u16;
	match (udp::recvfrom(sock.fd, sock.sendrecv.rbuf, &src, &port)) {
	case let err: net::error =>
		cb(sock, err);
	case let n: size =>
		cb(sock, (n, src, port));
	};
};

fn recvfrom_cancel(req: *req) void = {
	const sock = req.user: *file;
	assert(sock.op == op::RECVFROM);
	sock.op = op::NONE;
	file_epoll_ctl(sock);
};

// Callback for a [[recv]] operation.
export type recvcb = fn(file: *file, r: (size | net::error)) void;

// Schedules a receive operation on a (connected) socket.
export fn recv(
	sock: *file,
	cb: *recvcb,
	buf: []u8,
) req = {
	assert(sock.op == op::NONE);
	sock.op = op::RECV;
	sock.cb = cb;
	sock.sendrecv.rbuf = buf;
	file_epoll_ctl(sock);
	return mkreq(&recv_cancel, sock);
};

fn recv_ready(
	sock: *file,
	ev: *rt::epoll_event,
) void = {
	assert(sock.op == op::RECV);
	assert(sock.cb != null);
	const cb = sock.cb: *recvcb;
	sock.op = op::NONE;
	file_epoll_ctl(sock);

	const r = udp::recv(sock.fd, sock.sendrecv.rbuf);
	cb(sock, r);
};

fn recv_cancel(req: *req) void = {
	const sock = req.user: *file;
	assert(sock.op == op::RECV);
	sock.op = op::NONE;
	file_epoll_ctl(sock);
};

// Callback for a [[send]] or [[sendto]] operation.
export type sendtocb = fn(file: *file, r: (size | net::error)) void;

// Schedules a send operation on a (connected) socket.
export fn send(
	sock: *file,
	cb: *sendtocb,
	buf: []u8,
) req = {
	assert(sock.op == op::NONE);
	sock.op = op::SEND;
	sock.cb = cb;
	sock.sendrecv.sbuf = buf;
	file_epoll_ctl(sock);
	return mkreq(&send_cancel, sock);
};

fn send_ready(
	sock: *file,
	ev: *rt::epoll_event,
) void = {
	assert(sock.op == op::SEND);
	assert(sock.cb != null);
	const cb = sock.cb: *sendtocb;
	sock.op = op::NONE;
	file_epoll_ctl(sock);

	const r = udp::send(sock.fd, sock.sendrecv.sbuf);
	cb(sock, r);
};

fn send_cancel(req: *req) void = {
	const sock = req.user: *file;
	assert(sock.op == op::SEND);
	sock.op = op::NONE;
	file_epoll_ctl(sock);
};

// Schedules a send operation on a socket.
export fn sendto(
	sock: *file,
	cb: *sendtocb,
	buf: []u8,
	dest: ip::addr,
	port: u16,
) req = {
	assert(sock.op == op::NONE);
	sock.op = op::SENDTO;
	sock.cb = cb;
	sock.sendrecv.sbuf = buf;
	sock.sendrecv.dest = dest;
	sock.sendrecv.port = port;
	file_epoll_ctl(sock);
	return mkreq(&sendto_cancel, sock);
};

fn sendto_ready(
	sock: *file,
	ev: *rt::epoll_event,
) void = {
	assert(sock.op == op::SENDTO);
	assert(sock.cb != null);
	const cb = sock.cb: *sendtocb;
	sock.op = op::NONE;
	file_epoll_ctl(sock);

	const r = udp::sendto(
		sock.fd,
		sock.sendrecv.sbuf,
		sock.sendrecv.dest,
		sock.sendrecv.port,
	);
	cb(sock, r);
};

fn sendto_cancel(req: *req) void = {
	const sock = req.user: *file;
	assert(sock.op == op::SENDTO);
	sock.op = op::NONE;
	file_epoll_ctl(sock);
};

use errors;
use io;
use rt;
use time;

// A callback which executes when a timer expires.
export type timercb = fn(file: *file) void;

// Creates a new timer. By default, this timer never expires; configure it with
// [[timer_configure]].
export fn newtimer(
	loop: *loop,
	cb: *timercb,
	clock: time::clock,

) (*file | errors::error) = {

) (*file | errors::error | nomem) = {

	const fd = match (rt::timerfd_create(clock,
			rt::TFD_NONBLOCK | rt::TFD_CLOEXEC)) {
	case let fd: int =>
		yield fd: io::file;
	case let errno: rt::errno =>
		return errors::errno(errno);
	};
	const file = register(loop, fd)?;
	file.op = op::TIMER;
	file.cb = cb;
	file_epoll_ctl(file);
	return file;
};

// Starts a timer created with [[newtimer]] to expire after the given "delay"
// and indefinitely thereafter following each interval of "repeat". Setting both
// values to zero disarms the timer; setting either value non-zero arms the
// timer.
export fn timer_configure(
	timer: *file,
	delay: time::duration,
	repeat: time::duration,
) void = {
	assert(timer.op == op::TIMER);
	let spec = rt::itimerspec {
		it_value = time::duration_to_timespec(delay),
		it_interval = time::duration_to_timespec(repeat),
		...
	};
	rt::timerfd_settime(timer.fd, 0, &spec, null)!;
};

fn timer_ready(timer: *file, ev: *rt::epoll_event) void = {
	assert(timer.op == op::TIMER);
	let buf: [8]u8 = [0...];
	match (io::read(timer.fd, buf)) {
	case errors::again =>
		// This can occur if the timer was reconfigured in an event
		// handler while the expiration event was pending in the same
		// dispatch of the event loop. Just discard the event in this
		// case.
		return;
	case (io::EOF | io::error) => abort();
	case => void;
	};

	assert(timer.cb != null);
	const cb = timer.cb: *timercb;
	cb(timer);
};

use ev;
use bufio;
use errors;
use io;
use memio;
use net::http;
use net::ip;
use net::tcp;
use net;
use strings;
use types;

// A callback for an [[http_serve]] operation.
export type http_servecb = fn(user: nullable *opaque, request: http::request, rw: http::response_writer) void;

// Schedules an http serve operation on a socket. The user must free resources
// when done with [[server_finish]]
export fn http_serve(
	sock: *ev::file,
	cb: *http_servecb,
	user: nullable *opaque,

) *server = {
	const serv = newserver(sock, cb, user);

) (*server | nomem) = {
	const serv = newserver(sock, cb, user)?;

	ev::setuser(sock, serv);
	ev::accept(sock, &http_accept);
	return serv;
};

fn http_accept(sock: *ev::file, r: (*ev::file | net::error)) void = {

fn http_accept(sock: *ev::file, r: (*ev::file | net::error)) (void | nomem) = {

	const server = ev::getuser(sock): *server;

	const sock = match (r) {
	case let sock: *ev::file =>
		yield sock;
	case let err: net::error =>
		// TODO handle it
		return;
	};

	const client = newclient(
		server,
		sock,

	);

	)?;

	ev::setuser(client.sock, client);
	ev::read(client.sock, &http_read, client.rbuf);

	append(server.clients, client);

	append(server.clients, client)?;

	ev::accept(server.sock, &http_accept);
};

fn http_read(sock: *ev::file, r: (size | io::EOF | io::error)) void = {

fn http_read(sock: *ev::file, r: (size | io::EOF | io::error)) (void | nomem) = {

	const client = ev::getuser(sock): *server_client;

	const n = match (r) {
	case let err: io::error =>

		client_close(client);

		client_close(client)?;

		return;
	case io::EOF =>

		client_close(client);

		client_close(client)?;

		return;
	case let n: size =>
		yield n;
	};

	io::seek(&client.buf, 0, io::whence::END)!;
	io::write(&client.buf, client.rbuf[..n])!;
	io::seek(&client.buf, 0, io::whence::SET)!;

	let scan = bufio::newscanner(&client.buf, types::SIZE_MAX);
	defer bufio::finish(&scan);

	let req = match (http::request_scan(&scan)) {
	case let req: http::request => yield req;
	case io::EOF =>
		ev::read(client.sock, &http_read, client.rbuf);
		return;
	case let err: (http::protoerr | errors::unsupported | io::error) =>

		client_close(client);

		client_close(client)?;

		return;
	};
	defer http::parsed_request_finish(&req);

	memio::reset(&client.buf);

	const resp = http::response {
		version = (1, 1),
		status = http::STATUS_OK,

		reason = strings::dup(http::status_reason(http::STATUS_OK)),

		reason = strings::dup(http::status_reason(http::STATUS_OK))?,

		header = [],
		body = io::empty,
	};
	defer io::close(resp.body)!;

	const cb = client.server.cb: *http_servecb;
	cb(client.server.user, req, http::response_writer {
		sock = ev::getfd(client.sock),
		resp = resp,
		writeto = &client.buf,
	});

	client.wbuf = memio::buffer(&client.buf);
	ev::write(client.sock, &http_write, client.wbuf);
};

fn http_write(sock: *ev::file, r: (size | io::error)) void = {

fn http_write(sock: *ev::file, r: (size | io::error)) (void | nomem) = {

	const client = ev::getuser(sock): *server_client;

	const n = match (r) {
	case let err: io::error =>

		client_close(client);

		client_close(client)?;

		return;
	case let n: size =>
		yield n;
	};
	static delete(client.wbuf[..n]);
	if (len(client.wbuf) != 0) {
		ev::write(client.sock, &http_write, client.wbuf);
		return;
	};

	memio::reset(&client.buf);
	ev::read(client.sock, &http_read, client.rbuf);
};

use ev;
use io;
use memio;
use os;

export type server = struct {
	sock: *ev::file,
	clients: []*server_client,
	cb: *opaque,
	user: nullable *opaque
};

export type server_client = struct {
	sock: *ev::file,
	server: *server,

	rbuf: [os::BUFSZ]u8,
	wbuf: []u8,
	buf: memio::stream,
};

export fn newserver(sock: *ev::file, cb: *opaque, user: nullable *opaque) *server = {

export fn newserver(sock: *ev::file, cb: *opaque, user: nullable *opaque) (*server | nomem) = {

	return alloc(server {
		sock = sock,
		cb = cb,
		user = user,
		...

	});

	})?;

};

export fn server_finish(serv: *server) void = {

export fn server_finish(serv: *server) (void | nomem) = {

	for (let i = 0z; i < len(serv.clients); i += 1) {

		client_close(serv.clients[i]);

		client_close(serv.clients[i])?;

		i -= 1;
	};
	free(serv);
};

export fn newclient(serv: *server, sock: *ev::file) *server_client = {

export fn newclient(serv: *server, sock: *ev::file) (*server_client | nomem) = {

	return alloc(server_client {
		server = serv,
		sock = sock,
		buf = memio::dynamic(),
		...

	});

	})?;

};

export fn client_close(client: *server_client) void = {

export fn client_close(client: *server_client) (void | nomem) = {

	const server = client.server;
	for (let i = 0z; i < len(server.clients); i += 1) {
		if (server.clients[i] == client) {
			delete(server.clients[i]);
			break;
		};
	};

	ev::close(client.sock);

	ev::close(client.sock)?;

	io::close(&client.buf)!;

	free(client);
};