From eea95e5ee3a22a7ab5dc311b200788ff078ad02b Mon Sep 17 00:00:00 2001
From: Philippe PITTOLI
Date: Wed, 23 Aug 2017 20:57:44 +0200
Subject: [PATCH] dev remote -- draft 1
---
remote/app/Makefile | 25 ++++++
remote/app/remotec.c | 104 +++++++++++++++++++++++
remote/app/remoted.c | 59 +++++++++++++
remote/lib/pubsubd.c | 196 +++++++++++++++++++++++++++++++++++++++++++
remote/lib/pubsubd.h | 15 ++++
5 files changed, 399 insertions(+)
create mode 100644 remote/app/Makefile
create mode 100644 remote/app/remotec.c
create mode 100644 remote/app/remoted.c
create mode 100644 remote/lib/pubsubd.c
create mode 100644 remote/lib/pubsubd.h
diff --git a/remote/app/Makefile b/remote/app/Makefile
new file mode 100644
index 0000000..8ea2a9e
--- /dev/null
+++ b/remote/app/Makefile
@@ -0,0 +1,25 @@
+CC=gcc
+CFLAGS=-Wall -g -Wextra
+LDFLAGS= -pthread
+CFILES=$(wildcard *.c) # CFILES => recompiles everything on a C file change
+EXEC=$(basename $(wildcard *.c))
+SOURCES=$(wildcard ../lib/*.c ../../core/*.c)
+OBJECTS=$(SOURCES:.c=.o)
+TESTS=$(addsuffix .test, $(EXEC))
+
+all: $(SOURCES) $(EXEC)
+
+$(EXEC): $(OBJECTS) $(CFILES)
+ $(CC) $(CFLAGS) $(LDFLAGS) $(OBJECTS) $@.c -o $@.bin
+
+.c.o:
+ $(CC) -c $(CFLAGS) $< -o $@
+
+$(TESTS):
+ valgrind --show-leak-kinds=all --leak-check=full -v --track-origins=yes ./$(basename $@).bin
+
+clean:
+ @-rm $(OBJECTS)
+
+mrproper: clean
+ @-rm *.bin
diff --git a/remote/app/remotec.c b/remote/app/remotec.c
new file mode 100644
index 0000000..c935f42
--- /dev/null
+++ b/remote/app/remotec.c
@@ -0,0 +1,104 @@
+// int main(void) { return 0; }
+
+#include "../../core/error.h"
+#include "../lib/remote.h"
+#include "../lib/remoted.h"
+
+#include
+#include
+#include
+
+void usage (char **argv) {
+ printf ( "usage: %s uri service\n", argv[0]);
+}
+
+void * listener (void *params)
+{
+ int s = 0;
+ s = pthread_setcancelstate (PTHREAD_CANCEL_ENABLE, NULL);
+ if (s != 0) {
+ handle_err ("listener", "pthread_setcancelstate != 0");
+ }
+
+ struct service *srv = NULL;
+ srv = (struct service *) params;
+ if (srv == NULL) {
+ handle_err ("listener", "no service passed");
+ return NULL;
+ }
+
+ // main loop
+ while (1) {
+ struct remote_msg m;
+ memset (&m, 0, sizeof (struct remote_msg));
+
+ remote_msg_recv (srv, &m);
+ printf ("\r\033[31m>\033[00m %s\n", m.data);
+ print_cmd ();
+
+ remote_msg_free (&m);
+ }
+
+ pthread_exit (NULL);
+}
+
+void main_loop (int argc, char **argv, char **env
+ , int index, int version, char *uri, char *service)
+{
+ printf ("connection to remoted: index %d version %d uri %s service %s\n"
+ , index, version, uri, service);
+
+ struct service srv;
+ memset (&srv, 0, sizeof (struct service));
+ remote_connection (argc, argv, env, &srv);
+ printf ("connected\n");
+
+ if (strncmp (cmd, "sub", 3) == 0) {
+ chan_sub (&srv, chan);
+ }
+
+ printf ("main_loop\n");
+
+ struct remote_msg msg;
+ memset (&msg, 0, sizeof (struct remote_msg));
+
+ // msg loop
+ for (;;) {
+ char buf[BUFSIZ];
+ memset (buf, 0, BUFSIZ);
+
+ msg.datalen = /* TODO */0;
+ msg.data = /* TODO */ NULL;
+ memset (msg.data, 0, msg.datalen);
+ strncpy ((char *) msg.data, /* TODO */NULL, msg.datalen);
+ msg.data[msg.datalen -1] = '\0';
+
+ /* TODO */
+ remote_msg_send (&srv, &msg);
+ free (msg.data);
+ msg.data = NULL;
+ msg.datalen = 0;
+ }
+
+ // free everything
+ remote_msg_free (&msg);
+
+ printf ("disconnection...\n");
+ // disconnect from the server
+ remote_disconnect (&srv);
+}
+
+int main(int argc, char **argv, char **env)
+{
+ if (argc != 3) {
+ usage (argv);
+ exit (0);
+ }
+
+ int index = 0;
+ int version = 0;
+
+ main_loop (argc, argv, env, index, version, argv[1], argv[2]);
+
+ return EXIT_SUCCESS;
+}
diff --git a/remote/app/remoted.c b/remote/app/remoted.c
new file mode 100644
index 0000000..085773c
--- /dev/null
+++ b/remote/app/remoted.c
@@ -0,0 +1,59 @@
+#include "../../core/communication.h"
+#include "../../core/process.h"
+#include "../../core/error.h"
+#include "../lib/pubsubd.h"
+#include
+
+#include
+#include
+#include
+#include
+
+// to quit them properly if a signal occurs
+struct service srv;
+struct channels chans;
+
+void handle_signal (int signalnumber)
+{
+ // the application will shut down, and remove the service unix socket
+ if (srv_close (&srv) < 0) {
+ handle_error("srv_close < 0");
+ }
+
+ fprintf (stderr, "received a signal %d\n", signalnumber);
+ exit (EXIT_SUCCESS);
+}
+
+void remoted_init () {}
+
+int
+main(int argc, char **argv, char **env)
+{
+ memset (&srv, 0, sizeof (struct service));
+ srv.index = 0;
+ srv.version = 0;
+
+ signal(SIGHUP, handle_signal);
+ signal(SIGINT, handle_signal);
+ signal(SIGQUIT, handle_signal);
+
+ remoted_init ();
+
+ if (srv_init (argc, argv, env, &srv, REMOTED_SERVICE_NAME) < 0) {
+ handle_error("srv_init < 0");
+ return EXIT_FAILURE;
+ }
+ printf ("Listening on %s.\n", srv.spath);
+
+ printf("MAIN: server created\n" );
+
+ // the service will loop until the end of time, a specific message, a signal
+ pubsubd_main_loop (&srv, &chans);
+
+ // the application will shut down, and remove the service named pipe
+ if (srv_close (&srv) < 0) {
+ handle_error("srv_close < 0");
+ }
+
+ return EXIT_SUCCESS;
+}
diff --git a/remote/lib/pubsubd.c b/remote/lib/pubsubd.c
new file mode 100644
index 0000000..4d21856
--- /dev/null
+++ b/remote/lib/pubsubd.c
@@ -0,0 +1,196 @@
+#include "../../core/communication.h"
+#include "../../core/msg.h"
+#include "../../core/process.h"
+#include "../../core/utils.h"
+#include "../../core/error.h"
+
+#include "pubsubd.h"
+#include "channels.h"
+
+#include
+#include
+#include
+
+void pubsubd_send (const struct array_proc *ap, const struct pubsub_msg * m)
+{
+ if (ap == NULL) {
+ fprintf (stderr, "pubsubd_send: ap == NULL");
+ return;
+ }
+
+ if (m == NULL) {
+ fprintf (stderr, "pubsubd_send: m == NULL");
+ return;
+ }
+
+ char *buf = NULL;
+ size_t msize = 0;
+ pubsub_msg_serialize (m, &buf, &msize);
+
+ struct msg m_data;
+ memset (&m_data, 0, sizeof (struct msg));
+ msg_format_data (&m_data, buf, msize);
+
+ int i;
+ for (i = 0; i < ap->size ; i++) {
+ srv_write (ap->tab_proc[i], &m_data);
+ }
+ msg_free (&m_data);
+
+ if (buf != NULL) {
+ free (buf);
+ }
+}
+
+// void pubsubd_recv (struct process *p, struct pubsub_msg *m)
+// {
+// struct msg m_data;
+// memset (&m_data, 0, sizeof (struct msg));
+//
+// // read the message from the process
+// srv_read (p, &m_data);
+//
+// pubsub_msg_unserialize (m, m_data.val, m_data.valsize);
+//
+// msg_free (&m_data);
+// }
+
+void handle_new_connection (struct service *srv, struct array_proc *ap)
+{
+ struct process *p = malloc(sizeof(struct process));
+ memset(p, 0, sizeof(struct process));
+
+ if (srv_accept (srv, p) < 0) {
+ handle_error("srv_accept < 0");
+ } else {
+ printf("new connection\n");
+ }
+
+ if (add_proc (ap, p) < 0) {
+ handle_error("add_proc < 0");
+ }
+}
+
+void handle_new_msg (struct channels *chans
+ , struct array_proc *ap, struct array_proc *proc_to_read)
+{
+ struct msg m;
+ memset (&m, 0, sizeof (struct msg));
+ int i;
+ for (i = 0; i < proc_to_read->size; i++) {
+ // printf ("loop handle_new_msg\n");
+ if (srv_read (proc_to_read->tab_proc[i], &m) < 0) {
+ handle_error("srv_read < 0");
+ }
+
+ mprint_hexa ("msg received: ", (unsigned char *) m.val, m.valsize);
+
+ // close the process then delete it from the process array
+ if (m.type == MSG_TYPE_CLOSE) {
+ struct process *p = proc_to_read->tab_proc[i];
+
+ printf ("proc %d disconnecting\n", p->proc_fd);
+
+ // TODO: to test, unsubscribe when closing
+ pubsubd_channels_unsubscribe_everywhere (chans, p);
+
+ // close the connection to the process
+ if (srv_close_proc (p) < 0)
+ handle_error( "srv_close_proc < 0");
+
+
+ // remove the process from the processes list
+ if (del_proc (ap, p) < 0)
+ handle_error( "del_proc < 0");
+ if (del_proc (proc_to_read, p) < 0)
+ handle_err( "handle_new_msg", "del_proc < 0");
+
+ msg_free (&m);
+
+ // free process
+ free (p);
+
+ i--;
+ continue;
+ }
+
+ struct pubsub_msg m_data;
+ memset (&m_data, 0, sizeof (struct pubsub_msg));
+
+ pubsub_msg_unserialize (&m_data, m.val, m.valsize);
+
+ if (m_data.type == PUBSUB_MSG_TYPE_SUB) {
+ printf ("proc %d subscribing to %s\n"
+ , proc_to_read->tab_proc[i]->proc_fd
+ , m_data.chan);
+ pubsubd_channels_subscribe (chans
+ , m_data.chan, proc_to_read->tab_proc[i]);
+ }
+
+ if (m_data.type == PUBSUB_MSG_TYPE_UNSUB) {
+ printf ("proc %d unsubscribing to %s\n"
+ , proc_to_read->tab_proc[i]->proc_fd
+ , m_data.chan);
+ pubsubd_channels_unsubscribe (chans
+ , m_data.chan, proc_to_read->tab_proc[i]);
+ }
+
+ if (m_data.type == PUBSUB_MSG_TYPE_PUB) {
+ printf ("proc %d publishing to %s\n"
+ , proc_to_read->tab_proc[i]->proc_fd
+ , m_data.chan);
+ struct channel *chan = pubsubd_channel_search (chans, m_data.chan);
+ if (chan == NULL) {
+ handle_err ("handle_new_msg", "publish on nonexistent channel");
+ msg_free (&m);
+ return ;
+ }
+ pubsubd_send (chan->subs, &m_data);
+ }
+
+ pubsub_msg_free (&m_data);
+ msg_free (&m);
+ }
+}
+
+/*
+ * main loop
+ *
+ * accept new application connections
+ * read a message and send it back
+ * close a connection if MSG_TYPE_CLOSE received
+ */
+
+void pubsubd_main_loop (struct service *srv, struct channels *chans)
+{
+ int i, ret = 0;
+
+ struct array_proc ap;
+ memset(&ap, 0, sizeof(struct array_proc));
+
+ struct array_proc proc_to_read;
+ memset(&proc_to_read, 0, sizeof(struct array_proc));
+
+ while(1) {
+ ret = srv_select (&ap, srv, &proc_to_read);
+
+ if (ret == CONNECTION) {
+ handle_new_connection (srv, &ap);
+ } else if (ret == APPLICATION) {
+ handle_new_msg (chans, &ap, &proc_to_read);
+ } else { // both new connection and new msg from at least one client
+ handle_new_connection (srv, &ap);
+ handle_new_msg (chans, &ap, &proc_to_read);
+ }
+ array_proc_free (&proc_to_read);
+ }
+
+ for (i = 0; i < ap.size; i++) {
+ if (srv_close_proc (ap.tab_proc[i]) < 0) {
+ handle_error( "srv_close_proc < 0");
+ }
+ }
+
+ pubsubd_channels_del_all (chans);
+}
+
diff --git a/remote/lib/pubsubd.h b/remote/lib/pubsubd.h
new file mode 100644
index 0000000..0c548d2
--- /dev/null
+++ b/remote/lib/pubsubd.h
@@ -0,0 +1,15 @@
+#ifndef __PUBSUBD_H__
+#define __PUBSUBD_H__
+
+// #include "../../core/pubsub.h"
+#include "../../core/process.h"
+#include "../../core/msg.h"
+#include "msg.h"
+#include "channels.h"
+
+#define PUBSUBD_SERVICE_NAME "pubsubd"
+
+void pubsubd_main_loop (struct service *srv, struct channels * chans);
+void pubsubd_msg_send (const struct array_proc *ap, const struct pubsub_msg * m);
+
+#endif