//
// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
// file was obtained (LICENSE.txt).  A copy of the license may also be
// found online at https://opensource.org/licenses/MIT.
//

#include "nng/compat/nanomsg/nn.h"

// transports
#include "nng/compat/nanomsg/inproc.h"
#include "nng/compat/nanomsg/ipc.h"
#include "nng/compat/nanomsg/tcp.h"
#include "nng/compat/nanomsg/ws.h"

// protocols
#include "nng/compat/nanomsg/bus.h"
#include "nng/compat/nanomsg/pair.h"
#include "nng/compat/nanomsg/pipeline.h"
#include "nng/compat/nanomsg/pubsub.h"
#include "nng/compat/nanomsg/reqrep.h"
#include "nng/compat/nanomsg/survey.h"

// underlying NNG headers
#include "nng/nng.h"
#include "nng/protocol/bus0/bus.h"
#include "nng/protocol/pair0/pair.h"
#include "nng/protocol/pipeline0/pull.h"
#include "nng/protocol/pipeline0/push.h"
#include "nng/protocol/pubsub0/pub.h"
#include "nng/protocol/pubsub0/sub.h"
#include "nng/protocol/reqrep0/rep.h"
#include "nng/protocol/reqrep0/req.h"
#include "nng/protocol/survey0/respond.h"
#include "nng/protocol/survey0/survey.h"

#include "core/nng_impl.h"

#include <stdio.h>
#include <string.h>

// This file supplies the legacy compatibility API.  Applications should
// avoid using these if at all possible, and instead use the new style APIs.

static const struct {
	int nerr;
	int perr;
} nn_errnos[] = {
	// clang-format off
	{ NNG_EINTR,	    EINTR	  },
	{ NNG_ENOMEM,	    ENOMEM	  },
	{ NNG_EINVAL,	    EINVAL	  },
	{ NNG_EBUSY,	    EBUSY	  },
	{ NNG_ETIMEDOUT,    ETIMEDOUT	  },
	{ NNG_ECONNREFUSED, ECONNREFUSED  },
	{ NNG_ECLOSED,	    EBADF	  },
	{ NNG_EAGAIN,	    EAGAIN	  },
	{ NNG_ENOTSUP,	    ENOTSUP	  },
	{ NNG_EADDRINUSE,   EADDRINUSE	  },
	{ NNG_ESTATE,	    EFSM	  },
	{ NNG_ENOENT,	    ENOENT	  },
	{ NNG_EPROTO,	    EPROTO	  },
	{ NNG_EUNREACHABLE, EHOSTUNREACH  },
	{ NNG_EADDRINVAL,   EINVAL        },
	{ NNG_EPERM,	    EACCES	  },
	{ NNG_EMSGSIZE,	    EMSGSIZE	  },
	{ NNG_ECONNABORTED, ECONNABORTED  },
	{ NNG_ECONNRESET,   ECONNRESET	  },
	{ NNG_ECANCELED,    EBADF         },
	{ NNG_EEXIST,       EEXIST        },
	{ NNG_EWRITEONLY,   EACCES        },
	{ NNG_EREADONLY,    EACCES        },
	{ NNG_ECRYPTO,      EACCES        },
	{ NNG_EPEERAUTH,    EACCES        },
	{ NNG_EBADTYPE,     EINVAL        },
	{ NNG_EAMBIGUOUS,   EINVAL        },
	{ NNG_ENOFILES,     EMFILE        },
	{ NNG_ENOSPC,       ENOSPC        },
	{		 0,		0 },
	// clang-format on
};

const char *
nn_strerror(int err)
{
	int         i;
	static char msgbuf[32];

	for (i = 0; nn_errnos[i].perr != 0; i++) {
		if (nn_errnos[i].perr == err) {
			return (nng_strerror(nn_errnos[i].nerr));
		}
	}
	if (err == EIO) {
		return ("Unknown I/O error");
	}

	// Arguably we could use strerror() here, but we should only
	// be getting errnos we understand at this point.
	(void) snprintf(msgbuf, sizeof(msgbuf), "Unknown error %d", err);
	return (msgbuf);
}

static void
nn_seterror(int err)
{
	int i;

	for (i = 0; nn_errnos[i].nerr != 0; i++) {
		if (nn_errnos[i].nerr == err) {
			errno = nn_errnos[i].perr;
			return;
		}
	}
	// No idea...
	errno = EIO;
}

int
nn_errno(void)
{
	return (errno);
}

