Obsolete
/
libipc-old
Archived
3
0
Fork 0
Code

remote: initial draft

more_to_read
Philippe PITTOLI 2017-08-25 11:42:43 +02:00
parent eea95e5ee3
commit 2ad37c2a72
8 changed files with 54 additions and 216 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
remote/app/README.md Normal file
View File

@ -0,0 +1,3 @@
# remoted
This service creates a path on the relevent remote location, going through anything network-related: TCP, UDP, HTTP, ...

12
remote/app/remotec.c
View File

@ -1,8 +1,9 @@
// int main(void) { return 0; }
#include "../../core/communication.h"
#include "../../core/error.h"
#include "../lib/remote.h"
#include "../lib/remoted.h"
#include "../lib/remote.h"
#include <stdlib.h>
#include <pthread.h>
@ -12,6 +13,7 @@ void usage (char **argv) {
printf ( "usage: %s uri service\n", argv[0]);
}
#if 0
void * listener (void *params)
{
int s = 0;
@ -41,6 +43,7 @@ void * listener (void *params)
pthread_exit (NULL);
}
#endif
void main_loop (int argc, char **argv, char **env
, int index, int version, char *uri, char *service)
@ -48,6 +51,12 @@ void main_loop (int argc, char **argv, char **env
printf ("connection to remoted: index %d version %d uri %s service %s\n"
, index, version, uri, service);
(void) argc;
(void) argv;
(void) env;
#if 0
struct service srv;
memset (&srv, 0, sizeof (struct service));
remote_connection (argc, argv, env, &srv);
@ -86,6 +95,7 @@ void main_loop (int argc, char **argv, char **env
printf ("disconnection...\n");
// disconnect from the server
remote_disconnect (&srv);
#endif
}
int main(int argc, char **argv, char **env)

7
remote/app/remoted.c
View File

@ -1,7 +1,7 @@
#include "../../core/communication.h"
#include "../../core/process.h"
#include "../../core/error.h"
#include "../lib/pubsubd.h"
#include "../lib/remoted.h"
#include <stdlib.h>
#include <sys/socket.h>
@ -11,7 +11,6 @@
// to quit them properly if a signal occurs
struct service srv;
struct channels chans;
void handle_signal (int signalnumber)
{
@ -24,7 +23,7 @@ void handle_signal (int signalnumber)
exit (EXIT_SUCCESS);
}
void remoted_init () {}
void remoted_init () { /* TODO */}
int
main(int argc, char **argv, char **env)
@ -48,7 +47,7 @@ main(int argc, char **argv, char **env)
printf("MAIN: server created\n" );
// the service will loop until the end of time, a specific message, a signal
pubsubd_main_loop (&srv, &chans);
remoted_main_loop (&srv);
// the application will shut down, and remove the service named pipe
if (srv_close (&srv) < 0) {

196
remote/lib/pubsubd.c
View File

@ -1,196 +0,0 @@
#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);
}

15
remote/lib/pubsubd.h
View File

@ -1,15 +0,0 @@
#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

9
remote/lib/remote.h Normal file
View File

@ -0,0 +1,9 @@
#ifndef __REMOTE_H__
#define __REMOTE_H__
#include "../../core/process.h"
#include "../../core/msg.h"
/* TODO */
#endif

17
remote/lib/remoted.c Normal file
View File

@ -0,0 +1,17 @@
#include "../../core/communication.h"
#include "../../core/msg.h"
#include "../../core/process.h"
#include "../../core/utils.h"
#include "../../core/error.h"
#include "remoted.h"
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
void remoted_main_loop (struct service *srv)
{
(void) srv;
/* TODO */
}

11
remote/lib/remoted.h Normal file
View File

@ -0,0 +1,11 @@
#ifndef __REMOTED_H__
#define __REMOTED_H__
#include "../../core/process.h"
#include "../../core/msg.h"
#define REMOTED_SERVICE_NAME "remoted"
void remoted_main_loop (struct service *srv);
#endif