/* ==========================================================================
 * libevnet/src/bufio/socket.c - Network server library for libevent.
 * --------------------------------------------------------------------------
 * Copyright (c) 2006  Barracuda Networks, Inc.
 * Copyright (c) 2006  William Ahern
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to permit
 * persons to whom the Software is furnished to do so, subject to the
 * following conditions:
 *
 * The above copyright notice and this permission notice shall be included
 * in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
 * OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
 * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
 * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
 * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
 * USE OR OTHER DEALINGS IN THE SOFTWARE.
 * ==========================================================================
 */
#include <limits.h>	/* PIPE_BUF */

#include <errno.h>	/* EAGAIN EFAULT EPIPE ENOTSUP */

#include <assert.h>

#include <string.h>	/* memcpy(3) */

#include <sys/param.h>	/* MIN */

#include <sys/queue.h>	/* SLIST */

#include <sys/types.h>
#include <sys/time.h>	/* struct timeval timercmp(3) */

#if _WIN32
#include <winsock2.h>	/* send(2) recv(2) */
#else
#include <sys/socket.h>	/* recv(2) send(2) */
#endif

#include <windows.h>	/* GetLastError SetLastError */

#include <winsock2.h>	/* INVALID_SOCKET */

#include <event.h>	/* EV_READ EV_WRITE EV_TIMEOUT event_set(3)
			 * event_base_set(3) event_add(3) */

#include <arena/proto.h>

#include "tls.h"
#include "bufio.h"
#include "pagebuf.h"
#include "socket.h"

#include <stdio.h>
#define MARK (fprintf(stderr, "%s:%d\n", __FUNCTION__, __LINE__))


/*
 * Extension of timercmp() which treats NULL pointers as "unset" timevals.
 */
static const struct timeval tv_zero;

#define timereq(a, b)	\
	timercmp(((a)? (a) : &tv_zero), ((b)? (b) : &tv_zero), ==)


/*
 * Stack frame handling code for optimized event management. We postpone
 * deleting an event so we can reuse it if an immediate poll is requested. 
 * And we need to keep state--about live call frames--in case our object(s)
 * become invalid during a callback.
 */
#define BUFIO_SOCKET_FRAME_PUSH(s, m, f) do {				\
	(f)->sp	= &(s);							\
	SLIST_INSERT_HEAD(&(s)->m.poll.stack, (f), sle);		\
} while(0)

#define BUFIO_SOCKET_FRAME_POP(s, m, f) do {				\
	if ((s) != 0) {							\
		assert((f) == SLIST_FIRST(&(s)->m.poll.stack));		\
		SLIST_REMOVE_HEAD(&(s)->m.poll.stack, sle);		\
	}								\
} while(0)

#define BUFIO_SOCKET_FRAME_OKAY(f)	(*(f)->sp != 0)


const struct bufio_socket bufio_socket_initializer = {
	.fd	= INVALID_SOCKET,
}; /* bufio_socket_initializer */


const struct bufio_socket_options bufio_socket_defaults = {
	.page_size	= 4096,
}; /* bufio_socket_defaults */


static void bufio_socket_source_poll_handler(int, short, void *);
static void bufio_socket_sink_poll_handler(int, short, void *);


static enum bufio_errno bufio_socket_tls_errno_tr(struct bufio_socket *s) {
	switch (tls_errno(s->tls)) {
	case TLS_ESUCCESS:
		return 0;
	case TLS_ESYSTEM:
		return BUFIO_ESYSTEM;
	case TLS_ETIMEDOUT:
		return BUFIO_ETIMEDOUT;
	case TLS_ECANCELLED:
		return BUFIO_ECANCELLED;
	default:
		SetLastError(EFAULT);

		return BUFIO_ESYSTEM;
	}

	/* NOT REACHED */	
} /* bufio_socket_tls_errno_tr() */


