/*

 *   teamd_zmq.c - Teamd ZeroMQ socket api

 *   Copyright (C) 2013 Jiri Zupka <jzupka@redhat.com>

 *

 *   This library 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 2.1 of the License, or (at your option) any later version.

 *

 *   This library 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 this library; if not, write to the Free Software

 *   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA

 */

#include "config.h"

#ifdef ENABLE_ZMQ

#include <stdbool.h>

#include <stdlib.h>

#include <string.h>

#include <unistd.h>

#include <sys/types.h>

#include <sys/socket.h>

#include <sys/un.h>

#include <errno.h>

#include <ctype.h>

#include <private/misc.h>

#include <private/list.h>

#include <team.h>

#include "teamd.h"

#include "teamd_zmq.h"

#include "teamd_zmq_common.h"

#include "teamd_ctl.h"

#include "teamd_config.h"

struct zmq_ops_priv {

	char *rcv_msg_args;

	void *sock;

};

struct zmq_acc_conn {

	struct list_item list;

	int sock;

};

static int zmq_op_get_args(void *ops_priv, const char *fmt, ...)

{

	va_list ap;

	struct zmq_ops_priv *zmq_ops_priv = ops_priv;

	char **pstr;

	char *str;

	char *rest = zmq_ops_priv->rcv_msg_args;

	int err = 0;

	va_start(ap, fmt);

	while (*fmt) {

		switch (*fmt++) {

		case 's': /* string */

			pstr = va_arg(ap, char **);

			str = teamd_zmq_msg_getline(&rest);

			if (!str) {

				teamd_log_err("Insufficient number of arguments in message.");

				err = -EINVAL;

				goto out;

			}

			*pstr = str;

			break;

		default:

			teamd_log_err("Unknown argument type requested");

			err = -EINVAL;

			goto out;

		}

	}

out:

	va_end(ap);

	return err;

}

static void zmq_custom_send(struct zmq_ops_priv *zmq_ops_priv,

			    char *buf, size_t buflen)

{

	int ret;

	ret = zmq_send(zmq_ops_priv->sock, buf, buflen, 0);

	if (ret == -1)

		teamd_log_warn("zmq: send failed: %s", strerror(errno));

}

static int zmq_op_reply_err(void *ops_priv, const char *err_code,

			      const char *err_msg)

{

	struct zmq_ops_priv *zmq_ops_priv = ops_priv;

	char *strbuf;

	int err;

	err = asprintf(&strbuf, "%s\n%s\n%s\n", TEAMD_ZMQ_REPLY_ERR_PREFIX,

		       err_code, err_msg);

	if (err == -1)

		return -ENOMEM;

	zmq_custom_send(zmq_ops_priv, strbuf, strlen(strbuf));

	free(strbuf);

	return 0;

}

static int zmq_op_reply_succ(void *ops_priv, const char *msg)

{

	struct zmq_ops_priv *zmq_ops_priv = ops_priv;

	char *strbuf;

	int err;

	err = asprintf(&strbuf, "%s\n%s", TEAMD_ZMQ_REPLY_SUCC_PREFIX,

		       msg ? msg : "");

	if (err == -1)

		return -ENOMEM;

	zmq_custom_send(zmq_ops_priv, strbuf, strlen(strbuf));

	free(strbuf);

	return 0;

}

static const struct teamd_ctl_method_ops teamd_zmq_ctl_method_ops = {

	.get_args = zmq_op_get_args,

	.reply_err = zmq_op_reply_err,

	.reply_succ = zmq_op_reply_succ,

};

static int process_rcv_msg(struct teamd_context *ctx, char *rcv_msg)

