/*

 * Copyright (c) 2009-2012 Niels Provos, Nick Mathewson

 *

 * Redistribution and use in source and binary forms, with or without

 * modification, are permitted provided that the following conditions

 * are met:

 * 1. Redistributions of source code must retain the above copyright

 *    notice, this list of conditions and the following disclaimer.

 * 2. Redistributions in binary form must reproduce the above copyright

 *    notice, this list of conditions and the following disclaimer in the

 *    documentation and/or other materials provided with the distribution.

 * 3. The name of the author may not be used to endorse or promote products

 *    derived from this software without specific prior written permission.

 *

 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR

 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES

 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.

 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,

 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT

 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF

 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 */

#include <sys/types.h>

#ifdef WIN32

#include <winsock2.h>

#endif

#include "event2/event-config.h"

#include "event2/util.h"

#include "event2/buffer.h"

#include "event2/bufferevent.h"

#include "event2/bufferevent_struct.h"

#include "event2/event.h"

#include "defer-internal.h"

#include "bufferevent-internal.h"

#include "mm-internal.h"

#include "util-internal.h"

struct bufferevent_pair {

	struct bufferevent_private bev;

	struct bufferevent_pair *partner;

};

/* Given a bufferevent that's really a bev part of a bufferevent_pair,

 * return that bufferevent_filtered. Returns NULL otherwise.*/

static inline struct bufferevent_pair *

upcast(struct bufferevent *bev)

{

	struct bufferevent_pair *bev_p;

	if (bev->be_ops != &bufferevent_ops_pair)

		return NULL;

	bev_p = EVUTIL_UPCAST(bev, struct bufferevent_pair, bev.bev);

	EVUTIL_ASSERT(bev_p->bev.bev.be_ops == &bufferevent_ops_pair);

	return bev_p;

}

#define downcast(bev_pair) (&(bev_pair)->bev.bev)

static inline void

incref_and_lock(struct bufferevent *b)

{

	struct bufferevent_pair *bevp;

	_bufferevent_incref_and_lock(b);

	bevp = upcast(b);

	if (bevp->partner)

		_bufferevent_incref_and_lock(downcast(bevp->partner));

}

static inline void

decref_and_unlock(struct bufferevent *b)

{

	struct bufferevent_pair *bevp = upcast(b);

	if (bevp->partner)

		_bufferevent_decref_and_unlock(downcast(bevp->partner));

	_bufferevent_decref_and_unlock(b);

}

/* XXX Handle close */

static void be_pair_outbuf_cb(struct evbuffer *,

    const struct evbuffer_cb_info *, void *);

static struct bufferevent_pair *

bufferevent_pair_elt_new(struct event_base *base,

    int options)

{

	struct bufferevent_pair *bufev;

	if (! (bufev = mm_calloc(1, sizeof(struct bufferevent_pair))))

		return NULL;

	if (bufferevent_init_common(&bufev->bev, base, &bufferevent_ops_pair,

		options)) {

		mm_free(bufev);

		return NULL;

	}

	if (!evbuffer_add_cb(bufev->bev.bev.output, be_pair_outbuf_cb, bufev)) {

		bufferevent_free(downcast(bufev));

		return NULL;

	}

	_bufferevent_init_generic_timeout_cbs(&bufev->bev.bev);

	return bufev;

}

int

bufferevent_pair_new(struct event_base *base, int options,

    struct bufferevent *pair[2])

{

	struct bufferevent_pair *bufev1 = NULL, *bufev2 = NULL;

	int tmp_options;

	options |= BEV_OPT_DEFER_CALLBACKS;

	tmp_options = options & ~BEV_OPT_THREADSAFE;

	bufev1 = bufferevent_pair_elt_new(base, options);

	if (!bufev1)

		return -1;

	bufev2 = bufferevent_pair_elt_new(base, tmp_options);

	if (!bufev2) {

		bufferevent_free(downcast(bufev1));

		return -1;

	}

	if (options & BEV_OPT_THREADSAFE) {

		/*XXXX check return */

		bufferevent_enable_locking(downcast(bufev2), bufev1->bev.lock);

	}

	bufev1->partner = bufev2;

	bufev2->partner = bufev1;

	evbuffer_freeze(downcast(bufev1)->input, 0);

	evbuffer_freeze(downcast(bufev1)->output, 1);

	evbuffer_freeze(downcast(bufev2)->input, 0);

	evbuffer_freeze(downcast(bufev2)->output, 1);

	pair[0] = downcast(bufev1);

	pair[1] = downcast(bufev2);

	return 0;

}

static void

be_pair_transfer(struct bufferevent *src, struct bufferevent *dst,

    int ignore_wm)