static const struct {
	uint16_t p_id;
	int (*p_open)(nng_socket *);
	int (*p_open_raw)(nng_socket *);
} nn_protocols[] = {
#ifdef NNG_HAVE_BUS0
	{
	    .p_id       = NN_BUS,
	    .p_open     = nng_bus0_open,
	    .p_open_raw = nng_bus0_open_raw,
	},
#endif
#ifdef NNG_HAVE_PAIR0
	{
	    .p_id       = NN_PAIR,
	    .p_open     = nng_pair0_open,
	    .p_open_raw = nng_pair0_open_raw,
	},
#endif
#ifdef NNG_HAVE_PULL0
	{
	    .p_id       = NN_PULL,
	    .p_open     = nng_pull0_open,
	    .p_open_raw = nng_pull0_open_raw,
	},
#endif
#ifdef NNG_HAVE_PUSH0
	{
	    .p_id       = NN_PUSH,
	    .p_open     = nng_push0_open,
	    .p_open_raw = nng_push0_open_raw,
	},
#endif
#ifdef NNG_HAVE_PUB0
	{
	    .p_id       = NN_PUB,
	    .p_open     = nng_pub0_open,
	    .p_open_raw = nng_pub0_open_raw,
	},
#endif
#ifdef NNG_HAVE_SUB0
	{
	    .p_id       = NN_SUB,
	    .p_open     = nng_sub0_open,
	    .p_open_raw = nng_sub0_open_raw,
	},
#endif
#ifdef NNG_HAVE_REQ0
	{
	    .p_id       = NN_REQ,
	    .p_open     = nng_req0_open,
	    .p_open_raw = nng_req0_open_raw,
	},
#endif
#ifdef NNG_HAVE_REP0
	{
	    .p_id       = NN_REP,
	    .p_open     = nng_rep0_open,
	    .p_open_raw = nng_rep0_open_raw,
	},
#endif
#ifdef NNG_HAVE_SURVEYOR0
	{
	    .p_id       = NN_SURVEYOR,
	    .p_open     = nng_surveyor0_open,
	    .p_open_raw = nng_surveyor0_open_raw,
	},
#endif
#ifdef NNG_HAVE_RESPONDENT0
	{
	    .p_id       = NN_RESPONDENT,
	    .p_open     = nng_respondent0_open,
	    .p_open_raw = nng_respondent0_open_raw,
	},
#endif
	{
	    .p_id = 0,
	},
};

int
nn_socket(int domain, int protocol)
{
	nng_socket sock;
	int        rv;
	int        i;

	if ((domain != AF_SP) && (domain != AF_SP_RAW)) {
		errno = EAFNOSUPPORT;
		return (-1);
	}

	for (i = 0; nn_protocols[i].p_id != 0; i++) {
		if (nn_protocols[i].p_id == protocol) {
			break;
		}
	}
	if (nn_protocols[i].p_open == NULL) {
		errno = ENOTSUP;
		return (-1);
	}

	if (domain == AF_SP_RAW) {
		rv = nn_protocols[i].p_open_raw(&sock);
	} else {
		rv = nn_protocols[i].p_open(&sock);
	}
	if (rv != 0) {
		nn_seterror(rv);
		return (-1);
	}

	// Legacy sockets have nodelay disabled.
	(void) nng_socket_set_bool(sock, NNG_OPT_TCP_NODELAY, false);
	return ((int) sock.id);
}

