dev remote -- draft 1
Philippe PITTOLI
parent
ded94daf43
commit
eea95e5ee3
|
|
@ -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
|
||||
|
|
@ -0,0 +1,104 @@
|
|||
// int main(void) { return 0; }
|
||||
|
||||
#include "../../core/error.h"
|
||||
#include "../lib/remote.h"
|
||||
#include "../lib/remoted.h"
|
||||
|
||||
#include <stdlib.h>
|
||||
#include <pthread.h>
|
||||
#include <unistd.h>
|
||||
|
||||
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;
|
||||
}
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
#include "../../core/communication.h"
|
||||
#include "../../core/process.h"
|
||||
#include "../../core/error.h"
|
||||
#include "../lib/pubsubd.h"
|
||||
#include <stdlib.h>
|
||||
|
||||
#include <sys/socket.h>
|
||||
#include <sys/un.h>
|
||||
#include <unistd.h>
|
||||
#include <signal.h>
|
||||
|
||||
// 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;
|
||||
}
|
||||
|
|
@ -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 <sys/socket.h>
|
||||
#include <sys/un.h>
|
||||
#include <unistd.h>
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
|
|
@ -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
|
||||
Reference in New Issue