{

	size_t src_size, dst_size;

	size_t n;

	evbuffer_unfreeze(src->output, 1);

	evbuffer_unfreeze(dst->input, 0);

	if (dst->wm_read.high) {

		dst_size = evbuffer_get_length(dst->input);

		if (dst_size < dst->wm_read.high) {

			n = dst->wm_read.high - dst_size;

			evbuffer_remove_buffer(src->output, dst->input, n);

		} else {

			if (!ignore_wm)

				goto done;

			n = evbuffer_get_length(src->output);

			evbuffer_add_buffer(dst->input, src->output);

		}

	} else {

		n = evbuffer_get_length(src->output);

		evbuffer_add_buffer(dst->input, src->output);

	}

	if (n) {

		BEV_RESET_GENERIC_READ_TIMEOUT(dst);

		if (evbuffer_get_length(dst->output))

			BEV_RESET_GENERIC_WRITE_TIMEOUT(dst);

		else

			BEV_DEL_GENERIC_WRITE_TIMEOUT(dst);

	}

	src_size = evbuffer_get_length(src->output);

	dst_size = evbuffer_get_length(dst->input);

	if (dst_size >= dst->wm_read.low) {

		_bufferevent_run_readcb(dst);

	}

	if (src_size <= src->wm_write.low) {

		_bufferevent_run_writecb(src);

	}

done:

	evbuffer_freeze(src->output, 1);

	evbuffer_freeze(dst->input, 0);

}

static inline int

be_pair_wants_to_talk(struct bufferevent_pair *src,

    struct bufferevent_pair *dst)

{

	return (downcast(src)->enabled & EV_WRITE) &&

	    (downcast(dst)->enabled & EV_READ) &&

	    !dst->bev.read_suspended &&

	    evbuffer_get_length(downcast(src)->output);

}

static void

be_pair_outbuf_cb(struct evbuffer *outbuf,

    const struct evbuffer_cb_info *info, void *arg)

{

	struct bufferevent_pair *bev_pair = arg;

	struct bufferevent_pair *partner = bev_pair->partner;

	incref_and_lock(downcast(bev_pair));

	if (info->n_added > info->n_deleted && partner) {

		/* We got more data.  If the other side's reading, then

		   hand it over. */

		if (be_pair_wants_to_talk(bev_pair, partner)) {

			be_pair_transfer(downcast(bev_pair), downcast(partner), 0);

		}

	}

	decref_and_unlock(downcast(bev_pair));

}

static int

be_pair_enable(struct bufferevent *bufev, short events)

{

	struct bufferevent_pair *bev_p = upcast(bufev);

	struct bufferevent_pair *partner = bev_p->partner;

	incref_and_lock(bufev);

	if (events & EV_READ) {

		BEV_RESET_GENERIC_READ_TIMEOUT(bufev);

	}

	if ((events & EV_WRITE) && evbuffer_get_length(bufev->output))

		BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);

	/* We're starting to read! Does the other side have anything to write?*/

	if ((events & EV_READ) && partner &&

	    be_pair_wants_to_talk(partner, bev_p)) {

		be_pair_transfer(downcast(partner), bufev, 0);

	}

	/* We're starting to write! Does the other side want to read? */

	if ((events & EV_WRITE) && partner &&

	    be_pair_wants_to_talk(bev_p, partner)) {

		be_pair_transfer(bufev, downcast(partner), 0);

	}

	decref_and_unlock(bufev);

	return 0;

}

static int

be_pair_disable(struct bufferevent *bev, short events)

{

	if (events & EV_READ) {

		BEV_DEL_GENERIC_READ_TIMEOUT(bev);

	}

	if (events & EV_WRITE)

		BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);

	return 0;

}

static void

be_pair_destruct(struct bufferevent *bev)

{

	struct bufferevent_pair *bev_p = upcast(bev);

	if (bev_p->partner) {

		bev_p->partner->partner = NULL;

		bev_p->partner = NULL;

	}

	_bufferevent_del_generic_timeout_cbs(bev);

}

static int

be_pair_flush(struct bufferevent *bev, short iotype,

    enum bufferevent_flush_mode mode)

{

	struct bufferevent_pair *bev_p = upcast(bev);

	struct bufferevent *partner;

	incref_and_lock(bev);

	if (!bev_p->partner)

		return -1;

	partner = downcast(bev_p->partner);

	if (mode == BEV_NORMAL)

		return 0;

	if ((iotype & EV_READ) != 0)

		be_pair_transfer(partner, bev, 1);

	if ((iotype & EV_WRITE) != 0)

		be_pair_transfer(bev, partner, 1);

	if (mode == BEV_FINISHED) {

		_bufferevent_run_eventcb(partner, iotype|BEV_EVENT_EOF);

	}

	decref_and_unlock(bev);

	return 0;

}

struct bufferevent *

bufferevent_pair_get_partner(struct bufferevent *bev)

{

	struct bufferevent_pair *bev_p;

	struct bufferevent *partner = NULL;

	bev_p = upcast(bev);

	if (! bev_p)

		return NULL;

	incref_and_lock(bev);

	if (bev_p->partner)

		partner = downcast(bev_p->partner);

	decref_and_unlock(bev);

	return partner;

}

const struct bufferevent_ops bufferevent_ops_pair = {

	"pair_elt",

	evutil_offsetof(struct bufferevent_pair, bev.bev),

	be_pair_enable,

	be_pair_disable,

	be_pair_destruct,

	_bufferevent_generic_adj_timeouts,

	be_pair_flush,

	NULL, /* ctrl */

};