Add and improve asynchronous clients

The client program cynadm gets 2 new commands:
acheck and atest that are intended to use
asynchronous test and check.

The client programs cynadm and test-old-cynara
now read as much a possible input before to
process asynchronous answers. This is made to
simulate congestion on asynchronous client.

Change-Id: I1c9dbfe43f150725d530c155135ead8e758897e5
Signed-off-by: José Bollo <jose.bollo@iot.bzh>

2 files changed, 219 insertions, 53 deletions

diff --git a/src/main-cynadm.c b/src/main-cynadm.c
index 4be5bae..a4dd5a8 100644
--- a/src/main-cynadm.c
+++ b/src/main-cynadm.c
@@ -22,8 +22,12 @@
 #include <string.h>
 #include <getopt.h>
 #include <errno.h>
+#include <unistd.h>
+#include <fcntl.h>
 #include <time.h>
 #include <signal.h>
+#include <poll.h>
+#include <sys/epoll.h>
 
 #include "rcyn-client.h"
 #include "rcyn-protocol.h"
@@ -152,7 +156,7 @@ help_check_text[] =
 	"\n"
 	"Command: check client session user permission\n"
 	"\n"
-	"Check authorisation for the given 'client', 'session', 'user', 'permission'.\n"
+	"Check authorization for the given 'client', 'session', 'user', 'permission'.\n"
 	"\n"
 	"Examples:\n"
 	"\n"
@@ -164,11 +168,22 @@ help_check_text[] =
 
 static
 const char
+help_acheck_text[] =
+	"\n"
+	"Command: acheck client session user permission\n"
+	"\n"
+	"Check asynchronously authorization for the given 'client', 'session', 'user', 'permission'.\n"
+	"Same as check but don't wait the answer.\n"
+	"\n"
+;
+
+static
+const char
 help_test_text[] =
 	"\n"
 	"Command: test client session user permission\n"
 	"\n"
-	"Test authorisation for the given 'client', 'session', 'user', 'permission'.\n"
+	"Test authorization for the given 'client', 'session', 'user', 'permission'.\n"
 	"Same as command 'check' except that it doesn't use query agent if it were\n"
 	"needed to avoid asynchronous timely unlimited queries.\n"
 	"\n"
@@ -182,6 +197,17 @@ help_test_text[] =
 
 static
 const char
+help_atest_text[] =
+	"\n"
+	"Command: atest client session user permission\n"
+	"\n"
+	"Test asynchronously authorization for the given 'client', 'session', 'user', 'permission'.\n"
+	"Same as test but don't wait the answer.\n"
+	"\n"
+;
+
+static
+const char
 help_log_text[] =
 	"\n"
 	"Command: log [on|off]\n"
@@ -201,7 +227,7 @@ help_cache_text[] =
 	"\n"
 	"Command: cache client session user permission\n"
 	"\n"
-	"Test cache for authorisation for the given 'client', 'session', 'user', 'permission'.\n"
+	"Test cache for authorization for the given 'client', 'session', 'user', 'permission'.\n"
 	"\n"
 	"Examples:\n"
 	"\n"
@@ -265,8 +291,10 @@ help_expiration_text[] =
 
 static rcyn_t *rcyn;
 static char buffer[4000];
+static int bufill;
 static char *str[40];
 static int nstr;
+static int pending;
 
 rcyn_key_t key;
 rcyn_value_t value;
@@ -495,6 +523,33 @@ int do_check(int ac, char **av, int (*f)(rcyn_t*,const rcyn_key_t*))
 	return uc;
 }
 