static size_t bufio_socket_read(struct bufio_socket *s, void *buf, size_t bufsiz, enum bufio_errno *e) {
	ssize_t n;

	if (s->tls) {
		if (-1 == (n = tls_read(s->tls, buf, bufsiz, 0, 0, 0))) {
			*e	= s->source.lasterr = bufio_socket_tls_errno_tr(s);
			n	= 0;

#ifdef WSAEWOULDBLOCK
			if (GetLastError() == WSAEWOULDBLOCK)
				SetLastError(EAGAIN);
#endif
		} else if (n == 0)
			*e	= s->source.lasterr = BUFIO_EEOF;
	} else {
		if (s->fd == BUFIO_SOCKET_FD_WAIT) {
			*e	= s->source.lasterr = BUFIO_ESYSTEM;

			SetLastError(EAGAIN);

			n	= 0;
		} else if (-1 == (n = recv(s->fd, buf, bufsiz, 0))) {
			*e	= s->source.lasterr = BUFIO_ESYSTEM;
			n	= 0;

#ifdef WSAEWOULDBLOCK
			if (GetLastError() == WSAEWOULDBLOCK)
				SetLastError(EAGAIN);
#endif
		} else if (n == 0)
			*e	= s->source.lasterr = BUFIO_EEOF;
	}

	if (n == 0 && *e == BUFIO_ESYSTEM && GetLastError() == EAGAIN)
		*e	= BUFIO_EAGAIN;

	return n;
} /* bufio_socket_read() */


static size_t bufio_socket_write(struct bufio_socket *s, void *buf, size_t bufsiz, enum bufio_errno *e) {
	ssize_t n;

	if (bufsiz == 0)
		return 0;

	if (s->tls) {
		if (-1 == (n = tls_write(s->tls, buf, bufsiz, 0, 0, 0))) {
			*e	= s->sink.lasterr = bufio_socket_tls_errno_tr(s);
			n	= 0;

#ifdef WSAEWOULDBLOCK
			if (GetLastError() == WSAEWOULDBLOCK)
				SetLastError(EAGAIN);
#endif
		} else if (n == 0) {
			*e	= s->sink.lasterr = BUFIO_ESYSTEM;

			SetLastError(EPIPE);
		}
	} else {
		if (s->fd == BUFIO_SOCKET_FD_WAIT) {
			*e	= s->sink.lasterr = BUFIO_ESYSTEM;

			SetLastError(EAGAIN);

			n	= 0;
		} else if (-1 == (n = send(s->fd, buf, bufsiz, 0))) {
			*e	= s->sink.lasterr = BUFIO_ESYSTEM;
			n	= 0;

#ifdef WSAEWOULDBLOCK
			if (GetLastError() == WSAEWOULDBLOCK)
				SetLastError(EAGAIN);
#endif
		} else if (n == 0) {
			*e	= s->sink.lasterr = BUFIO_ESYSTEM;

			SetLastError(EPIPE);
		}
	}

	if (n == 0 && *e == BUFIO_ESYSTEM) {
		if (GetLastError() == EAGAIN)
			*e	= BUFIO_EAGAIN;
		else if (GetLastError() == EPIPE)
			*e	= BUFIO_EPIPE;
	}

	return n;
} /* bufio_socket_write() */


static size_t bufio_socket_copyout(struct bufio_source *src, void *buf, size_t bufsiz, int flags, enum bufio_errno *e) {
	struct bufio_socket *s	= BUFIO_CAST_FROM(src, bufio_socket, bufio_source);

	if (flags & BUFIO_PEEK) {
		*e	= s->source.lasterr = BUFIO_ESYSTEM;

		SetLastError(ENOTSUP);

		return 0;
	}

	return bufio_socket_read(s, buf, bufsiz, e);
} /* bufio_socket_copyout() */


static size_t bufio_socket_copyto(struct bufio_source *src, struct bufio_sink *snk, int flags, enum bufio_errno *e) {
	struct bufio_socket *s	= BUFIO_CAST_FROM(src, bufio_socket, bufio_source);

	return snk->copyfrom(snk, src, flags, e);
} /* bufio_socket_copyto() */


static size_t bufio_socket_copyin(struct bufio_sink *snk, void *buf, size_t bufsiz, int flags, enum bufio_errno *e) {
	struct bufio_socket *s	= BUFIO_CAST_FROM(snk, bufio_socket, bufio_sink);

	return bufio_socket_write(s, buf, bufsiz, e);
} /* bufio_socket_copyin() */


