libepistle/msg_server.c

/*    This file is part of libepistle.
 *
 *    libepistle is free software: you can redistribute it and/or modify
 *    it under the terms of the GNU Lesser General Public License as published by
 *    the Free Software Foundation, either version 3 of the License, or
 *    (at your option) any later version.
 *
 *    libepistle is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Lesser General Public License for more details.
 *
 *    You should have received a copy of the GNU Lesser General Public License
 *    along with libepistle.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <unistd.h>
#include <sys/epoll.h>

#include "err.h"
#include "polling.h"
#include "msg_server.h"

struct tmp_extra {
	Msg_server *server;
	Add_msgr add;
	Remove_msgr remove;
	React_msg react;
	void *extra;
};

int
msg_server_init(Msg_server *server, int epoll_fd, uint32_t max_len,
		Disposer *extra_dspr, Err *err)
{
	server->max_len = max_len;

	if (msgrs_init(&server->msgrs, extra_dspr, err) < 0)
	        return -1;

	if (epoll_fd < 0 && (epoll_fd = epoll_create1(0)) < 0) {
		err_std(err);
		map_dispose(&server->msgrs);
		return -1;
	}

        server->epoll_fd = epoll_fd;
	return 0;
}

void
msg_server_dispose(Msg_server *server)
{
	close(server->epoll_fd);
	map_dispose(&server->msgrs);
}

int
msg_server_on(Msg_server *server, int socket, Err *err)
{
	return poll_add(server->epoll_fd, socket, err);
}

int
msg_server_off(Msg_server *server, int socket, Err *err)
{
	return poll_remove(server->epoll_fd, socket, err);
}

static int
s_check(int fd, void *extra)
{
	struct tmp_extra *tmp_extra = extra;

	/* If msgrs doesn't contains the fd it is a server */
	return !set_contains(&tmp_extra->server->msgrs, sizeof(fd), &fd);
}

static int
msg_react(int epoll_fd, int fd, int server_fd, int pm_event,
	  Err *err, void *extra)
{
	struct tmp_extra *tmp_extra = extra;
	Msg_server *server = tmp_extra->server;
	Msgr *msgr;

	(void) epoll_fd;

	switch (pm_event) {
	case PM_ADD:
		return tmp_extra->add(server, fd, server_fd,
				     err, tmp_extra->extra);
	case PM_REMOVE:
		return tmp_extra->remove(server, fd, err,
					tmp_extra->extra);
	case PM_INPUT:
		msgr = msgr_get(&server->msgrs, sizeof(fd), &fd);

		/* we haven't yet dealt with previous message! */
		if (msgr->wip && msg_tnsfd(msgr->wip))
			return tmp_extra->react(server, msgr, err,
					       tmp_extra->extra);

		if (msgr_recv(msgr, server->max_len, 0, err) < 0)
			return tmp_extra->remove(server, fd, err,
						tmp_extra->extra);

		if (msgr->wip && msg_tnsfd(msgr->wip))
			return tmp_extra->react(server, msgr, err,
					       tmp_extra->extra);

		return POLLING_OKAY;
	case PM_OUTPUT:
		msgr = msgr_get(&server->msgrs, sizeof(fd), &fd);

		if (msgr_send(msgr, server->epoll_fd, 0, err) < 0)
			return tmp_extra->remove(server, fd, err,
						tmp_extra->extra);

		return POLLING_OKAY;
	}

	err_epistle(err, EPISTLE_ERR_TYPE);
	return POLLING_FATAL;
}

Msgr *
msg_server_add(Msg_server *server, int fd, void *extra, Err *err)
{
	return msgr_add(&server->msgrs, sizeof(fd),
			&fd, fd, server->epoll_fd, extra, err);
}

int
msg_server_remove(Msg_server *server, int fd, void **extra, Err *err)
{
	return msgr_remove(&server->msgrs, sizeof(fd), &fd, extra, err);
}

Msgr *
msg_server_get(Msg_server *server, int fd)
{
	return msgr_get(&server->msgrs, sizeof(fd), &fd);
}

int
msg_server(Msg_server *server, Add_msgr add, Remove_msgr remove,
	   React_msg react, Err *err, void *extra)
{
	struct tmp_extra tmp_extra = {
		.server = server, .add = add, .remove = remove,
		.react = react, .extra = extra
	};


	return polling_mode(msg_react, &server->epoll_fd,
			    NULL, s_check, err, &tmp_extra);
}