2017-01-19 22:07:52 +01:00
|
|
|
#include "../../core/communication.h"
|
|
|
|
#include "../../core/msg.h"
|
|
|
|
#include "../../core/process.h"
|
|
|
|
#include "../../core/utils.h"
|
|
|
|
#include "../../core/error.h"
|
2016-09-14 21:09:14 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
#include "pubsubd.h"
|
|
|
|
#include "channels.h"
|
2016-09-14 21:09:14 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
#include <sys/socket.h>
|
|
|
|
#include <sys/un.h>
|
|
|
|
#include <unistd.h>
|
2016-09-14 21:09:14 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
void pubsubd_send (const struct array_proc *ap, const struct pubsub_msg * m)
|
2016-09-14 21:09:14 +02:00
|
|
|
{
|
2017-01-19 22:07:52 +01:00
|
|
|
if (ap == NULL) {
|
|
|
|
fprintf (stderr, "pubsubd_send: ap == NULL");
|
2016-09-14 21:09:14 +02:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
if (m == NULL) {
|
|
|
|
fprintf (stderr, "pubsubd_send: m == NULL");
|
2016-09-14 21:09:14 +02:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
char *buf = NULL;
|
|
|
|
size_t msize = 0;
|
|
|
|
pubsub_msg_serialize (m, &buf, &msize);
|
2016-06-07 13:49:23 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
struct msg m_data;
|
|
|
|
memset (&m_data, 0, sizeof (struct msg));
|
|
|
|
msg_format_data (&m_data, buf, msize);
|
2016-06-07 13:49:23 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
int i;
|
|
|
|
for (i = 0; i < ap->size ; i++) {
|
|
|
|
srv_write (ap->tab_proc[i], &m_data);
|
2016-06-07 13:49:23 +02:00
|
|
|
}
|
2017-01-19 22:07:52 +01:00
|
|
|
msg_free (&m_data);
|
2016-06-07 17:44:18 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
if (buf != NULL) {
|
|
|
|
free (buf);
|
2016-09-11 01:47:25 +02:00
|
|
|
}
|
2016-06-05 20:48:13 +02:00
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
// 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);
|
|
|
|
// }
|
2016-09-07 23:46:00 +02:00
|
|
|
|
2017-08-28 00:03:35 +02:00
|
|
|
/**
|
|
|
|
* new connection, once accepted the process is added to the array_proc
|
|
|
|
* structure to be checked periodically for new messages
|
|
|
|
*/
|
2017-01-19 22:07:52 +01:00
|
|
|
void handle_new_connection (struct service *srv, struct array_proc *ap)
|
2016-06-06 02:25:28 +02:00
|
|
|
{
|
2017-01-19 22:07:52 +01:00
|
|
|
struct process *p = malloc(sizeof(struct process));
|
|
|
|
memset(p, 0, sizeof(struct process));
|
2016-06-07 17:44:18 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
if (srv_accept (srv, p) < 0) {
|
|
|
|
handle_error("srv_accept < 0");
|
|
|
|
} else {
|
|
|
|
printf("new connection\n");
|
2016-06-07 17:44:18 +02:00
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
if (add_proc (ap, p) < 0) {
|
|
|
|
handle_error("add_proc < 0");
|
2016-06-07 17:44:18 +02:00
|
|
|
}
|
2016-06-06 02:25:28 +02:00
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
void handle_new_msg (struct channels *chans
|
|
|
|
, struct array_proc *ap, struct array_proc *proc_to_read)
|
2016-09-11 01:47:25 +02:00
|
|
|
{
|
2017-01-19 22:07:52 +01:00
|
|
|
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");
|
2016-09-11 02:18:12 +02:00
|
|
|
}
|
2016-09-11 01:47:25 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
mprint_hexa ("msg received: ", (unsigned char *) m.val, m.valsize);
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
// 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];
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
printf ("proc %d disconnecting\n", p->proc_fd);
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
// TODO: to test, unsubscribe when closing
|
|
|
|
pubsubd_channels_unsubscribe_everywhere (chans, p);
|
2016-06-07 17:44:18 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
// close the connection to the process
|
|
|
|
if (srv_close_proc (p) < 0)
|
|
|
|
handle_error( "srv_close_proc < 0");
|
2016-06-05 20:48:13 +02:00
|
|
|
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
// 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");
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
msg_free (&m);
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
// free process
|
|
|
|
free (p);
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
i--;
|
|
|
|
continue;
|
2016-06-05 20:48:13 +02:00
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
struct pubsub_msg m_data;
|
|
|
|
memset (&m_data, 0, sizeof (struct pubsub_msg));
|
2016-09-11 01:47:25 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
pubsub_msg_unserialize (&m_data, m.val, m.valsize);
|
2016-06-13 09:47:19 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
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]);
|
|
|
|
}
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
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]);
|
|
|
|
}
|
2016-06-07 17:44:18 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
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);
|
|
|
|
}
|
2016-06-07 17:44:18 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
pubsub_msg_free (&m_data);
|
|
|
|
msg_free (&m);
|
2016-06-07 17:44:18 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
/*
|
|
|
|
* main loop
|
|
|
|
*
|
|
|
|
* accept new application connections
|
|
|
|
* read a message and send it back
|
|
|
|
* close a connection if MSG_TYPE_CLOSE received
|
|
|
|
*/
|
2016-06-05 20:48:13 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
void pubsubd_main_loop (struct service *srv, struct channels *chans)
|
2016-06-05 20:48:13 +02:00
|
|
|
{
|
2017-01-19 22:07:52 +01:00
|
|
|
int i, ret = 0;
|
2016-06-07 17:44:18 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
struct array_proc ap;
|
|
|
|
memset(&ap, 0, sizeof(struct array_proc));
|
2016-06-07 11:45:21 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
struct array_proc proc_to_read;
|
|
|
|
memset(&proc_to_read, 0, sizeof(struct array_proc));
|
2016-06-07 11:45:21 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
while(1) {
|
|
|
|
ret = srv_select (&ap, srv, &proc_to_read);
|
2016-09-10 17:16:39 +02:00
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
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);
|
2016-06-07 11:45:21 +02:00
|
|
|
}
|
2017-01-19 22:07:52 +01:00
|
|
|
array_proc_free (&proc_to_read);
|
2016-06-07 11:45:21 +02:00
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
for (i = 0; i < ap.size; i++) {
|
|
|
|
if (srv_close_proc (ap.tab_proc[i]) < 0) {
|
|
|
|
handle_error( "srv_close_proc < 0");
|
|
|
|
}
|
2016-06-08 17:57:05 +02:00
|
|
|
}
|
|
|
|
|
2017-01-19 22:07:52 +01:00
|
|
|
pubsubd_channels_del_all (chans);
|
2016-06-08 17:57:05 +02:00
|
|
|
}
|
|
|
|
|