static size_t bufio_socket_copyfrom(struct bufio_sink *snk, struct bufio_source *src, int flags, enum bufio_errno *e) {
	struct bufio_socket *s	= BUFIO_CAST_FROM(snk, bufio_socket, bufio_sink);

	return src->copyto(src, snk, flags, e);
} /* bufio_socket_copyfrom() */


static short bufio_socket_event_tr(struct bufio_socket *s, short want) {
	if (!s->tls)
		return want;

	switch ((TLS_S_NEED_READ | TLS_S_NEED_WRITE) & tls_state(s->tls)) {
	case TLS_S_NEED_READ:
		return EV_READ;
	case TLS_S_NEED_WRITE:
		return EV_WRITE;
	case TLS_S_NEED_READ|TLS_S_NEED_WRITE:
		return EV_READ|EV_WRITE;
	default:
		return want;
	}

	/* NOT REACHED */
} /* bufio_socket_event_tr() */


static void bufio_socket_event_timer_wake(int fd, short events, void *arg) {
	struct bufio_socket_event *e	= arg;
	struct bufio_socket *s;

	timerclear(&e->timer.timeout);	/* Timers are always one-shot. */

	if (e->notify && e->want && (s = e->timer.ev_so)) {
		e->timer.ev_so	= 0;

		e->notify(s->fd, EV_TIMEOUT, s);
	} /* else log some information about this odd condition. */

	return /* void */;
} /* bufio_socket_event_timer_wake() */


static void bufio_socket_event_timer_stop(struct bufio_socket *s, struct bufio_socket_event *e) {
	if (!timerisset(&e->timer.timeout))
		return /* void */;

	evtimer_del(&e->timer.ev_op);

	e->timer.ev_so	= 0;

	timerclear(&e->timer.timeout);

	return /* void */;
} /* bufio_socket_event_timer_stop() */


static void bufio_socket_event_timer_start(struct bufio_socket *s, struct bufio_socket_event *e, struct timeval *tv) {
	if (timerisset(&e->timer.timeout))
		bufio_socket_event_timer_stop(s, e);

	if (tv == 0 || !timerisset(tv))
		return /* void */;

	evtimer_set(&e->timer.ev_op, &bufio_socket_event_timer_wake, e);

	if (s->ev_base)
		event_base_set(s->ev_base, &e->timer.ev_op);

	evtimer_add(&e->timer.ev_op, tv);

	e->timer.ev_so		= s;
	e->timer.timeout	= *tv;

	return /* void */;
} /* bufio_socket_event_timer_start() */


static void bufio_socket_event_run(int fd, short events, void *arg) {
	struct bufio_socket *s	= arg;
	/*
	 * We fire the sink event right off the bat. Since the source event
	 * comes second, we need to make sure the state hasn't changed so
	 * that we can mimic as best as possible the behavior the
	 * application would see if libevent was directly firing the second
	 * event. Checking the number of calls to the source poller
	 * guarantees that if there's an intervening cancel and then a
	 * subsequent poll (all from the sink callback), the event won't
	 * fire til another roundtrip through the event loop.
	 */
	unsigned int ncalls	= s->source.poll.ncalls;
	struct bufio_socket_frame f;

	BUFIO_SOCKET_FRAME_PUSH(s, source, &f);

	if (s->sink.event.want & events)
		s->sink.event.notify(fd, events, s);

	if (!BUFIO_SOCKET_FRAME_OKAY(&f))
		return /* void */;

	if (ncalls == s->source.poll.ncalls && (s->source.event.want & events))
		s->source.event.notify(fd, events, s);

	BUFIO_SOCKET_FRAME_POP(s, source, &f);

	return /* void */;
} /* bufio_socket_event_run() */


static enum bufio_errno bufio_socket_event_mod(struct bufio_socket *s) {
	short want;

	if (s->ev_pending == (want = (s->source.event.want | s->sink.event.want)))
		return 0;

	if (s->ev_pending)
		(void)event_del(&s->ev_op), s->ev_pending = 0;