+void acheck_cb(void *closure, int status)
+{
+	if (status > 0)
+		fprintf(stdout, "allowed\n");
+	else if (status == 0)
+		fprintf(stdout, "denied\n");
+	else
+		fprintf(stderr, "error %s\n", strerror(-status));
+	pending--;
+}
+
+int do_acheck(int ac, char **av, bool simple)
+{
+	int uc, rc;
+
+	rc = get_csup(ac, av, &uc, NULL);
+	if (rc == 0) {
+		pending++;
+		rc = rcyn_async_check(rcyn, &key, simple, acheck_cb, NULL);
+		if (rc < 0) {
+			fprintf(stderr, "error %s\n", strerror(-rc));
+			pending--;
+		}
+	}
+	return uc;
+}
+
 int do_log(int ac, char **av)
 {
 	int uc, rc;
@@ -527,8 +582,12 @@ int do_help(int ac, char **av)
 		fprintf(stdout, "%s", help_drop_text);
 	else if (ac > 1 && !strcmp(av[1], "check"))
 		fprintf(stdout, "%s", help_check_text);
+	else if (ac > 1 && !strcmp(av[1], "acheck"))
+		fprintf(stdout, "%s", help_acheck_text);
 	else if (ac > 1 && !strcmp(av[1], "test"))
 		fprintf(stdout, "%s", help_test_text);
+	else if (ac > 1 && !strcmp(av[1], "atest"))
+		fprintf(stdout, "%s", help_atest_text);
 	else if (ac > 1 && !strcmp(av[1], "cache"))
 		fprintf(stdout, "%s", help_cache_text);
 	else if (ac > 1 && !strcmp(av[1], "clear"))
@@ -565,9 +624,15 @@ int do_any(int ac, char **av)
 	if (!strcmp(av[0], "check"))
 		return do_check(ac, av, rcyn_check);
 
+	if (!strcmp(av[0], "acheck"))
+		return do_acheck(ac, av, 0);
+
 	if (!strcmp(av[0], "test"))
 		return do_check(ac, av, rcyn_test);
 
+	if (!strcmp(av[0], "atest"))
+		return do_acheck(ac, av, 1);
+
 	if (!strcmp(av[0], "cache"))
 		return do_check(ac, av, rcyn_cache_check);
 
@@ -602,6 +667,22 @@ void do_all(int ac, char **av)
 	}
 }
 
