Obsolete
/
libipc-old
Archived
3
0
Fork 0
Code

beaucoup de modifications à l'arrache

more_to_read
Philippe PITTOLI 2016-12-17 18:00:04 +01:00
parent 044e8d034e
commit b426ab7d7b
7 changed files with 219 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
core/communication.c
View File

@ -1,54 +1,23 @@
#include "communication.h"
#include "usocket.h"
#include <assert.h>
#include <stdio.h>
#include <time.h>
#include <errno.h>
#include <sys/socket.h>
#include <sys/un.h>
#define LISTEN_BACKLOG 50
#define handle_error(msg) \
do { perror(msg); exit(EXIT_FAILURE); } while (0)
int msg_send (const int fd, const char *buf, const int msize)
void service_path (char *path, const char *name, int index, int version)
{
int ret = 0;
//printf ("%ld bytes to write\n", msize);
ret = send (fd, buf, msize, 0);
if (ret <= 0) {
fprintf (stderr, "err: written %d\n", fd);
}
return ret;
memset (path, 0, PATH_MAX);
snprintf (path, PATH_MAX, "%s/%s-%d-%d", TMPDIR, sname, index, version);
}
int msg_recv (const int fd, char **buf)
{
int ret = 0;
ret = recv (fd, *buf, BUFSIZ, 0);
if (ret < 0) {
fprintf (stderr, "err: read %d\n", fd);
}
return ret;
}
int close_socket (int fd)
{
int ret;
ret = close (fd);
if (ret < 0) {
fprintf (stderr, "err: close [err: %d] %d\n", ret, fd);
perror ("closing");
}
return ret;
}
// SERVICE
// init unix socket + srv->spath filled
int srv_init (int argc, char **argv, char **env, struct service *srv, const char *sname)
int srv_init (int argc, char **argv, char **env
, struct service *srv, const char *sname)
{
if (srv == NULL)
return ER_PARAMS;
@ -63,47 +32,18 @@ int srv_init (int argc, char **argv, char **env, struct service *srv, const char
argv = argv;
env = env;
// srv->version => already set
// srv->index => already set
// gets the service path
service_path (srv->spath, sname, srv->index, srv->version);
// gets the service path, such as /tmp/ipc/<service>
memset (srv->spath, 0, PATH_MAX);
snprintf (srv->spath, PATH_MAX, "%s/%s-%d-%d"
, TMPDIR, sname, srv->index, srv->version);
// TODO TEST create a unix socket
int sfd;
struct sockaddr_un my_addr;
sfd = socket(AF_UNIX, SOCK_STREAM, 0);
if (sfd == -1)
return -1;
// clear structure
memset(&my_addr, 0, sizeof(struct sockaddr_un));
my_addr.sun_family = AF_UNIX;
strncpy(my_addr.sun_path, srv->spath, strlen (srv->spath)); // TODO check size
// delete the unix socket if already created
// TODO FIXME
unlink(my_addr.sun_path);
if (bind(sfd, (struct sockaddr *) &my_addr, sizeof(struct sockaddr_un)) == -1)
return -1;
if (listen(sfd, LISTEN_BACKLOG) == -1)
return -1;
srv->service_fd = sfd;
usock_listen (&srv->service_fd, srv->spath);
return 0;
}
int srv_close (struct service *srv)
{
close_socket (srv->service_fd);
usock_close (srv->service_fd);
// TODO FIXME is unlink really necessary
if (unlink (srv->spath)) {
return 1;
}
@ -111,62 +51,33 @@ int srv_close (struct service *srv)
return 0;
}
int srv_read (const struct service *srv, char ** buf)
int srv_read (const struct service *srv, char ** buf, size_t *msize)
{
//printf("---%s\n", srv->spath);
return msg_recv (srv->service_fd, buf);
return usock_recv (srv->service_fd, buf, msize);
}
int srv_write (const struct service *srv, const char * buf, size_t msize)
{
//printf("---%s\n", srv->spath);
return msg_send (srv->service_fd, buf, msize);
return usock_send (srv->service_fd, buf, msize);
}
// APPLICATION
// Initialize connection with unix socket
// send the connection string to $TMP/<service>
// fill srv->spath && srv->service_fd
int app_connection (struct service *srv, const char *sname
, const char *connectionstr, size_t msize)
{
assert (srv != NULL);
assert (sname != NULL);
if (srv == NULL) {
return -1;
}
// srv->version => already set
// srv->index => already set
// gets the service path
service_path (srv->spath, sname, srv->index, srv->version);
// gets the service path, such as /tmp/ipc/<service>
memset (srv->spath, 0, PATH_MAX);
snprintf (srv->spath, PATH_MAX, "%s/%s-%d-%d"
, TMPDIR, sname, srv->index, srv->version);
usock_connect(&srv->service_fd, srv->spath);
int sfd;
struct sockaddr_un my_addr;
socklen_t peer_addr_size;
sfd = socket(AF_UNIX, SOCK_STREAM, 0);
if (sfd == -1)
return -1;
// clear structure
memset(&my_addr, 0, sizeof(struct sockaddr_un));
my_addr.sun_family = AF_UNIX;
strncpy(my_addr.sun_path, srv->spath, strlen (srv->spath)); // TODO check size
peer_addr_size = sizeof(struct sockaddr_un);
if(connect(sfd, (struct sockaddr *) &my_addr, peer_addr_size) == -1)
{
perror("connect()");
exit(errno);
}
srv->service_fd = sfd;
// TODO FIXME
// TODO: connection algorithm
// send connection string and receive acknowledgement
srv_write(srv, connectionstr, msize);
@ -175,15 +86,15 @@ int app_connection (struct service *srv, const char *sname
int app_close (struct service *srv)
{
return close_socket (srv->service_fd);
return usock_close (srv->service_fd);
}
int app_read (struct service *srv, char ** buf)
int app_read (struct service *srv, char ** buf, size_t *msize)
{
return msg_recv (srv->service_fd, buf);
return usock_recv (srv->service_fd, buf, msize);
}
int app_write (struct service *srv, char * buf, size_t msize)
{
return msg_send (srv->service_fd, buf, msize);
return usock_send (srv->service_fd, buf, msize);
}

21
core/communication.h
View File

@ -22,8 +22,12 @@
#define ER_FILE_WRITE 4
#define ER_FILE_WRITE_PARAMS 5
#define ER_MEM_ALLOC 100
#define ER_PARAMS 101
#define ER_MEM_ALLOC 100
#define ER_PARAMS 101
#define TMPDIR "/tmp/ipc/"
#define PATH_MAX BUFSIZ
struct service {
unsigned int version;
@ -32,14 +36,11 @@ struct service {
int service_fd;
};
// wrappers
int msg_recv (int fd, char **buf);
int msg_send (int fd, const char *buf, const int m_size);
int close_socket (int fd);
// SERVICE
// srv->version and srv->index must be already set
// init unix socket + fill srv->spath
int srv_init (int argc, char **argv, char **env
, struct service *srv, const char *sname);
int srv_close (struct service *srv);
@ -49,11 +50,13 @@ int srv_write (const struct service *, const char * buf, size_t);
// APPLICATION
// Initialize connection with unix socket
// send the connection string to $TMP/<service>
// fill srv->spath && srv->service_fd
int app_connection (struct service *, const char *, const char *, size_t);
int app_close (struct service *);
int app_read (struct service *, char ** buf);
int app_write (struct service *, char * buf, size_t);
int app_read (struct service *srv, char ** buf, size_t *msize);
int app_write (struct service *, char * buf, size_t msize);
#endif

5
core/process.h
View File

@ -6,11 +6,6 @@
#include <stdio.h>
#include <stdlib.h>
#define TMPDIR "/tmp/ipc/"
// TODO to check the right length for a path
#define PATH_MAX BUFSIZ
#include <string.h>
struct process {

152
core/usocket.c Normal file
View File

@ -0,0 +1,152 @@
#include "usocket.h"
#include <assert.h>
#define handle_err(fun,msg)\
fprintf (stderr, "%s: file %s line %d %s\n", fun, __FILE__, __LINE__, msg);
int usock_send (const int fd, const char *buf, const int msize)
{
int ret = 0;
//printf ("%ld bytes to write\n", msize);
ret = send (fd, buf, msize, 0);
if (ret <= 0) {
fprintf (stderr, "usock_send: file %s line %d send ret <= 0\n"
, __FILE__, __LINE__);
}
return ret;
}
int usock_recv (const int fd, char **buf, size_t *msize)
{
assert(buf != NULL);
assert(msize != NULL);
if (buf == NULL) {
handle_err ("usock_recv", "buf == NULL");
return -1;
}
if (msize == NULL) {
handle_err ("usock_recv", "msize == NULL");
return -1;
}
if (*buf == NULL) {
// do not allocate too much memory
*buf = malloc ((*msize < BUFSIZ) ? *msize : BUFSIZ);
}
int ret = 0;
ret = recv (fd, *buf, *msize, 0);
if (ret < 0) {
handle_err ("usock_recv", "recv ret < 0");
*msize = 0;
return ret;
}
*msize = (size_t) ret;
return ret;
}
int usock_connect (int *fd, const char *path)
{
assert (fd != NULL);
assert (path != NULL);
if (fd == NULL) {
handle_err ("usock_connect", "fd == NULL");
return -1;
}
if (path == NULL) {
handle_err ("usock_connect", "path == NULL");
return -1;
}
int sfd;
struct sockaddr_un my_addr;
socklen_t peer_addr_size;
sfd = socket(AF_UNIX, SOCK_STREAM, 0);
if (sfd == -1) {
handle_err ("usock_connect", "sfd == -1");
return -1;
}
// clear structure
memset(&my_addr, 0, sizeof(struct sockaddr_un));
my_addr.sun_family = AF_UNIX;
strncpy(my_addr.sun_path, path, strlen (path));
peer_addr_size = sizeof(struct sockaddr_un);
if(connect(sfd, (struct sockaddr *) &my_addr, peer_addr_size) == -1) {
handle_err ("usock_connect", "connect == -1");
perror("connect()");
exit(errno);
}
*fd = sfd;
}
int usock_listen (int *fd, const char *path)
{
assert (fd != NULL);
assert (path != NULL);
if (fd == NULL) {
handle_err ("usock_listen", "fd == NULL");
return -1;
}
if (path == NULL) {
handle_err ("usock_listen", "path == NULL");
return -1;
}
int sfd;
struct sockaddr_un my_addr;
socklen_t peer_addr_size;
sfd = socket(AF_UNIX, SOCK_STREAM, 0);
if (sfd == -1) {
handle_err ("usock_listen", "sfd == -1");
return -1;
}
// clear structure
memset(&my_addr, 0, sizeof(struct sockaddr_un));
my_addr.sun_family = AF_UNIX;
strncpy(my_addr.sun_path, path, strlen (path));
// TODO FIXME
// delete the unix socket if already created
unlink(my_addr.sun_path);
peer_addr_size = sizeof(struct sockaddr_un);
if (bind(sfd, (struct sockaddr *) &my_addr, peer_addr_size) == -1) {
handle_err ("usock_listen", "bind == -1");
perror("bind()");
return -1;
}
if (listen(sfd, LISTEN_BACKLOG) == -1) {
handle_err ("usock_listen", "listen == -1");
perror("listen()");
return -1;
}
*fd = sfd;
}
int usock_close (int fd)
{
int ret;
ret = close (fd);
if (ret < 0) {
handle_err ("usock_close", "close ret < 0");
perror ("closing");
}
return ret;
}

24
core/usocket.h Normal file
View File

@ -0,0 +1,24 @@
#ifndef __USOCKET_H__
#define __USOCKET_H__
#include <time.h>
#include <sys/socket.h>
#include <sys/un.h>
// same as recv(2)
int usock_send (int fd, const char *buf, const int m_size);
// same as send(2)
// if msize == NULL => -1
// if buf == NULL => -1
// if *buf == NULL => allocation of *msize bytes
int usock_recv (int fd, char **buf, size_t *msize);
// same as close(2)
int usock_close (int fd);
// same as connect(2)
// if fd == NULL => -1
int usock_connect (int *fd, const char *path)
#endif

4
pubsub/lib/pubsubd.c
View File

@ -575,7 +575,7 @@ void pubsubd_msg_send (const struct app_list_head *alh, const struct pubsub_msg
pubsubd_msg_serialize (m, &buf, &msize);
LIST_FOREACH(ale, alh, entries) {
srv_write (ale->p, buf, msize);
srv_write (ale->p->proc_fd, buf, msize);
}
if (buf != NULL) {
@ -589,7 +589,7 @@ void pubsubd_msg_recv (struct process *p, struct pubsub_msg *m)
size_t mlen = 0;
char *buf = NULL;
while (buf == NULL || mlen == 0) {
srv_read (p, &buf, &mlen);
srv_read (p->proc_fd, &buf, &mlen);
}
pubsubd_msg_unserialize (m, buf, mlen);

2
pubsub/lib/pubsubd.h
View File

@ -1,7 +1,7 @@
#ifndef __PUBSUBD_H__
#define __PUBSUBD_H__
#include "../../lib/pubsub.h"
#include "../../core/pubsub.h"
#include <pthread.h>
struct channel;