	/*
	 * If we have shim descriptor, just fake the event enqueue. When
	 * bufio_socket_set_fd() gives a valid descriptor it will manually
	 * re-issue this routine.
	 */
	if (want == 0 || s->fd == BUFIO_SOCKET_FD_WAIT)
		return 0;

	event_set(&s->ev_op, s->fd, want | EV_PERSIST, &bufio_socket_event_run, s);

	if (s->ev_base)
		event_base_set(s->ev_base, &s->ev_op);

	if (0 != event_add(&s->ev_op, 0))
		return BUFIO_ESYSTEM;

	s->ev_pending	= want;

	return 0;
} /* bufio_socket_event_mod() */


static enum bufio_errno bufio_socket_event_del(struct bufio_socket *s, struct bufio_socket_event *e) {
	e->want		= 0;
	e->notify	= 0;

	bufio_socket_event_timer_stop(s, e);

	return bufio_socket_event_mod(s);
} /* bufio_socket_event_del() */


static enum bufio_errno bufio_socket_event_add(struct bufio_socket *s, struct bufio_socket_event *e, short events, void (*fn)(int, short, void *), struct timeval *tv) {
	e->want		= events;
	e->notify	= fn;

	bufio_socket_event_timer_start(s, e, tv); /* Begin the stopwatch. */

	return bufio_socket_event_mod(s);
} /* bufio_socket_event_add() */


static void bufio_socket_source_poll_handler(int fd, short events, void *arg) {
	struct bufio_socket *s		= arg;
	unsigned int ncalls		= s->source.poll.ncalls;
	bufio_source_poll_cb fn;
	enum bufio_errno retval;
	struct bufio_socket_frame f;

	BUFIO_SOCKET_FRAME_PUSH(s, source, &f);

	s->source.event.want	= 0;

	bufio_socket_event_timer_stop(s, &s->source.event);

	fn		= s->source.cb.fn;
	arg		= s->source.cb.arg;
	retval		= s->source.cb.retval;

	s->source.cb.fn		= 0;
	s->source.cb.arg	= 0;
	s->source.cb.retval	= 0;

	if (events & EV_TIMEOUT)
		retval	= s->source.lasterr = BUFIO_ETIMEDOUT;

	fn(BUFIO_CAST_TO(s, bufio_socket, bufio_source), retval, arg);

	if (!BUFIO_SOCKET_FRAME_OKAY(&f))
		return /* void */;

	if (s->source.poll.ncalls == ncalls)
		(void)bufio_socket_event_del(s, &s->source.event);

	BUFIO_SOCKET_FRAME_POP(s, source, &f);

	return /* void */;
} /* bufio_socket_source_poll_handler() */


static enum bufio_errno bufio_socket_source_poll(struct bufio_source *src, bufio_source_poll_cb cb, void *arg, struct timeval *timeout) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(src, bufio_socket, bufio_source);
	short want			= bufio_socket_event_tr(s, EV_READ);
	enum bufio_errno our_errno;

	++s->source.poll.ncalls;

	s->source.cb.fn		= cb;
	s->source.cb.arg	= arg;
	s->source.cb.retval	= 0;

	if (0 != (our_errno = bufio_socket_event_add(s, &s->source.event, want, &bufio_socket_source_poll_handler, timeout)))
		return s->source.lasterr = our_errno;

	return 0;	
} /* bufio_socket_source_poll() */


static enum bufio_errno bufio_socket_source_cancel(struct bufio_source *src, int notify) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(src, bufio_socket, bufio_source);
	bufio_source_poll_cb fn;
	void *arg;

	(void)bufio_socket_event_del(s, &s->source.event);

	if (!s->source.cb.fn)
		return BUFIO_ENOTPOLLING;

	fn	= s->source.cb.fn;
	arg	= s->source.cb.arg;

	s->source.cb.fn		= 0;
	s->source.cb.arg	= 0;
	s->source.cb.retval	= 0;

	if (notify)
		fn(BUFIO_CAST_TO(s, bufio_socket, bufio_source), BUFIO_ECANCELLED, arg);

	return 0;
} /* bufio_socket_source_cancel() */


