Baguette
/
libipc-old
3
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

cleaning names

more_to_read
Philippe PITTOLI 2018-10-04 01:54:12 +02:00
parent 21434c600e
commit e48b5f2ce4
18 changed files with 142 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
core-test/app/array_proc.c
View File

@ -5,8 +5,8 @@
int main()
{
int ret;
struct ipc_client_array tab_proc;
memset(&tab_proc, 0, sizeof(struct ipc_client_array));
struct ipc_client_array clients;
memset(&clients, 0, sizeof(struct ipc_client_array));
struct ipc_client client_tab[5];
memset(&client_tab, 0, sizeof(struct ipc_client) * 5);
@ -14,21 +14,21 @@ int main()
int i;
for (i = 0; i < 5; i++) {
client_tab[i].proc_fd = i;
ret = ipc_client_add(&tab_proc, &client_tab[i]);
ret = ipc_client_add(&clients, &client_tab[i]);
if (ret == -1) {
printf("erreur realloc\n");
}
}
ipc_client_array_print(&tab_proc);
ipc_client_array_print(&clients);
ret = ipc_client_del(&tab_proc, &client_tab[2]);
ret = ipc_client_del(&clients, &client_tab[2]);
if(ret < 0) {
printf("erreur %d\n", ret );
}
ipc_client_array_print(&tab_proc);
ipc_client_array_print(&clients);
ipc_client_array_free (&tab_proc);
ipc_client_array_free (&clients);
return 0;
}

2
core-test/app/communication-client.c
View File