{

	struct zmq_ops_priv zmq_ops_priv;

	char *str;

	char *rest = rcv_msg;

	str = teamd_zmq_msg_getline(&rest);

	if (!str) {

		teamd_log_dbg(ctx, "zmq: Incomplete message.");

		return 0;

	}

	if (strcmp(TEAMD_ZMQ_REQUEST_PREFIX, str)) {

		teamd_log_dbg(ctx, "zmq: Unsupported message type.");

		return 0;

	}

	str = teamd_zmq_msg_getline(&rest);

	if (!str) {

		teamd_log_dbg(ctx, "zmq: Incomplete message.");

		return 0;

	}

	if (!teamd_ctl_method_exists(str)) {

		teamd_log_dbg(ctx, "zmq: Unknown method \"%s\".", str);

		return 0;

	}

	zmq_ops_priv.sock = ctx->zmq.sock;

	zmq_ops_priv.rcv_msg_args = rest;

	teamd_log_dbg(ctx, "zmq: calling method \"%s\"", str);

	return teamd_ctl_method_call(ctx, str, &teamd_zmq_ctl_method_ops,

				     &zmq_ops_priv);

}

static int callback_zmq(struct teamd_context *ctx, int events, void *priv)

{

	int err = 0;

	int poolmask;

	size_t poolmask_size = sizeof(poolmask);

	err = zmq_getsockopt(ctx->zmq.sock, ZMQ_EVENTS, &poolmask,

			     &poolmask_size);

	if (err == -1)

		return -errno;

	while (poolmask & ZMQ_POLLIN) {

		zmq_msg_t msg;

		zmq_msg_init(&msg;;

		if (zmq_msg_recv(&msg, ctx->zmq.sock, 0) == -1) {

			zmq_msg_close(&msg;;

			return -errno;

		}

		err = process_rcv_msg(ctx, zmq_msg_data(&msg));

		zmq_msg_close(&msg;;

		if (err == -1)

			break;

		err = zmq_getsockopt(ctx->zmq.sock, ZMQ_EVENTS, &poolmask,

				     &poolmask_size);

		if (err == -1)

			return -errno;

	}

	return err;

}

#define ZMQ_MAX_CLIENT_COUNT 10

static int teamd_zmq_sock_open(struct teamd_context *ctx)

{

	int err;

	void *context, *sock;

	int rc;

	const char *addr;

	context = zmq_ctx_new();

	if (!context) {

		teamd_log_err("zmq: Failed to create context.");

		return -errno;

	}

	sock = zmq_socket(context, ZMQ_REP);

	if (!sock) {

		teamd_log_err("zmq: Failed to create socket.");

		return -errno;

	}

	if (ctx->zmq.addr) {

		addr = ctx->zmq.addr;

	} else {

		err = teamd_config_string_get(ctx, &addr, "$.runner.addr");

		if (err) {

			teamd_log_err("zmq: Failed to get address from config.");

			return err;

		}

	}

	rc = zmq_bind(sock, addr);

	if (rc != 0) {

		teamd_log_err("zmq: Failed to bind socket.");

		err = -errno;

		goto close_sock;

	}

	ctx->zmq.context = context;

	ctx->zmq.sock = sock;

	return 0;

close_sock:

	zmq_close(sock);

	zmq_ctx_destroy(context);

	return err;

}

static void teamd_zmq_sock_close(struct teamd_context *ctx)

{

	zmq_close(ctx->zmq.sock);

	zmq_ctx_destroy(ctx->zmq.context);

}

#define ZMQ_CB_NAME "zmq"

int teamd_zmq_init(struct teamd_context *ctx)

{

	int err;

	int fd;

	size_t fd_size;

	if (!ctx->zmq.enabled)

		return 0;

	err = teamd_zmq_sock_open(ctx);

	if (err)

		return err;

	fd_size = sizeof(fd);

	zmq_getsockopt(ctx->zmq.sock, ZMQ_FD, &fd, &fd_size);

	err = teamd_loop_callback_fd_add(ctx, ZMQ_CB_NAME, ctx, callback_zmq,

					 fd, TEAMD_LOOP_FD_EVENT_READ);

	if (err)

		goto sock_close;

	teamd_loop_callback_enable(ctx, ZMQ_CB_NAME, ctx);

	return 0;

sock_close:

	teamd_zmq_sock_close(ctx);

	return err;

}

void teamd_zmq_fini(struct teamd_context *ctx)

{

	if (!ctx->zmq.enabled)

		return;

	teamd_loop_callback_del(ctx, ZMQ_CB_NAME, ctx);

	teamd_zmq_sock_close(ctx);

}

#endif /* ENABLE_ZMQ */