static enum bufio_errno bufio_socket_source_lasterr(struct bufio_source *src) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(src, bufio_socket, bufio_source);

	return s->source.lasterr;
} /* bufio_socket_source_lasterr() */


static void bufio_socket_source_clearerr(struct bufio_source *src) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(src, bufio_socket, bufio_source);

	s->source.lasterr	= 0;

	return /* void */;
} /* bufio_socket_source_clearerr() */


static void bufio_socket_sink_poll_handler(int fd, short events, void *arg) {
	struct bufio_socket *s		= arg;
	unsigned int ncalls		= s->sink.poll.ncalls;
	bufio_sink_poll_cb fn;
	enum bufio_errno retval;
	struct bufio_socket_frame f;

	BUFIO_SOCKET_FRAME_PUSH(s, sink, &f);

	s->sink.event.want	= 0;

	bufio_socket_event_timer_stop(s, &s->sink.event);

	fn		= s->sink.cb.fn;
	arg		= s->sink.cb.arg;
	retval		= s->sink.cb.retval;

	s->sink.cb.fn		= 0;
	s->sink.cb.arg		= 0;
	s->sink.cb.retval	= 0;

	if (events & EV_TIMEOUT)
		retval	= s->sink.lasterr = BUFIO_ETIMEDOUT;

	fn(BUFIO_CAST_TO(s, bufio_socket, bufio_sink), retval, arg);

	if (!BUFIO_SOCKET_FRAME_OKAY(&f))
		return /* void */;

	if (s->sink.poll.ncalls == ncalls)
		(void)bufio_socket_event_del(s, &s->sink.event);

	BUFIO_SOCKET_FRAME_POP(s, sink, &f);

	return /* void */;
} /* bufio_socket_sink_poll_handler() */


static enum bufio_errno bufio_socket_sink_poll(struct bufio_sink *src, bufio_sink_poll_cb cb, void *arg, struct timeval *timeout) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(src, bufio_socket, bufio_sink);
	short want			= bufio_socket_event_tr(s, EV_WRITE);
	enum bufio_errno our_errno;

	++s->sink.poll.ncalls;

	s->sink.cb.fn		= cb;
	s->sink.cb.arg		= arg;
	s->sink.cb.retval	= 0;

	if (0 != (our_errno = bufio_socket_event_add(s, &s->sink.event, want, &bufio_socket_sink_poll_handler, timeout)))
		return s->sink.lasterr = our_errno;

	return 0;	
} /* bufio_socket_sink_poll() */


static enum bufio_errno bufio_socket_sink_cancel(struct bufio_sink *snk, int notify) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(snk, bufio_socket, bufio_sink);
	bufio_sink_poll_cb fn;
	void *arg;

	(void)bufio_socket_event_del(s, &s->sink.event);

	if (!s->sink.cb.fn)
		return BUFIO_ENOTPOLLING;

	fn	= s->sink.cb.fn;
	arg	= s->sink.cb.arg;

	s->sink.cb.fn		= 0;
	s->sink.cb.arg		= 0;
	s->sink.cb.retval	= 0;

	if (notify)
		fn(BUFIO_CAST_TO(s, bufio_socket, bufio_sink), BUFIO_ECANCELLED, arg);

	return 0;
} /* bufio_socket_sink_cancel() */


static enum bufio_errno bufio_socket_sink_lasterr(struct bufio_sink *snk) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(snk, bufio_socket, bufio_sink);

	return s->sink.lasterr;
} /* bufio_socket_sink_lasterr() */

static void bufio_socket_sink_clearerr(struct bufio_sink *snk) {
	struct bufio_socket *s		= BUFIO_CAST_FROM(snk, bufio_socket, bufio_sink);

	s->sink.lasterr	= 0;

	return /* void */;
} /* bufio_socket_sink_clearerr() */


struct bufio_socket *bufio_socket_init(struct bufio_socket *s, int fd, const struct bufio_socket_options *opts, struct event_base *evb, const struct arena_prototype *ap, enum bufio_errno *e) {
	static const struct bufio_socket bufio_socket_initializer;
	struct bufio_source *src;
	struct bufio_sink *snk;