@ -42,7 +42,7 @@ int main (int argc, char *argv[], char *env[])
return EXIT_FAILURE;
}
printf ("msg recv: %s\n", m.val);
printf ("msg recv: %s\n", m.payload);
ipc_message_free (&m);
if (application_close (&srv) < 0) {

6
core-test/app/communication-server.c
View File

@ -39,7 +39,7 @@ int main (int argc, char *argv[], char *env[])
return EXIT_FAILURE;
}
printf ("msg recv: %s\n", m.val);
printf ("msg recv: %s\n", m.payload);
if (server_write (&p, &m) < 0) {
handle_err("main", "server_write < 0");
@ -60,8 +60,8 @@ int main (int argc, char *argv[], char *env[])
}
ipc_message_free (&m);
if (server_close_proc (&p) < 0) {
handle_err("main", "server_close_proc < 0");
if (server_close_client (&p) < 0) {
handle_err("main", "server_close_client < 0");
return EXIT_FAILURE;
}

46
core/client.c
View File

@ -29,45 +29,45 @@ void ipc_server_client_gen (struct ipc_client *p
p->index = index;
}
int ipc_client_add (struct ipc_client_array *aproc, struct ipc_client *p)
int ipc_client_add (struct ipc_client_array *clients, struct ipc_client *p)
{
assert(aproc != NULL);
assert(clients != NULL);
assert(p != NULL);
aproc->size++;
aproc->tab_proc = realloc(aproc->tab_proc
, sizeof(struct ipc_client) * aproc->size);
clients->size++;
clients->clients = realloc(clients->clients
, sizeof(struct ipc_client) * clients->size);
if (aproc->tab_proc == NULL) {
if (clients->clients == NULL) {
return -1;
}
aproc->tab_proc[aproc->size - 1] = p;
clients->clients[clients->size - 1] = p;
return 0;
}
int ipc_client_del (struct ipc_client_array *aproc, struct ipc_client *p)
int ipc_client_del (struct ipc_client_array *clients, struct ipc_client *p)
{
assert(aproc != NULL);
assert(clients != NULL);
assert(p != NULL);
if (aproc->tab_proc == NULL) {
if (clients->clients == NULL) {
return -1;
}
int i;
for (i = 0; i < aproc->size; i++) {
if (aproc->tab_proc[i] == p) {
for (i = 0; i < clients->size; i++) {
if (clients->clients[i] == p) {
aproc->tab_proc[i] = aproc->tab_proc[aproc->size-1];
aproc->size--;
if (aproc->size == 0) {
ipc_client_array_free (aproc);
clients->clients[i] = clients->clients[clients->size-1];
clients->size--;
if (clients->size == 0) {
ipc_client_array_free (clients);
}
else {
aproc->tab_proc = realloc(aproc->tab_proc
, sizeof(struct ipc_client) * aproc->size);
clients->clients = realloc(clients->clients
, sizeof(struct ipc_client) * clients->size);
if (aproc->tab_proc == NULL) {
if (clients->clients == NULL) {
return -2;
}
}
@ -91,15 +91,15 @@ void ipc_client_array_print (struct ipc_client_array *ap)
int i;
for (i = 0; i < ap->size; i++) {
printf("%d : ", i);
client_print(ap->tab_proc[i]);
client_print(ap->clients[i]);
}
}
void ipc_client_array_free (struct ipc_client_array *ap)
{
if (ap->tab_proc != NULL) {
free (ap->tab_proc);
ap->tab_proc = NULL;
if (ap->clients != NULL) {
free (ap->clients);
ap->clients = NULL;
}
ap->size = 0;
}

4
core/client.h
View File

@ -8,12 +8,12 @@ struct ipc_client {
};
struct ipc_client_array {
struct ipc_client **tab_proc;
struct ipc_client **clients;
int size;
};
int ipc_client_add (struct ipc_client_array *, struct ipc_client *);
int ipc_client_del (struct ipc_client_array *aproc, struct ipc_client *p);
int ipc_client_del (struct ipc_client_array *clients, struct ipc_client *p);
void ipc_client_array_print (struct ipc_client_array *);
void ipc_client_array_free (struct ipc_client_array *);

26
core/communication.c
View File

@ -65,14 +65,14 @@ int ipc_server_close (struct ipc_service *srv)
return usock_remove (srv->spath);
}
int ipc_server_close_proc (struct ipc_client *p)
int ipc_server_close_client (struct ipc_client *p)
{
// struct ipc_message m_ack_dis;
// memset (&m_ack_dis, 0, sizeof (struct ipc_message));
// m_ack_dis.type = MSG_TYPE_CLOSE;
// if (ipc_message_write (p->proc_fd, &m_ack_dis) < 0) {
// handle_err ("server_close_proc", "ipc_message_write < 0");
// handle_err ("server_close_client", "ipc_message_write < 0");
// }
return usock_close (p->proc_fd);
@ -128,7 +128,7 @@ int ipc_application_connection (int argc, char **argv, char **env
ipc_application_read (srv, &m_ack);
assert (m_ack.type == MSG_TYPE_ACK);
assert (m_ack.valsize == 0);
assert (m_ack.length == 0);
ipc_message_free (&m_ack);
return 0;
@ -166,8 +166,8 @@ static int getMaxFd(struct ipc_client_array *ap)
int max = 0;
for (i = 0; i < ap->size; i++ ) {
if (ap->tab_proc[i]->proc_fd > max) {
max = ap->tab_proc[i]->proc_fd;
if (ap->clients[i]->proc_fd > max) {
max = ap->clients[i]->proc_fd;
}
}
@ -188,13 +188,13 @@ static int getMaxFd(struct ipc_client_array *ap)
*/
int ipc_server_select (struct ipc_client_array *ap, struct ipc_service *srv
, struct ipc_client_array *proc)
, struct ipc_client_array *client)
{
assert (ap != NULL);
assert (proc != NULL);
assert (client != NULL);
// delete previous read client array
ipc_client_array_free (proc);
ipc_client_array_free (client);
int i, j;
/* master file descriptor list */
@ -213,7 +213,7 @@ int ipc_server_select (struct ipc_client_array *ap, struct ipc_service *srv
FD_SET(listener, &master);
for (i=0; i < ap->size; i++) {
FD_SET(ap->tab_proc[i]->proc_fd, &master);
FD_SET(ap->clients[i]->proc_fd, &master);
}
/* keep track of the biggest file descriptor */
@ -238,8 +238,8 @@ int ipc_server_select (struct ipc_client_array *ap, struct ipc_service *srv
} else {
for(j = 0; j < ap->size; j++) {
// printf ("loop ipc_server_select inner inner loop\n");
if(i == ap->tab_proc[j]->proc_fd ) {
ipc_client_add (proc, ap->tab_proc[j]);
if(i == ap->clients[j]->proc_fd ) {
ipc_client_add (client, ap->clients[j]);
}
}
}
@ -247,9 +247,9 @@ int ipc_server_select (struct ipc_client_array *ap, struct ipc_service *srv
}
} while (0);
if (proc->size > 0 && is_listener)
if (client->size > 0 && is_listener)
return CON_APP;
if (proc->size > 0)
if (client->size > 0)
return APPLICATION;
return CONNECTION;
}

2
core/communication.h
View File

@ -37,7 +37,7 @@ struct ipc_service {
int ipc_server_init (int argc, char **argv, char **env
, struct ipc_service *srv, const char *sname);
int ipc_server_close (struct ipc_service *srv);
int ipc_server_close_proc (struct ipc_client *p);
int ipc_server_close_client (struct ipc_client *p);
int ipc_server_accept (struct ipc_service *srv, struct ipc_client *p);
int ipc_server_read (const struct ipc_client *, struct ipc_message *m);

64
core/msg.c
View File

@ -7,7 +7,7 @@
void ipc_message_print (const struct ipc_message *m)
{
assert (m != NULL);
printf ("msg: type %d len %d\n", m->type, m->valsize);
printf ("msg: type %d len %d\n", m->type, m->length);
}
int ipc_message_format_read (struct ipc_message *m, const char *buf, size_t msize)
@ -20,18 +20,18 @@ int ipc_message_format_read (struct ipc_message *m, const char *buf, size_t msiz
return -1;
m->type = buf[0];
memcpy (&m->valsize, buf+1, 2);
memcpy (&m->length, buf+1, 2);
assert (m->valsize <= BUFSIZ -3);
// printf ("type %d : msize = %ld, valsize = %d\n", m->type, msize, m->valsize);
assert (m->valsize == msize - 3);
assert (m->length <= BUFSIZ -3);
// printf ("type %d : msize = %ld, length = %d\n", m->type, msize, m->length);
assert (m->length == msize - 3);
if (m->val != NULL)
free (m->val), m->val = NULL;
if (m->payload != NULL)
free (m->payload), m->payload = NULL;
if (m->val == NULL && m->valsize > 0) {
m->val = malloc (m->valsize);
memcpy (m->val, buf+3, m->valsize);
if (m->payload == NULL && m->length > 0) {
m->payload = malloc (m->length);
memcpy (m->payload, buf+3, m->length);
}
return 0;
@ -42,7 +42,7 @@ int ipc_message_format_write (const struct ipc_message *m, char **buf, size_t *m
assert (m != NULL);
assert (buf != NULL);
assert (msize != NULL);
assert (m->valsize <= BUFSIZ -3);
assert (m->length <= BUFSIZ -3);
if (m == NULL)
return -1;
@ -54,16 +54,16 @@ int ipc_message_format_write (const struct ipc_message *m, char **buf, size_t *m
return -3;
if (*buf == NULL) {
*buf = malloc (3 + m->valsize);
*buf = malloc (3 + m->length);
}
char *buffer = *buf;
buffer[0] = m->type;
memcpy (buffer + 1, &m->valsize, 2);
memcpy (buffer + 3, m->val, m->valsize);
memcpy (buffer + 1, &m->length, 2);
memcpy (buffer + 3, m->payload, m->length);
*msize = 3 + m->valsize;
*msize = 3 + m->length;
return 0;
}
@ -111,40 +111,40 @@ int ipc_message_write (int fd, const struct ipc_message *m)
// MSG FORMAT
int ipc_message_format (struct ipc_message *m, char type, const char *val, size_t valsize)
int ipc_message_format (struct ipc_message *m, char type, const char *payload, size_t length)
{
assert (m != NULL);
assert (valsize + 3 <= BUFSIZ);
assert ((valsize == 0 && val == NULL) || (valsize > 0 && val != NULL));
assert (length + 3 <= BUFSIZ);
assert ((length == 0 && payload == NULL) || (length > 0 && payload != NULL));
if (valsize + 3 > BUFSIZ) {
if (length + 3 > BUFSIZ) {
handle_err ("msg_format_con", "msgsize > BUFSIZ");
return -1;
}
m->type = type;
m->valsize = (short) valsize;
m->length = (short) length;
if (m->val == NULL)
m->val = malloc (valsize);
if (m->payload == NULL)
m->payload = malloc (length);
memcpy (m->val, val, valsize);
memcpy (m->payload, payload, length);
return 0;
}
int ipc_message_format_con (struct ipc_message *m, const char *val, size_t valsize)
int ipc_message_format_con (struct ipc_message *m, const char *payload, size_t length)
{
return ipc_message_format (m, MSG_TYPE_CON, val, valsize);
return ipc_message_format (m, MSG_TYPE_CON, payload, length);
}
int ipc_message_format_data (struct ipc_message *m, const char *val, size_t valsize)
int ipc_message_format_data (struct ipc_message *m, const char *payload, size_t length)
{
return ipc_message_format (m, MSG_TYPE_DATA, val, valsize);
return ipc_message_format (m, MSG_TYPE_DATA, payload, length);
}
int ipc_message_format_ack (struct ipc_message *m, const char *val, size_t valsize)
int ipc_message_format_ack (struct ipc_message *m, const char *payload, size_t length)
{
return ipc_message_format (m, MSG_TYPE_ACK, val, valsize);
return ipc_message_format (m, MSG_TYPE_ACK, payload, length);
}
int ipc_message_free (struct ipc_message *m)
@ -154,10 +154,10 @@ int ipc_message_free (struct ipc_message *m)
if (m == NULL)
return -1;
if (m->val != NULL)
free (m->val), m->val = NULL;
if (m->payload != NULL)
free (m->payload), m->payload = NULL;
m->valsize = 0;
m->length = 0;
return 0;
}

10
core/msg.h
View File

@ -13,8 +13,8 @@
struct ipc_message {
char type;
unsigned short valsize;
char *val;
unsigned short length;
char *payload;
};
// used to create msg structure from buffer
@ -27,9 +27,9 @@ int ipc_message_read (int fd, struct ipc_message *m);
// write a structure msg to fd
int ipc_message_write (int fd, const struct ipc_message *m);
int ipc_message_format_con (struct ipc_message *m, const char *val, size_t valsize);
int ipc_message_format_data (struct ipc_message *m, const char *val, size_t valsize);
int ipc_message_format_ack (struct ipc_message *m, const char *val, size_t valsize);
int ipc_message_format_con (struct ipc_message *m, const char *payload, size_t length);
int ipc_message_format_data (struct ipc_message *m, const char *payload, size_t length);
int ipc_message_format_ack (struct ipc_message *m, const char *payload, size_t length);
int ipc_message_free (struct ipc_message *m);
void ipc_message_print (const struct ipc_message *m);

6
core/utils.c
View File

@ -1,8 +1,8 @@
#include "utils.h"
void print_hexa (const char *prefix, unsigned char *val, size_t size)
void print_hexa (const char *prefix, unsigned char *payload, size_t size)
{
if (! val)
if (! payload)
return ;
size_t i;
@ -10,7 +10,7 @@ void print_hexa (const char *prefix, unsigned char *val, size_t size)
{
if(! (i % 4))
printf("\n%s (%ld) ", prefix, size);
printf("%2x ", val[i]);
printf("%2x ", payload[i]);
}
printf("\n");
}

2
core/utils.h
View File

@ -5,7 +5,7 @@
#include <stdlib.h>
#include <string.h>
void print_hexa (const char *prefix, unsigned char *val, size_t size);
void print_hexa (const char *prefix, unsigned char *payload, size_t size);
void mprint_hexa (char *prefix, unsigned char *buf, size_t length);
#endif

4
man/communication.h.3
View File

@ -13,7 +13,7 @@ communication.h \- all functions explained
.BI "int ipc_server_write (const struct ipc_client *" p ", const struct ipc_message *" message );
.sp
.BI "int ipc_server_close (struct ipc_service *" srv );
.BI "int ipc_server_close_proc (struct ipc_client *" p );
.BI "int ipc_server_close_client (struct ipc_client *" p );
.BI "int ipc_server_select (struct ipc_client_array *" fds ", struct ipc_service *" srv ", struct ipc_client_array *" readfds );
.BI "int ipc_application_connection (int " argc ", char **" argv ", char **" env ", struct ipc_service *" srv
@ -55,7 +55,7 @@ and
functions take respectively a message to read from, and a message to write to a client.
.PP
The
.BR ipc_server_close_proc ()
.BR ipc_server_close_client ()
and
.BR ipc_server_close ()
functions terminate respectively a client (closing its unix socket) and the service (closing and removing its named unix socket).

4
pong/app/pong.c
View File

@ -42,7 +42,7 @@ void non_interactive (int argc, char *argv[], char *env[])
exit (EXIT_FAILURE);
}
printf ("msg recv: %s\n", m.val);
printf ("msg recv: %s\n", m.payload);
ipc_message_free (&m);
if (ipc_application_close (&srv) < 0) {
@ -96,7 +96,7 @@ void interactive (int argc, char *argv[], char *env[])
exit (EXIT_FAILURE);
}
printf ("msg recv: %s", m.val);
printf ("msg recv: %s", m.payload);
ipc_message_free (&m);
}

18
pong/app/pongd.c
View File

@ -36,7 +36,7 @@ void handle_new_msg (struct ipc_client_array *ap, struct ipc_client_array *proc_
int i;
for (i = 0; i < proc_to_read->size; i++) {
// printf ("loop handle_new_msg\n");
if (ipc_server_read (proc_to_read->tab_proc[i], &m) < 0) {
if (ipc_server_read (proc_to_read->clients[i], &m) < 0) {
handle_error("server_read < 0");
}
@ -45,18 +45,18 @@ void handle_new_msg (struct ipc_client_array *ap, struct ipc_client_array *proc_
cpt--;
printf ("disconnection => %d client(s) remaining\n", cpt);
if (ipc_server_close_proc (proc_to_read->tab_proc[i]) < 0)
handle_err( "handle_new_msg", "server_close_proc < 0");
if (ipc_client_del (ap, proc_to_read->tab_proc[i]) < 0)
if (ipc_server_close_client (proc_to_read->clients[i]) < 0)
handle_err( "handle_new_msg", "server_close_client < 0");
if (ipc_client_del (ap, proc_to_read->clients[i]) < 0)
handle_err( "handle_new_msg", "ipc_client_del < 0");
if (ipc_client_del (proc_to_read, proc_to_read->tab_proc[i]) < 0)
if (ipc_client_del (proc_to_read, proc_to_read->clients[i]) < 0)
handle_err( "handle_new_msg", "ipc_client_del < 0");
i--;
continue;
}
printf ("new message : %s", m.val);
if (ipc_server_write (proc_to_read->tab_proc[i], &m) < 0) {
printf ("new message : %s", m.payload);
if (ipc_server_write (proc_to_read->clients[i], &m) < 0) {
handle_err( "handle_new_msg", "server_write < 0");
}
}
@ -98,8 +98,8 @@ void main_loop (struct ipc_service *srv)
}
for (i = 0; i < ap.size; i++) {
if (ipc_server_close_proc (ap.tab_proc[i]) < 0) {
handle_error( "server_close_proc < 0");
if (ipc_server_close_client (ap.clients[i]) < 0) {
handle_error( "server_close_client < 0");
}
}
}

2
pubsub/lib/pubsub.c
View File

@ -93,7 +93,7 @@ int pubsub_message_recv (struct ipc_service *srv, struct pubsub_msg *m)
memset (&m_recv, 0, sizeof (struct ipc_message));
ipc_application_read (srv, &m_recv);
pubsub_message_unserialize (m, m_recv.val, m_recv.valsize);
pubsub_message_unserialize (m, m_recv.payload, m_recv.length);
ipc_message_free (&m_recv);

38
pubsub/lib/pubsubd.c
View File

@ -33,7 +33,7 @@ void pubsubd_send (const struct ipc_client_array *ap, const struct pubsub_msg *
int i;
for (i = 0; i < ap->size ; i++) {
ipc_server_write (ap->tab_proc[i], &m_data);
ipc_server_write (ap->clients[i], &m_data);
}
ipc_message_free (&m_data);
@ -50,7 +50,7 @@ void pubsubd_send (const struct ipc_client_array *ap, const struct pubsub_msg *
// // read the message from the client
// ipc_server_read (p, &m_data);
//
// pubsub_message_unserialize (m, m_data.val, m_data.valsize);
// pubsub_message_unserialize (m, m_data.payload, m_data.length);
//
// ipc_message_free (&m_data);
// }
@ -83,24 +83,24 @@ void handle_new_msg (struct channels *chans
int i;
for (i = 0; i < proc_to_read->size; i++) {
// printf ("loop handle_new_msg\n");
if (ipc_server_read (proc_to_read->tab_proc[i], &m) < 0) {
if (ipc_server_read (proc_to_read->clients[i], &m) < 0) {
handle_error("server_read < 0");
}
mprint_hexa ("msg received: ", (unsigned char *) m.val, m.valsize);
mprint_hexa ("msg received: ", (unsigned char *) m.payload, m.length);
// close the client then delete it from the client array
if (m.type == MSG_TYPE_CLOSE) {
struct ipc_client *p = proc_to_read->tab_proc[i];
struct ipc_client *p = proc_to_read->clients[i];
printf ("proc %d disconnecting\n", p->proc_fd);
printf ("client %d disconnecting\n", p->proc_fd);
// TODO: to test, unsubscribe when closing
pubsubd_channels_unsubscribe_everywhere (chans, p);
// close the connection to the client
if (ipc_server_close_proc (p) < 0)
handle_error( "server_close_proc < 0");
if (ipc_server_close_client (p) < 0)
handle_error( "server_close_client < 0");
// remove the client from the clientes list
@ -121,27 +121,27 @@ void handle_new_msg (struct channels *chans
struct pubsub_msg m_data;
memset (&m_data, 0, sizeof (struct pubsub_msg));
pubsub_message_unserialize (&m_data, m.val, m.valsize);
pubsub_message_unserialize (&m_data, m.payload, m.length);
if (m_data.type == PUBSUB_MSG_TYPE_SUB) {
printf ("proc %d subscribing to %s\n"
, proc_to_read->tab_proc[i]->proc_fd
printf ("client %d subscribing to %s\n"
, proc_to_read->clients[i]->proc_fd
, m_data.chan);
pubsubd_channels_subscribe (chans
, m_data.chan, proc_to_read->tab_proc[i]);
, m_data.chan, proc_to_read->clients[i]);
}
if (m_data.type == PUBSUB_MSG_TYPE_UNSUB) {
printf ("proc %d unsubscribing to %s\n"
, proc_to_read->tab_proc[i]->proc_fd
printf ("client %d unsubscribing to %s\n"
, proc_to_read->clients[i]->proc_fd
, m_data.chan);
pubsubd_channels_unsubscribe (chans
, m_data.chan, proc_to_read->tab_proc[i]);
, m_data.chan, proc_to_read->clients[i]);
}
if (m_data.type == PUBSUB_MSG_TYPE_PUB) {
printf ("proc %d publishing to %s\n"
, proc_to_read->tab_proc[i]->proc_fd
printf ("client %d publishing to %s\n"
, proc_to_read->clients[i]->proc_fd
, m_data.chan);
struct channel *chan = pubsubd_channel_search (chans, m_data.chan);
if (chan == NULL) {
@ -190,8 +190,8 @@ void pubsubd_main_loop (struct ipc_service *srv, struct channels *chans)
}
for (i = 0; i < ap.size; i++) {
if (ipc_server_close_proc (ap.tab_proc[i]) < 0) {
handle_error( "server_close_proc < 0");
if (ipc_server_close_client (ap.clients[i]) < 0) {
handle_error( "server_close_client < 0");
}
}

2
remote/lib/remotec.c
View File

@ -66,7 +66,7 @@ int remotec_message_recv (struct ipc_service *srv, struct remoted_msg *m)
memset (&m_recv, 0, sizeof (struct ipc_message));
ipc_application_read (srv, &m_recv);
remote_message_unserialize (m, m_recv.val, m_recv.valsize);
remote_message_unserialize (m, m_recv.payload, m_recv.length);
ipc_message_free (&m_recv);

34
remote/lib/remoted.c
View File

@ -37,21 +37,21 @@ void handle_new_msg (struct ipc_client_array *ap, struct ipc_client_array *proc_
memset (&m, 0, sizeof (struct ipc_message));
int i;
for (i = 0; i < proc_to_read->size; i++) {
if (ipc_server_read (proc_to_read->tab_proc[i], &m) < 0) {
if (ipc_server_read (proc_to_read->clients[i], &m) < 0) {
handle_error("ipc_server_read < 0");
}
mprint_hexa ("msg received: ", (unsigned char *) m.val, m.valsize);
mprint_hexa ("msg received: ", (unsigned char *) m.payload, m.length);
// close the client then delete it from the client array
if (m.type == MSG_TYPE_CLOSE) {
struct ipc_client *p = proc_to_read->tab_proc[i];
struct ipc_client *p = proc_to_read->clients[i];
log_debug ("remoted, proc %d disconnecting", p->proc_fd);
log_debug ("remoted, client %d disconnecting", p->proc_fd);
// close the connection to the client
if (ipc_server_close_proc (p) < 0)
handle_error( "ipc_server_close_proc < 0");
if (ipc_server_close_client (p) < 0)
handle_error( "ipc_server_close_client < 0");
// remove the client from the clientes list
if (ipc_client_del (ap, p) < 0)
@ -72,27 +72,27 @@ void handle_new_msg (struct ipc_client_array *ap, struct ipc_client_array *proc_
struct pubsub_msg m_data;
memset (&m_data, 0, sizeof (struct pubsub_msg));
pubsub_message_unserialize (&m_data, m.val, m.valsize);
pubsub_message_unserialize (&m_data, m.payload, m.length);
if (m_data.type == PUBSUB_MSG_TYPE_SUB) {
printf ("proc %d subscribing to %s\n"
, proc_to_read->tab_proc[i]->proc_fd
printf ("client %d subscribing to %s\n"
, proc_to_read->clients[i]->proc_fd
, m_data.chan);
pubsubd_channels_subscribe (chans
, m_data.chan, proc_to_read->tab_proc[i]);
, m_data.chan, proc_to_read->clients[i]);
}
if (m_data.type == PUBSUB_MSG_TYPE_UNSUB) {
printf ("proc %d unsubscribing to %s\n"
, proc_to_read->tab_proc[i]->proc_fd
printf ("client %d unsubscribing to %s\n"
, proc_to_read->clients[i]->proc_fd
, m_data.chan);
pubsubd_channels_unsubscribe (chans
, m_data.chan, proc_to_read->tab_proc[i]);
, m_data.chan, proc_to_read->clients[i]);
}
if (m_data.type == PUBSUB_MSG_TYPE_PUB) {
printf ("proc %d publishing to %s\n"
, proc_to_read->tab_proc[i]->proc_fd
printf ("client %d publishing to %s\n"
, proc_to_read->clients[i]->proc_fd
, m_data.chan);
struct channel *chan = pubsubd_channel_search (chans, m_data.chan);
if (chan == NULL) {
@ -136,8 +136,8 @@ void remoted_main_loop (struct ipc_service *srv, struct remoted_ctx *ctx)
}
for (i = 0; i < ap.size; i++) {
if (ipc_server_close_proc (ap.tab_proc[i]) < 0) {
handle_error( "ipc_server_close_proc < 0");
if (ipc_server_close_client (ap.clients[i]) < 0) {
handle_error( "ipc_server_close_client < 0");
}
}
}