int
nn_close(int s)
{
	int        rv;
	nng_socket sid;

	sid.id = (uint32_t) s;

	if ((rv = nng_close(sid)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return (0);
}

int
nn_bind(int s, const char *addr)
{
	int          rv;
	nng_listener l;
	nng_socket   sid;

	sid.id = (uint32_t) s;
	if ((rv = nng_listen(sid, addr, &l, 0)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return ((int) l.id);
}

int
nn_connect(int s, const char *addr)
{
	int        rv;
	nng_dialer d;
	nng_socket sid;

	sid.id = (uint32_t) s;
	if ((rv = nng_dial(sid, addr, &d, NNG_FLAG_NONBLOCK)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return ((int) d.id);
}

int
nn_shutdown(int s, int ep)
{
	int rv;
	(void) s; // Unused
	nng_dialer   d;
	nng_listener l;

	// Socket is wired into the endpoint... so passing a bad endpoint
	// ID can result in affecting the wrong socket.  But this requires
	// a buggy application, and because we don't recycle endpoints
	// until wrap, its unlikely to actually come up in practice.
	// Note that listeners and dialers share the same namespace
	// in the core, so we can close either one this way.

	d.id = l.id = (uint32_t) ep;
	if (((rv = nng_dialer_close(d)) != 0) &&
	    ((rv = nng_listener_close(l)) != 0)) {
		nn_seterror(rv);
		return (-1);
	}
	return (0);
}

void *
nn_allocmsg(size_t size, int type)
{
	nng_msg *msg;
	int      rv;

	// Validate type and non-zero size.  This also checks for overflow.
	if ((type != 0) || (size < 1) || ((size + sizeof(msg) < size))) {
		nn_seterror(NNG_EINVAL);
		return (NULL);
	}

	// So our "messages" from nn are really going to be nng messages
	// but to make this work, we use a bit of headroom in the message
	// to stash the message header.
	if ((rv = nng_msg_alloc(&msg, size + (sizeof(msg)))) != 0) {
		nn_seterror(rv);
		return (NULL);
	}

	// This counts on message bodies being aligned sensibly.
	*(nng_msg **) (nng_msg_body(msg)) = msg;

	// We are counting on the implementation of nn_msg_trim to not
	// reallocate the message but just to leave the prefix inplace.
	(void) nng_msg_trim(msg, sizeof(msg));

	return (nng_msg_body(msg));
}

int
nn_freemsg(void *ptr)
{
	nng_msg *msg;

	msg = *(nng_msg **) (((char *) ptr) - sizeof(msg));
	nng_msg_free(msg);
	return (0);
}

void *
nn_reallocmsg(void *ptr, size_t len)
{
	nng_msg *msg;
	int      rv;

	if ((len + sizeof(msg)) < len) {
		// overflowed!
		nn_seterror(NNG_EINVAL);
		return (NULL);
	}

	// This counts on message bodies being aligned sensibly.
	msg = *(nng_msg **) (((char *) ptr) - sizeof(msg));

	// We need to realloc the requested len, plus size for our header.
	if ((rv = nng_msg_realloc(msg, len + sizeof(msg))) != 0) {
		// We don't free the old message.  Code is free to cope
		// as it sees fit.
		nn_seterror(rv);
		return (NULL);
	}
	// Stash the msg header pointer
	*(nng_msg **) (nng_msg_body(msg)) = msg;
	nng_msg_trim(msg, sizeof(msg));
	return (nng_msg_body(msg));
}

static int
nn_flags(int flags)
{
	switch (flags) {
	case 0:
		return (0);

	case NN_DONTWAIT:
		return (NNG_FLAG_NONBLOCK);

	default:
		nn_seterror(NNG_EINVAL);
		return (-1);
	}
}

int
nn_send(int s, const void *buf, size_t len, int flags)
{
	struct nn_iovec  iov;
	struct nn_msghdr hdr;

	iov.iov_base = (void *) buf;
	iov.iov_len  = len;

	hdr.msg_iov        = &iov;
	hdr.msg_iovlen     = 1;
	hdr.msg_control    = NULL;
	hdr.msg_controllen = 0;

	return (nn_sendmsg(s, &hdr, flags));
}

int
nn_recv(int s, void *buf, size_t len, int flags)
{
	struct nn_iovec  iov;
	struct nn_msghdr hdr;

	iov.iov_base = buf;
	iov.iov_len  = len;

	hdr.msg_iov        = &iov;
	hdr.msg_iovlen     = 1;
	hdr.msg_control    = NULL;
	hdr.msg_controllen = 0;

	return (nn_recvmsg(s, &hdr, flags));
}

int
nn_recvmsg(int s, struct nn_msghdr *mh, int flags)
{
	int        rv;
	nng_msg *  msg;
	size_t     len;
	int        keep = 0;
	nng_socket sid;

	if ((flags = nn_flags(flags)) == -1) {
		return (-1);
	}
	if (mh == NULL) {
		nn_seterror(NNG_EINVAL);
		return (-1);
	}
	if (mh->msg_iovlen < 0) {
		nn_seterror(NNG_EMSGSIZE);
		return (-1);
	}

	sid.id = (uint32_t) s;
	if ((rv = nng_recvmsg(sid, &msg, flags)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	if ((mh->msg_iovlen == 1) && (mh->msg_iov[0].iov_len == NN_MSG)) {
		// Receiver wants to have a dynamically allocated message.
		// There can only be one of these.
		if ((rv = nng_msg_insert(msg, &msg, sizeof(msg))) != 0) {
			nng_msg_free(msg);
			nn_seterror(rv);
			return (-1);
		}
		nng_msg_trim(msg, sizeof(msg));
		*(void **) (mh->msg_iov[0].iov_base) = nng_msg_body(msg);
		len                                  = nng_msg_len(msg);
		keep = 1; // Do not discard message!
	} else {
		// copyout to multiple iovecs.
		char *ptr = nng_msg_body(msg);
		len       = nng_msg_len(msg);

		for (int i = 0; i < mh->msg_iovlen; i++) {
			size_t n;
			if ((n = mh->msg_iov[i].iov_len) == NN_MSG) {
				// This is forbidden!
				nn_seterror(NNG_EINVAL);
				nng_msg_free(msg);
				return (-1);
			}
			if (n > len) {
				n = len;
			}
			memcpy(mh->msg_iov[i].iov_base, ptr, n);
			len -= n;
			ptr += n;
		}

		// If we copied everything, len will be zero, otherwise,
		// it represents the amount of data that we were unable to
		// copyout.  The caller is responsible for noticing this,
		// as there is no API to pass this information out.
		len = nng_msg_len(msg);
	}

	// If the caller has requested control information (header details),
	// we grab it.
	if (mh->msg_control != NULL) {
		char *             cdata;
		size_t             clen;
		size_t             tlen;
		size_t             spsz;
		struct nn_cmsghdr *hdr;

		spsz = nng_msg_header_len(msg);
		clen = NN_CMSG_SPACE(sizeof(spsz) + spsz);

		if ((tlen = mh->msg_controllen) == NN_MSG) {
			// Ideally we'd use the same msg, but we would need
			// to set up reference counts on the message, so
			// instead we just make a new message.
			nng_msg *nmsg;

			rv = nng_msg_alloc(&nmsg, clen + sizeof(nmsg));
			if (rv != 0) {
				nng_msg_free(msg);
				nn_seterror(rv);
				return (-1);
			}
			memcpy(nng_msg_body(nmsg), &nmsg, sizeof(nmsg));
			nng_msg_trim(nmsg, sizeof(nmsg));
			cdata                      = nng_msg_body(nmsg);
			*(void **) mh->msg_control = cdata;
			tlen                       = clen;
		} else {
			cdata = mh->msg_control;
			memset(cdata, 0,
			    tlen > sizeof(*hdr) ? sizeof(*hdr) : tlen);
		}

		if (clen <= tlen) {
			uint8_t *ptr    = NN_CMSG_DATA(cdata);
			hdr             = (void *) cdata;
			hdr->cmsg_len   = clen;
			hdr->cmsg_level = PROTO_SP;
			hdr->cmsg_type  = SP_HDR;

			memcpy(ptr, &spsz, sizeof(spsz));
			ptr += sizeof(spsz);
			memcpy(ptr, nng_msg_header(msg), spsz);
		}
	}

	if (!keep) {
		nng_msg_free(msg);
	}
	return ((int) len);
}

int
nn_sendmsg(int s, const struct nn_msghdr *mh, int flags)
{
	nng_msg *  msg  = NULL;
	nng_msg *  cmsg = NULL;
	nng_socket sid;
	char *     cdata;
	int        keep = 0;
	size_t     sz;
	int        rv;

	sid.id = (uint32_t) s;

	if ((flags = nn_flags(flags)) == -1) {
		return (-1);
	}

	if (mh == NULL) {
		nn_seterror(NNG_EINVAL);
		return (-1);
	}

	if (mh->msg_iovlen < 0) {
		nn_seterror(NNG_EMSGSIZE);
		return (-1);
	}

	if ((mh->msg_iovlen == 1) && (mh->msg_iov[0].iov_len == NN_MSG)) {
		char *bufp = *(char **) (mh->msg_iov[0].iov_base);

		msg  = *(nng_msg **) (bufp - sizeof(msg));
		keep = 1; // keep the message on error
	} else {
		char *ptr;
		int   i;

		sz = 0;
		// Get the total message size.
		for (i = 0; i < mh->msg_iovlen; i++) {
			sz += mh->msg_iov[i].iov_len;
		}
		if ((rv = nng_msg_alloc(&msg, sz)) != 0) {
			nn_seterror(rv);
			return (-1);
		}
		// Now copy it out.
		ptr = nng_msg_body(msg);
		for (i = 0; i < mh->msg_iovlen; i++) {
			memcpy(ptr, mh->msg_iov[i].iov_base,
			    mh->msg_iov[i].iov_len);
			ptr += mh->msg_iov[i].iov_len;
		}
	}

	// Now suck up the control data...
	// This POSIX-inspired API is one of the most painful for
	// usability we've ever seen.
	cmsg = NULL;
	if ((cdata = mh->msg_control) != NULL) {
		size_t         clen;
		size_t         offs;
		size_t         spsz;
		unsigned char *data;

		if ((clen = mh->msg_controllen) == NN_MSG) {
			// Underlying data is a message.  This is awkward,
			// because we have to copy the data, but we should
			// only free this message on success.  So we save the
			// message now.
			cdata = *(void **) cdata;
			cmsg  = *(nng_msg **) (cdata - sizeof(cmsg));
			clen  = nng_msg_len(cmsg);
		} else {
			clen = mh->msg_controllen;
		}

		offs = 0;
		while ((offs + sizeof(NN_CMSG_LEN(0))) < clen) {
			struct nn_cmsghdr *chdr = (void *) (cdata + offs);
			if ((chdr->cmsg_level != PROTO_SP) ||
			    (chdr->cmsg_type != SP_HDR)) {
				offs += chdr->cmsg_len;
			}

			// SP header in theory.  Starts with size, then
			// any backtrace details.
			if (chdr->cmsg_len < sizeof(size_t)) {
				offs += chdr->cmsg_len;
				continue;
			}
			data = NN_CMSG_DATA(chdr);
			memcpy(&spsz, data, sizeof(spsz));
			if ((spsz + sizeof(spsz)) > chdr->cmsg_len) {
				// Truncated header?  Ignore it.
				offs += chdr->cmsg_len;
				continue;
			}
			data += sizeof(spsz);
			rv = nng_msg_header_append(msg, data, spsz);
			if (rv != 0) {
				if (!keep) {
					nng_msg_free(msg);
				}
				nn_seterror(rv);
				return (-1);
			}

			break;
		}
	}

	sz = nng_msg_len(msg);
	if ((rv = nng_sendmsg(sid, msg, flags)) != 0) {
		if (!keep) {
			nng_msg_free(msg);
		}
		nn_seterror(rv);
		return (-1);
	}

	if (cmsg != NULL) {
		// We sent successfully, so free up the control message.
		nng_msg_free(cmsg);
	}
	return ((int) sz);
}

static int
nn_getdomain(nng_socket s, void *valp, size_t *szp)
{
	int  i;
	bool b;
	int  rv;

	if ((rv = nng_socket_get_bool(s, NNG_OPT_RAW, &b)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	i = b ? AF_SP_RAW : AF_SP;
	memcpy(valp, &i, *szp < sizeof(int) ? *szp : sizeof(int));
	*szp = sizeof(int);
	return (0);
}

#ifndef NNG_PLATFORM_WINDOWS
#define SOCKET int
#endif

static int
nn_getfd(nng_socket s, void *valp, size_t *szp, const char *opt)
{
	int    ifd;
	int    rv;
	SOCKET sfd;

	if ((rv = nng_socket_get_int(s, opt, &ifd)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	sfd = (SOCKET) ifd;
	memcpy(valp, &sfd, *szp < sizeof(sfd) ? *szp : sizeof(sfd));
	*szp = sizeof(sfd);
	return (0);
}

static int
nn_getrecvfd(nng_socket s, void *valp, size_t *szp)
{
	return (nn_getfd(s, valp, szp, NNG_OPT_RECVFD));
}

static int
nn_getsendfd(nng_socket s, void *valp, size_t *szp)
{
	return (nn_getfd(s, valp, szp, NNG_OPT_SENDFD));
}

static int
nn_getzero(nng_socket s, void *valp, size_t *szp)
{
	int zero = 0;
	NNI_ARG_UNUSED(s);
	memcpy(valp, &zero, *szp < sizeof(zero) ? *szp : sizeof(zero));
	*szp = sizeof(zero);
	return (0);
}

static int
nn_setignore(nng_socket s, const void *valp, size_t sz)
{
	NNI_ARG_UNUSED(valp);
	NNI_ARG_UNUSED(s);
	if (sz != sizeof(int)) {
		nn_seterror(NNG_EINVAL);
		return (-1);
	}
	return (0);
}

static int
nn_getwsmsgtype(nng_socket s, void *valp, size_t *szp)
{
	int val = NN_WS_MSG_TYPE_BINARY;
	NNI_ARG_UNUSED(s);
	memcpy(valp, &val, *szp < sizeof(val) ? *szp : sizeof(val));
	*szp = sizeof(val);
	return (0);
}

static int
nn_setwsmsgtype(nng_socket s, const void *valp, size_t sz)
{
	int val;
	NNI_ARG_UNUSED(s);
	if (sz != sizeof(val)) {
		nn_seterror(NNG_EINVAL);
		return (-1);
	}
	memcpy(&val, valp, sizeof(val));
	if (val != NN_WS_MSG_TYPE_BINARY) {
		nn_seterror(NNG_EINVAL);
		return (-1);
	}
	return (0);
}

static int
nn_settcpnodelay(nng_socket s, const void *valp, size_t sz)
{
	bool val;
	int  ival;
	int  rv;

	if (sz != sizeof(ival)) {
		errno = EINVAL;
		return (-1);
	}
	memcpy(&ival, valp, sizeof(ival));
	switch (ival) {
	case 0:
		val = false;
		break;
	case 1:
		val = true;
		break;
	default:
		nn_seterror(NNG_EINVAL);
		return (-1);
	}

	if ((rv = nng_socket_set_bool(s, NNG_OPT_TCP_NODELAY, val)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return (0);
}

static int
nn_gettcpnodelay(nng_socket s, void *valp, size_t *szp)
{
	bool val;
	int  ival;
	int  rv;

	if ((rv = nng_socket_get_bool(s, NNG_OPT_TCP_NODELAY, &val)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	ival = val ? 1 : 0;
	memcpy(valp, &ival, *szp < sizeof(ival) ? *szp : sizeof(ival));
	*szp = sizeof(ival);
	return (0);
}

static int
nn_getrcvbuf(nng_socket s, void *valp, size_t *szp)
{
	int cnt;
	int rv;

	if ((rv = nng_socket_get_int(s, NNG_OPT_RECVBUF, &cnt)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	cnt *= 1024;
	memcpy(valp, &cnt, *szp < sizeof(cnt) ? *szp : sizeof(cnt));
	*szp = sizeof(cnt);
	return (0);
}

static int
nn_setrcvbuf(nng_socket s, const void *valp, size_t sz)
{
	int cnt;
	int rv;

	if (sz != sizeof(cnt)) {
		errno = EINVAL;
		return (-1);
	}
	memcpy(&cnt, valp, sizeof(cnt));
	// Round up to a whole number of kilobytes, then divide by kB to
	// go from buffer size in bytes to messages.  This is a coarse
	// estimate, and assumes messages are 1kB on average.
	cnt += 1023;
	cnt /= 1024;
	if ((rv = nng_socket_set_int(s, NNG_OPT_RECVBUF, cnt)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return (0);
}

static int
nn_getsndbuf(nng_socket s, void *valp, size_t *szp)
{
	int cnt;
	int rv;

	if ((rv = nng_socket_get_int(s, NNG_OPT_SENDBUF, &cnt)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	cnt *= 1024;
	memcpy(valp, &cnt, *szp < sizeof(cnt) ? *szp : sizeof(cnt));
	*szp = sizeof(cnt);
	return (0);
}

static int
nn_setsndbuf(nng_socket s, const void *valp, size_t sz)
{
	int cnt;
	int rv;

	if (sz != sizeof(cnt)) {
		errno = EINVAL;
		return (-1);
	}
	memcpy(&cnt, valp, sizeof(cnt));
	// Round up to a whole number of kilobytes, then divide by kB to
	// go from buffer size in bytes to messages.  This is a coarse
	// estimate, and assumes messages are 1kB on average.
	cnt += 1023;
	cnt /= 1024;
	if ((rv = nng_socket_set_int(s, NNG_OPT_SENDBUF, cnt)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return (0);
}

static int
nn_setrcvmaxsz(nng_socket s, const void *valp, size_t sz)
{
	int    ival;
	size_t val;
	int    rv;

	if (sz != sizeof(ival)) {
		errno = EINVAL;
		return (-1);
	}
	memcpy(&ival, valp, sizeof(ival));
	if (ival == -1) {
		val = 0;
	} else if (ival >= 0) {
		// Note that if the user sets 0, it disables the limit.
		// This is a different semantic.
		val = (size_t) ival;
	} else {
		errno = EINVAL;
		return (-1);
	}
	if ((rv = nng_socket_set_size(s, NNG_OPT_RECVMAXSZ, val)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return (0);
}

static int
nn_getrcvmaxsz(nng_socket s, void *valp, size_t *szp)
{
	int    ival;
	int    rv;
	size_t val;

	if ((rv = nng_socket_get_size(s, NNG_OPT_RECVMAXSZ, &val)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	// Legacy uses -1 to mean unlimited.  New code uses 0.  Note that
	// as a consequence, we can't set a message limit of zero.
	// We report any size beyond 2GB as effectively unlimited.
	// There is an implicit assumption here that ints are 32-bits,
	// but that's generally true of any platform we support.
	if ((val == 0) || (val > 0x7FFFFFFF)) {
		ival = -1;
	} else {
		ival = (int) val;
	}
	memcpy(valp, &ival, *szp < sizeof(ival) ? *szp : sizeof(ival));
	*szp = sizeof(ival);
	return (0);
}

// options which we convert -- most of the array is initialized at run time.
static const struct {
	int         nnlevel;
	int         nnopt;
	const char *opt;
	int (*get)(nng_socket, void *, size_t *);
	int (*set)(nng_socket, const void *, size_t);
} options[] = {
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_LINGER,
	    .get     = nn_getzero,
	    .set     = nn_setignore,
	}, // review
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_DOMAIN,
	    .get     = nn_getdomain,
	    .set     = NULL,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_RCVBUF,
	    .get     = nn_getrcvbuf,
	    .set     = nn_setrcvbuf,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_SNDBUF,
	    .get     = nn_getsndbuf,
	    .set     = nn_setsndbuf,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_RECONNECT_IVL,
	    .opt     = NNG_OPT_RECONNMINT,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_RECONNECT_IVL_MAX,
	    .opt     = NNG_OPT_RECONNMAXT,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_SNDFD,
	    .opt     = NNG_OPT_SENDFD,
	    .get     = nn_getsendfd,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_RCVFD,
	    .opt     = NNG_OPT_RECVFD,
	    .get     = nn_getrecvfd,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_RCVMAXSIZE,
	    .get     = nn_getrcvmaxsz,
	    .set     = nn_setrcvmaxsz,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_MAXTTL,
	    .opt     = NNG_OPT_MAXTTL,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_RCVTIMEO,
	    .opt     = NNG_OPT_RECVTIMEO,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_SNDTIMEO,
	    .opt     = NNG_OPT_SENDTIMEO,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_PROTOCOL,
	    .opt     = NNG_OPT_PROTO,
	},
	{
	    .nnlevel = NN_SOL_SOCKET,
	    .nnopt   = NN_SOCKET_NAME,
	    .opt     = NNG_OPT_SOCKNAME,
	},
	{
	    .nnlevel = NN_REQ,
	    .nnopt   = NN_REQ_RESEND_IVL,
	    .opt     = NNG_OPT_REQ_RESENDTIME,
	},
	{
	    .nnlevel = NN_SUB,
	    .nnopt   = NN_SUB_SUBSCRIBE,
	    .opt     = NNG_OPT_SUB_SUBSCRIBE,
	},
	{
	    .nnlevel = NN_SUB,
	    .nnopt   = NN_SUB_UNSUBSCRIBE,
	    .opt     = NNG_OPT_SUB_UNSUBSCRIBE,
	},
	{
	    .nnlevel = NN_SURVEYOR,
	    .nnopt   = NN_SURVEYOR_DEADLINE,
	    .opt     = NNG_OPT_SURVEYOR_SURVEYTIME,
	},
	{
	    .nnlevel = NN_TCP,
	    .nnopt   = NN_TCP_NODELAY,
	    .get     = nn_gettcpnodelay,
	    .set     = nn_settcpnodelay,
	},
	{
	    .nnlevel = NN_WS,
	    .nnopt   = NN_WS_MSG_TYPE,
	    .get     = nn_getwsmsgtype,
	    .set     = nn_setwsmsgtype,
	}
	// XXX: IPV4ONLY, SNDPRIO, RCVPRIO
};

int
nn_getsockopt(int s, int nnlevel, int nnopt, void *valp, size_t *szp)
{
	const char *name                         = NULL;
	int (*get)(nng_socket, void *, size_t *) = NULL;
	int        rv;
	nng_socket sid;

	sid.id = (uint32_t) s;

	for (unsigned i = 0; i < sizeof(options) / sizeof(options[0]); i++) {
		if ((options[i].nnlevel == nnlevel) &&
		    (options[i].nnopt == nnopt)) {
			get  = options[i].get;
			name = options[i].opt;
			break;
		}
	}

	if (get != NULL) {
		return (get(sid, valp, szp));
	}

	if (name == NULL) {
		errno = ENOPROTOOPT;
		return (-1);
	}

	if ((rv = nng_socket_get(sid, name, valp, szp)) != 0) {
		nn_seterror(rv);
		return (-1);
	}

	return (0);
}

int
nn_setsockopt(int s, int nnlevel, int nnopt, const void *valp, size_t sz)
{
	nng_socket  sid;
	const char *name                             = NULL;
	int (*set)(nng_socket, const void *, size_t) = NULL;
	int rv;

	sid.id = (uint32_t) s;

	for (unsigned i = 0; i < sizeof(options) / sizeof(options[0]); i++) {
		if ((options[i].nnlevel == nnlevel) &&
		    (options[i].nnopt == nnopt)) {

			set  = options[i].set;
			name = options[i].opt;
			break;
		}
	}

	if (set != NULL) {
		return (set(sid, valp, sz));
	}

	if (name == NULL) {
		errno = ENOPROTOOPT;
		return (-1);
	}

	if ((rv = nng_socket_set(sid, name, valp, sz)) != 0) {
		nn_seterror(rv);
		return (-1);
	}
	return (0);
}

struct nn_cmsghdr *
nn_cmsg_next(struct nn_msghdr *mh, struct nn_cmsghdr *first)
{
	size_t clen;
	char * data;

	// We only support SP headers, so there can be at most one header.
	if (first != NULL) {
		return (NULL);
	}
	if ((clen = mh->msg_controllen) == NN_MSG) {
		nng_msg *msg;
		data = *((void **) (mh->msg_control));
		msg  = *(nng_msg **) (data - sizeof(msg));
		clen = nng_msg_len(msg);
	} else {
		data = mh->msg_control;
	}

	if (first == NULL) {
		first = (void *) data;
	} else {
		first = first + first->cmsg_len;
	}

	if (((char *) first + sizeof(*first)) > (data + clen)) {
		return (NULL);
	}
	return (first);
}

int
nn_device(int s1, int s2)
{
	int        rv;
	nng_socket sid1;
	nng_socket sid2;

	sid1.id = (uint32_t) s1;
	sid2.id = (uint32_t) s2;

	rv = nng_device(sid1, sid2);
	// rv must always be nonzero
	nn_seterror(rv);
	return (-1);
}

// Windows stuff.
#ifdef NNG_PLATFORM_WINDOWS
#define poll WSAPoll
#ifndef WIN32_LEAN_AND_MEAN
#define WIN32_LEAN_AND_MEAN
#endif
#include <windows.h>
#include <mswsock.h>
#elif defined NNG_PLATFORM_POSIX
#include <poll.h>
#endif

int
nn_poll(struct nn_pollfd *fds, int nfds, int timeout)
{
// This function is rather unfortunate.  poll() is available
// on POSIX, and on Windows as WSAPoll.  On other systems it might
// not exist at all.  We could also benefit from using a notification
// that didn't have to access file descriptors... sort of access to
// the pollable element on the socket.  We don't have that, so we
// just use poll.  This function is definitely suboptimal compared to
// using callbacks.
#if defined(NNG_PLATFORM_WINDOWS) || defined(NNG_PLATFORM_POSIX)
	struct pollfd *pfd;
	int            npfd;
	int            rv;

	if ((pfd = NNI_ALLOC_STRUCTS(pfd, nfds * 2)) == NULL) {
		errno = ENOMEM;
		return (-1);
	}

	// First prepare the master polling structure.
	npfd = 0;
	for (int i = 0; i < nfds; i++) {
		int fd;
		if (fds[i].events & NN_POLLIN) {
			nng_socket s;
			s.id = fds[i].fd;
			if ((rv = nng_socket_get_int(s, NNG_OPT_RECVFD, &fd)) !=
			    0) {
				nn_seterror(rv);
				NNI_FREE_STRUCTS(pfd, nfds * 2);
				return (-1);
			}
#ifdef NNG_PLATFORM_WINDOWS
			pfd[npfd].fd = (SOCKET) fd;
#else
			pfd[npfd].fd = fd;
#endif
			pfd[npfd].events = POLLIN;
			npfd++;
		}
		if (fds[i].events & NN_POLLOUT) {
			nng_socket s;
			s.id = fds[i].fd;
			if ((rv = nng_socket_get_int(s, NNG_OPT_SENDFD, &fd)) !=
			    0) {
				nn_seterror(rv);
				NNI_FREE_STRUCTS(pfd, nfds * 2);
				return (-1);
			}
#ifdef NNG_PLATFORM_WINDOWS
			pfd[npfd].fd = (SOCKET) fd;
#else
			pfd[npfd].fd = fd;
#endif
			pfd[npfd].events = POLLIN;
			npfd++;
		}
	}

	rv = poll(pfd, npfd, timeout);
	if (rv < 0) {
		int e = errno;
		NNI_FREE_STRUCTS(pfd, nfds * 2);
		errno = e;
		return (-1);
	}

	// Now update the nn_poll from the system poll.
	npfd = 0;
	rv   = 0;
	for (int i = 0; i < nfds; i++) {
		fds[i].revents = 0;
		if (fds[i].events & NN_POLLIN) {
			if (pfd[npfd].revents & POLLIN) {
				fds[i].revents |= NN_POLLIN;
			}
			npfd++;
		}
		if (fds[i].events & NN_POLLOUT) {
			if (pfd[npfd].revents & POLLIN) {
				fds[i].revents |= NN_POLLOUT;
			}
			npfd++;
		}
		if (fds[i].revents) {
			rv++;
		}
	}
	NNI_FREE_STRUCTS(pfd, nfds * 2);
	return (rv);

#else // NNG_PLATFORM_WINDOWS or NNG_PLATFORM_POSIX
	NNI_ARG_UNUSED(fds);
	NNI_ARG_UNUSED(nfds);
	NNI_ARG_UNUSED(timeout);
	errno = ENOTSUP;
	return (-1);
#endif
}

// nn_term is suitable only for shutting down the entire library,
// and is not thread-safe with other functions.
void
nn_term(void)
{
	// This function is relatively toxic, since it can affect
	// all sockets in the process, including those
	// in use by libraries, etc.  Accordingly, do not use this
	// in a library -- only e.g. atexit() and similar.
	nni_sock_closeall();
}

uint64_t
nn_get_statistic(int x, int y)
{
	(void) x;
	(void) y;

	return (0);
}