	*s		= bufio_socket_initializer;
	s->fd		= fd;
	s->opts		= *((opts != 0)? opts : &bufio_socket_defaults);
	s->ev_base	= evb;
	s->ap		= ap;

	src		= BUFIO_CAST_TO(s, bufio_socket, bufio_source);
	*src		= BUFIO_SOURCE_INITIALIZER;
	src->copyto	= &bufio_socket_copyto;
	src->copyout	= &bufio_socket_copyout;
	src->poll	= &bufio_socket_source_poll;
	src->cancel	= &bufio_socket_source_cancel;
	src->lasterr	= &bufio_socket_source_lasterr;
	src->clearerr	= &bufio_socket_source_clearerr;

	snk		= BUFIO_CAST_TO(s, bufio_socket, bufio_sink);
	*snk		= BUFIO_SINK_INITIALIZER;
	snk->copyfrom	= &bufio_socket_copyfrom;
	snk->copyin	= &bufio_socket_copyin;
	snk->poll	= &bufio_socket_sink_poll;
	snk->cancel	= &bufio_socket_sink_cancel;
	snk->lasterr	= &bufio_socket_sink_lasterr;
	snk->clearerr	= &bufio_socket_sink_clearerr;

	SLIST_INIT(&s->source.poll.stack);
	SLIST_INIT(&s->sink.poll.stack);

	return s;
} /* bufio_socket_init() */


void bufio_socket_destroy(struct bufio_socket *s) {
	struct bufio_socket_frame *fp;

	if (!s)
		return /* void */;

	SLIST_FOREACH(fp, &s->source.poll.stack, sle)
		*fp->sp	= 0;

	SLIST_FOREACH(fp, &s->sink.poll.stack, sle)
		*fp->sp	= 0;

	SLIST_INIT(&s->source.poll.stack);
	SLIST_INIT(&s->sink.poll.stack);

	s->source.event.want	= 0;
	s->sink.event.want	= 0;

	(void)bufio_socket_event_del(s, &s->source.event);
	(void)bufio_socket_event_del(s, &s->sink.event);

	return /* void */;
} /* bufio_socket_destroy() */


struct bufio_socket *bufio_socket_open(int fd, const struct bufio_socket_options *opts, struct event_base *evb, const struct arena_prototype *ap, enum bufio_errno *e) {
	struct bufio_socket *s	= 0;
	int sys_errno;

	if (!(s = ap->malloc(ap, sizeof *s, 0)))
		goto sysfail;

	return bufio_socket_init(s, fd, opts, evb, ap, e);
sysfail:
	*e	= BUFIO_ESYSTEM;

	sys_errno	= GetLastError();

	bufio_socket_destroy(s);

	SetLastError(sys_errno);

	return 0;
} /* bufio_socket_open() */


void bufio_socket_close(struct bufio_socket *s) {
	if (s == 0)
		return /* void */;

	bufio_socket_destroy(s);

	s->ap->free(s->ap, s);

	return /* void */;
} /* bufio_socket_close() */


void bufio_socket_switch_tls(struct bufio_socket *s, struct tls *tls) {
	s->tls	= tls;

	return /* void */;
} /* bufio_socket_switch_tls() */


void bufio_socket_set_fd(struct bufio_socket *s, int fd) {
	/*
	 * Must remove any libevent handler so we don't get confused over
	 * which descriptor is actually registered for polling.
	 */
	if (s->ev_pending)
		(void)event_del(&s->ev_op), s->ev_pending = 0;

	s->fd	= fd;

	(void)bufio_socket_event_mod(s);

	return /* void */;		
} /* bufio_socket_set_fd() */


struct bufio_source *bufio_socket_to_source(struct bufio_socket *s) {
	return BUFIO_CAST_TO(s, bufio_socket, bufio_source);
} /* bufio_socket_to_source() */


struct bufio_sink *bufio_socket_to_sink(struct bufio_socket *s) {
	return BUFIO_CAST_TO(s, bufio_socket, bufio_sink);
} /* bufio_socket_to_sink() */