+int async_ctl(void *closure, int op, int fd, uint32_t events)
+{
+	int *pfd = closure;
+
+	switch(op) {
+	case EPOLL_CTL_ADD:
+	case EPOLL_CTL_MOD:
+		*pfd = fd;
+		break;
+	case EPOLL_CTL_DEL:
+		*pfd = -1;
+		break;
+	}
+	return 0;
+}
+
 int main(int ac, char **av)
 {
 	int opt;
@@ -610,6 +691,8 @@ int main(int ac, char **av)
 	int version = 0;
 	int error = 0;
 	const char *socket = NULL;
+	struct pollfd fds[2];
+	char *p;
 
 	/* scan arguments */
 	for (;;) {
@@ -649,7 +732,14 @@ int main(int ac, char **av)
 	signal(SIGPIPE, SIG_IGN); /* avoid SIGPIPE! */
 	rc = rcyn_open(&rcyn, rcyn_Admin, 5000, socket);
 	if (rc < 0) {
-		fprintf(stderr, "initialisation failed: %s\n", strerror(-rc));
+		fprintf(stderr, "initialization failed: %s\n", strerror(-rc));
+		return 1;
+	}
+
+	fds[1].fd = -1;
+	rc = rcyn_async_setup(rcyn, async_ctl, &fds[1].fd);
+	if (rc < 0) {
+		fprintf(stderr, "asynchronous setup failed: %s\n", strerror(-rc));
 		return 1;
 	}
 
@@ -658,20 +748,55 @@ int main(int ac, char **av)
 		return 0;
 	}
 
+	fcntl(0, F_SETFL, O_NONBLOCK);
+	bufill = 0;
+	fds[0].fd = 0;
+	fds[0].events = fds[1].events = POLLIN;
 	for(;;) {
-		if (!fgets(buffer, sizeof buffer, stdin))
-			break;
-
-		str[nstr = 0] = strtok(buffer, " \t\n");
-		while(str[nstr])
-			str[++nstr] = strtok(NULL, " \t\n");
-
-		ac = 0;
-		while(ac < nstr) {
-			rc = do_any(nstr - ac, &str[ac]);
-			if (rc <= 0)
-				exit(1);
-			ac += rc;
+		rc = poll(fds, 2, -1);
+		if (fds[0].revents & POLLIN) {
+			rc = (int)sizeof buffer - bufill;
+			rc = (int)read(0, buffer, rc);
+			if (rc == 0)
+				break;
+			if (rc > 0) {
+				bufill += rc;
+				while((p = memchr(buffer, '\n', bufill))) {
+					/* process one line */
+					*p++ = 0;
+					str[nstr = 0] = strtok(buffer, " \t");
+					while(str[nstr])
+						str[++nstr] = strtok(NULL, " \t");
+					ac = 0;
+					while(ac < nstr) {
+						rc = do_any(nstr - ac, &str[ac]);
+						if (rc <= 0)
+							exit(1);
+						ac += rc;
+					}
+					bufill -= (int)(p - buffer);
+					if (!bufill)
+						break;
+					memmove(buffer, p, bufill);
+				}
+			}
+		}
+		if (fds[0].revents & POLLHUP) {
+			if (!pending)
+				break;
+			fds[0].fd = -1;
+		}
+		if (fds[1].revents & POLLIN) {
+			rc = rcyn_async_process(rcyn);
+			if (rc < 0)
+				fprintf(stderr, "asynchronous processing failed: %s\n", strerror(-rc));
+			if (fds[0].fd < 0 && !pending)
+				break;
+		}
+		if (fds[1].revents & POLLHUP) {
+			if (fds[0].fd < 0)
+				break;
+			fds[1].fd = -1;
 		}
 	}
 	return 0;
diff --git a/src/main-test-old-cynara.c b/src/main-test-old-cynara.c
index 9f9ea1b..79e386c 100644
--- a/src/main-test-old-cynara.c
+++ b/src/main-test-old-cynara.c
@@ -19,6 +19,8 @@
 #include <stdbool.h>
 #include <stdio.h>
 #include <stdlib.h>
+#include <unistd.h>
+#include <fcntl.h>
 #include <string.h>
 #include <sys/epoll.h>
 
@@ -41,9 +43,12 @@ struct cynara_async *aclient;
 struct cynara_configuration *conf;
 struct cynara *client;
 char buffer[4000];
+int bufill;
 char *str[40];
 int nstr;
 int pollfd;
+int pending;
+int ending;
 
 #define BUCKET "BUCK"
 
@@ -167,10 +172,14 @@ void asyncb(cynara_check_id check_id, cynara_async_call_cause cause,
            int response, void *user_response_data)
 {
 	printf("RECEIVE %d %d\n", cause, response);
+	pending--;
+	if (ending && !pending)
+		exit(0);
 }
 
 void asy_check(char *cli, char *ses, char *usr, char *perm, int simple)
 {
+	pending++;
 	if (simple)
 		CKEX(cynara_async_create_simple_request(aclient, cli, ses, usr, perm, NULL, asyncb, NULL));
 	else
@@ -202,9 +211,42 @@ void asyncstscb(int old_fd, int new_fd, cynara_async_status status, void *data)
 	}
 }
 
+int action()
+{
+	if (is("admin", "listall", 0))
+		adm_list("#", "#", "#");
+	else if (is("admin", "list", 3))
+		adm_list(str[2], str[3], str[4]);
+	else if (is("admin", "check", 3))
+		adm_check(str[2], str[3], str[4]);
+	else if (is("admin", "set", 4))
+		adm_set();
+	else if (is("admin", "erase", 3))
+		adm_erase(str[2], str[3], str[4]);
+	else if (is("admin", "desc", 0))
+		adm_desc();
+	else if (is("async", "cache", 4))
+		asy_cache(str[2], str[3], str[4], str[5]);
+	else if (is("async", "check", 4))
+		asy_check(str[2], str[3], str[4], str[5], 0);
+	else if (is("async", "test", 4))
+		asy_check(str[2], str[3], str[4], str[5], 1);
+	else if (is("sync", "check", 4))
+		syn_check(str[2], str[3], str[4], str[5], 0);
+	else if (is("sync", "test", 4))
+		syn_check(str[2], str[3], str[4], str[5], 1);
+	else if (nstr > 0 && !strcmp(str[0], "exit"))
+		return 1;
+	else if (nstr > 0 && str[0][0] != '#')
+		printf("ERROR bad input\n");
+	return 0;
+}
+
 int main(int ac, char **av)
 {
 	struct epoll_event ev;
+	char *p;
+	int rc;
 
 	pollfd = epoll_create(10);
 	ev.data.fd = 0;
@@ -223,49 +265,48 @@ int main(int ac, char **av)
 	CKEX(cynara_initialize(&client, conf));
 	cynara_configuration_destroy(conf);
 
+	fcntl(0, F_SETFL, O_NONBLOCK);
+	bufill = 0;
 	for(;;) {
 		epoll_wait(pollfd, &ev, 1, -1);
 
-		if (ev.data.fd) {
+		if (ev.data.fd == 0) {
+			if (ev.events & EPOLLIN) {
+				rc = (int)sizeof buffer - bufill;
+				rc = (int)read(0, buffer, rc);
+				if (rc == 0)
+					break;
+				if (rc > 0) {
+					bufill += rc;
+					while((p = memchr(buffer, '\n', bufill))) {
+						/* process one line */
+						*p++ = 0;
+						str[nstr = 0] = strtok(buffer, " \t");
+						while(str[nstr])
+							str[++nstr] = strtok(NULL, " \t");
+						if (action())
+							goto terminate;
+						/* next line if any */
+						bufill -= (int)(p - buffer);
+						if (!bufill)
+							break;
+						memmove(buffer, p, bufill);
+					}
+				}
+			}
+			if (ev.events & EPOLLHUP) {
+				if (!pending)
+					break;
+				epoll_ctl(pollfd, EPOLL_CTL_DEL, 0, &ev);
+				ending = 1;
+			}
+		}
+		else  {
 			cynara_async_process(aclient);
-			continue;
 		}
 
-		if (!fgets(buffer, sizeof buffer, stdin))
-			break;
-
-		str[nstr = 0] = strtok(buffer, " \t\n");
-		while(str[nstr])
-			str[++nstr] = strtok(NULL, " \t\n");
-
-		if (is("admin", "listall", 0))
-			adm_list("#", "#", "#");
-		else if (is("admin", "list", 3))
-			adm_list(str[2], str[3], str[4]);
-		else if (is("admin", "check", 3))
-			adm_check(str[2], str[3], str[4]);
-		else if (is("admin", "set", 4))
-			adm_set();
-		else if (is("admin", "erase", 3))
-			adm_erase(str[2], str[3], str[4]);
-		else if (is("admin", "desc", 0))
-			adm_desc();
-		else if (is("async", "cache", 4))
-			asy_cache(str[2], str[3], str[4], str[5]);
-		else if (is("async", "check", 4))
-			asy_check(str[2], str[3], str[4], str[5], 0);
-		else if (is("async", "test", 4))
-			asy_check(str[2], str[3], str[4], str[5], 1);
-		else if (is("sync", "check", 4))
-			syn_check(str[2], str[3], str[4], str[5], 0);
-		else if (is("sync", "test", 4))
-			syn_check(str[2], str[3], str[4], str[5], 1);
-		else if (nstr > 0 && !strcmp(str[0], "exit"))
-			break;
-		else if (nstr > 0 && str[0][0] != '#')
-			printf("ERROR bad input\n");
 	}
-
+terminate:
 	cynara_finish(client);
 	cynara_async_finish(aclient);
 	cynara_admin_finish(admin);