diff --git a/dist/antd-publishers-0.1.2a.tar.gz b/dist/antd-publishers-0.1.2a.tar.gz index d3cd265..6979617 100644 Binary files a/dist/antd-publishers-0.1.2a.tar.gz and b/dist/antd-publishers-0.1.2a.tar.gz differ diff --git a/wfifo/wfifo.c b/wfifo/wfifo.c index cdca67d..a5c2ef3 100644 --- a/wfifo/wfifo.c +++ b/wfifo/wfifo.c @@ -10,6 +10,7 @@ #include #include #include +#include #include #include @@ -20,6 +21,7 @@ #define MODULE_NAME "wfifo" static bst_node_t *clients = NULL; +static bst_node_t *fifo_handles = NULL; static volatile int running = 1; @@ -31,14 +33,67 @@ static void int_handler(int dummy) static void send_data(bst_node_t *node, void **argv, int argc) { (void)argc; - tunnel_msg_t *msg = (tunnel_msg_t *)argv[0]; - int *fd = (int *)argv[1]; + tunnel_msg_t *msg = (tunnel_msg_t *)argv[2]; + int *fd = (int *)argv[0]; + int *ffd = (int*)argv[1]; + if(!node || !node->data || (int)node->data != *ffd) + { + return; + } + msg->header.client_id = node->key; if (msg_write(*fd, msg) == -1) { M_ERROR(MODULE_NAME, "Unable to write data message to client %d", node->key); } + M_DEBUG(MODULE_NAME, "Message sent to client %d", node->key); } + +static void prepare_fd_set(bst_node_t *node, void **argv, int argc) +{ + (void)argc; + fd_set* fd_in = (fd_set *)argv[1]; + int* max_fd = (int*) argv[2]; + if(!node || ! node->data) + { + return; + } + FD_SET((int)node->data, fd_in); + *max_fd = (int)node->data > *max_fd? (int)node->data: *max_fd; +} + +static void monitor_fifo_handles(bst_node_t *node, void **argv, int argc) +{ + (void)argc; + int ffd, status; + int* fd; + fd_set* fd_in = (fd_set *)argv[1]; + tunnel_msg_t* msg = (tunnel_msg_t*) argv[2]; + void * fargv[3]; + fd = (int*) argv[0]; + if(!node || ! node->data) + { + return; + } + ffd = (int) node->data; + if(FD_ISSET(ffd, fd_in)) + { + if ((status = read(ffd, msg->data, BUFFLEN)) == -1) + { + M_ERROR(MODULE_NAME, "Unable to read data from the FIFO %d: %s", ffd, strerror(errno)); + return; + } + else + { + msg->header.size = status; + fargv[0] = argv[0]; + fargv[1] = &ffd; + fargv[2] = argv[2]; + bst_for_each(clients, send_data, fargv, 2); + } + } +} + static void unsubscribe(bst_node_t *node, void **args, int argc) { (void)argc; @@ -52,14 +107,98 @@ static void unsubscribe(bst_node_t *node, void **args, int argc) M_ERROR(MODULE_NAME, "Unable to request unsubscribe to client %d", node->key); } } +static void close_fifo_handles(bst_node_t *node, void **args, int argc) +{ + (void)argc; + (void) args; + if(!node || !node->data) + { + return; + } + if((int) node->data > 0) + { + M_DEBUG(MODULE_NAME, "Close fifo handle %d", (int)node->data); + (void) close((int) node->data); + } +} + +int init_fifo(char *buff, const char *base, const char *user) +{ + int fd, hash; + int generic_fd = 1; + struct stat path_stat; + uid_t uid; + struct passwd *pwd = NULL; + bst_node_t * node; + (void)memset(buff, 0, BUFFLEN); + if (stat(base, &path_stat) == 0) + { + if (S_ISREG(path_stat.st_mode)) + { + // is directory + if (user == NULL) + { + M_ERROR(MODULE_NAME, "Cannot init fifo for null user"); + return -1; + } + (void)snprintf(buff, BUFFLEN, "%s/%s.fifo", base, user); + generic_fd = 0; + pwd = getpwnam(user); /* Try getting UID for username */ + if (pwd == NULL) + { + M_ERROR(MODULE_NAME, "Unable to get userid from user %s: %s", user, strerror(errno)); + return -1; + } + uid = pwd->pw_uid; + } + } + if (generic_fd) + { + (void)snprintf(buff, BUFFLEN, "%s", base); + } + // check if file exists + hash = simple_hash(buff); + node = bst_find(fifo_handles,hash); + if(node && node->data) + { + M_DEBUG(MODULE_NAME, "handle for file %s exists (%d)", buff, (int)node->data); + return (int) node->data; + } + (void)unlink(buff); + if (mkfifo(buff, 0666) == -1) + { + M_ERROR(MODULE_NAME, "Unable to create FIFO %s: %s", buff, strerror(errno)); + return -1; + } + fd = open(buff, O_RDWR); + if (fd == -1) + { + M_ERROR(MODULE_NAME, "Unable to open FIFO %s: %s", buff, strerror(errno)); + return -1; + } + if (!generic_fd) + { + // chown file by user + if (chown(buff, uid, -1) == -1) + { + M_ERROR(MODULE_NAME, "Unable to change ownerfor file %s to %s: %s", buff, user, strerror(errno)); + (void) close(fd); + return -1; + } + } + fifo_handles = bst_insert(fifo_handles, hash, (void*)fd); + M_LOG(MODULE_NAME, "FIFO: %s created", buff); + return fd; +} int main(int argc, char **argv) { - int fd, fifo_fd; + int fd, ffd; tunnel_msg_t msg; fd_set fd_in; int status, maxfd; char buff[BUFFLEN + 1]; - void *fargv[2]; + void *fargv[3]; + bst_node_t* node = NULL; uint8_t *tmp; LOG_INIT(MODULE_NAME); @@ -71,6 +210,7 @@ int main(int argc, char **argv) signal(SIGPIPE, SIG_IGN); signal(SIGABRT, SIG_IGN); signal(SIGINT, int_handler); + // create the fifo first (void)unlink(argv[3]); if (mkfifo(argv[3], 0666) == -1) @@ -78,21 +218,11 @@ int main(int argc, char **argv) M_ERROR(MODULE_NAME, "Unable to create FIFO %s: %s", argv[3], strerror(errno)); return -1; } - fifo_fd = open(argv[3], O_RDWR); - if (fifo_fd == -1) - { - M_ERROR(MODULE_NAME, "Unable to open FIFO %s: %s", argv[3], strerror(errno)); - return -1; - } - M_LOG(MODULE_NAME, "FIFO: %s created", argv[3]); - - M_LOG(MODULE_NAME, "Hotline is: %s", argv[1]); // now try to request new channel from hotline fd = open_unix_socket(argv[1]); if (fd == -1) { M_ERROR(MODULE_NAME, "Unable to open the hotline: %s", argv[1]); - (void)close(fifo_fd); return -1; } msg.header.type = CHANNEL_OPEN; @@ -106,7 +236,6 @@ int main(int argc, char **argv) { M_ERROR(MODULE_NAME, "Unable to write message to hotline"); (void)close(fd); - (void)close(fifo_fd); return -1; } M_LOG(MODULE_NAME, "Wait for comfirm creation of %s", argv[2]); @@ -115,7 +244,6 @@ int main(int argc, char **argv) { M_ERROR(MODULE_NAME, "Unable to read message from hotline"); (void)close(fd); - (void)close(fifo_fd); return -1; } if (msg.header.type == CHANNEL_OK) @@ -131,15 +259,24 @@ int main(int argc, char **argv) free(msg.data); running = 0; } + /** + * @brief init global fifo handle + * + * If the publisher is configured to be user base fifo, + * a error LOG will be shown + */ + (void)init_fifo(buff, argv[3], NULL); + fargv[0] = (void *)&fd; // now read data while (running) { FD_ZERO(&fd_in); FD_SET(fd, &fd_in); - FD_SET(fifo_fd, &fd_in); - maxfd = fd > fifo_fd ? fd : fifo_fd; - + fargv[1] = &fd_in; + maxfd = fd; + fargv[2] = &maxfd; + bst_for_each(fifo_handles, prepare_fd_set, fargv, 3); status = select(maxfd + 1, &fd_in, NULL, NULL, NULL); switch (status) @@ -164,8 +301,12 @@ int main(int argc, char **argv) switch (msg.header.type) { case CHANNEL_SUBSCRIBE: - M_LOG(MODULE_NAME, "Client %d subscribes to the chanel", msg.header.client_id); - clients = bst_insert(clients, msg.header.client_id, NULL); + ffd = init_fifo(buff, argv[3], (char*)msg.data); + if(ffd != -1) + { + clients = bst_insert(clients, msg.header.client_id, (void*)ffd); + M_LOG(MODULE_NAME, "Client %d subscribes to the chanel", msg.header.client_id); + } break; case CHANNEL_UNSUBSCRIBE: @@ -176,13 +317,17 @@ int main(int argc, char **argv) case CHANNEL_DATA: if (argv[4][0] == 'w') { - // write data to the FIFO - if (msg.header.size > 0) + node = bst_find(clients, msg.header.client_id); + if(node && node->data) { - if (write(fifo_fd, msg.data, msg.header.size) == -1) + // write data to the FIFO + if (msg.header.size > 0) { - M_ERROR(MODULE_NAME, "Unable to write data to the FIFO %s from client %d: %s", argv[3], msg.header.client_id, strerror(errno)); - running = 0; + if (write((int)node->data, msg.data, msg.header.size) == -1) + { + M_ERROR(MODULE_NAME, "Unable to write data to the FIFO %s from client %d: %s", argv[3], msg.header.client_id, strerror(errno)); + running = 0; + } } } } @@ -214,30 +359,20 @@ int main(int argc, char **argv) } } } - else if (FD_ISSET(fifo_fd, &fd_in) && argv[4][0] == 'r') + else if (argv[4][0] == 'r') { // on the fifo side - if ((status = read(fifo_fd, buff, BUFFLEN)) == -1) - { - M_ERROR(MODULE_NAME, "Unable to read data from the FIFO %s: %s", argv[3], strerror(errno)); - running = 0; - } - else - { - msg.header.type = CHANNEL_DATA; - msg.header.size = status; - msg.data = (uint8_t *)buff; - fargv[0] = (void *)&msg; - fargv[1] = (void *)&fd; - bst_for_each(clients, send_data, fargv, 2); - } + fargv[1] = &fd_in; + fargv[2] = &msg; + msg.data = (uint8_t *)buff; + msg.header.type = CHANNEL_DATA; + bst_for_each(fifo_handles, monitor_fifo_handles, fargv, 3); } } } // unsubscribe all client - fargv[0] = (void *)&fd; bst_for_each(clients, unsubscribe, fargv, 1); - + bst_for_each(fifo_handles, close_fifo_handles, NULL, 0); // close the channel M_LOG(MODULE_NAME, "Close the channel %s (%d)", argv[2], fd); msg.header.type = CHANNEL_CLOSE; @@ -251,6 +386,7 @@ int main(int argc, char **argv) (void)msg_read(fd, &msg); (void)close(fd); - (void)close(fifo_fd); + bst_free(clients); + bst_free(fifo_handles); return 0; } \ No newline at end of file