+
+ /* 1. Check if there is a message in the buf */
+ /* 2. If not, do:
+ 2.1 receive chunk and put it in buffer
+ 2.2 process full message if there is one
+ -- while no message arrived
+ */
+
+ for(;;) {
+ int i;
+ int nulfound = 0;
+
+ /* Search for full message in buffer */
+ for(i=0; i<*recv_buf_size; i++) {
+ if((*recv_buf)[i] == '\0') {
+ nulfound = 1;
+ break;
+ }
+ }
+
+ /* Process found message */
+ if(nulfound == 1) {
+ char *newbuf;
+
+ if(i == 0) {
+ /* problem */
+ WARN("received empty message");
+ }
+ *msg = strndup(*recv_buf, i);
+
+ /* Remove processed message from buffer */
+ newbuf = (char *) malloc(*recv_buf_size - (i+1));
+ memcpy(newbuf, *recv_buf + (i+1), *recv_buf_size - (i+1));
+ free(*recv_buf);
+ *recv_buf = newbuf;
+ *recv_buf_size -= (i+1);
+ *recv_buf_alloc -= (i+1);
+
+ return 1;
+ }
+
+ /* Receive a chunk from the fd */
+ if(*recv_buf_alloc - *recv_buf_size < RECV_INCREMENT) {
+ *recv_buf_alloc += RECV_INCREMENT - (*recv_buf_alloc - *recv_buf_size);
+ *recv_buf = (char *) realloc(*recv_buf, *recv_buf_alloc);
+ }
+
+ result = recv(fd, *recv_buf+*recv_buf_size, RECV_INCREMENT, 0);
+ if(result == -1) {
+ if(errno == ECONNRESET) {
+ *recv_buf_size = 0;
+ return 0;
+ }
+ else if(errno == EINTR) {
+ return -1;
+ }
+ else {
+ PERROR("recv");
+ return -1;
+ }
+ }
+ if(result == 0) {
+ return 0;
+ }
+ *recv_buf_size += result;
+
+ /* Go back to the beginning to check if there is a full message in the buffer */
+ }
+
+ DBG("received message \"%s\"", *recv_buf);
+
+ return 1;
+
+}
+
+static int recv_message_conn(struct ustcomm_connection *conn, char **msg)
+{
+ return recv_message_fd(conn->fd, &conn->recv_buf, &conn->recv_buf_size, &conn->recv_buf_alloc, msg);
+}
+
+int ustcomm_send_reply(struct ustcomm_server *server, char *msg, struct ustcomm_source *src)
+{
+ int result;
+
+ result = send_message_fd(src->fd, msg);
+ if(result < 0) {
+ ERR("error in send_message_fd");
+ return -1;
+ }
+
+ return 0;
+}
+
+/* Called after a fork. */
+
+int ustcomm_close_all_connections(struct ustcomm_server *server)
+{
+ struct ustcomm_connection *conn;
+ struct ustcomm_connection *deletable_conn = NULL;
+
+ list_for_each_entry(conn, &server->connections, list) {
+ free(deletable_conn);
+ deletable_conn = conn;
+ ustcomm_close_app(conn);
+ list_del(&conn->list);
+ }
+
+ return 0;
+}
+
+/* @timeout: max blocking time in milliseconds, -1 means infinity
+ *
+ * returns 1 to indicate a message was received
+ * returns 0 to indicate no message was received
+ * returns -1 to indicate an error
+ */
+
+int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustcomm_source *src, int timeout)
+{
+ struct pollfd *fds;
+ struct ustcomm_connection **conn_table;
+ struct ustcomm_connection *conn;
+ int result;
+ int retval;
+
+ for(;;) {
+ int idx = 0;
+ int n_fds = 1;
+
+ list_for_each_entry(conn, &server->connections, list) {
+ n_fds++;
+ }
+
+ fds = (struct pollfd *) zmalloc(n_fds * sizeof(struct pollfd));
+ if(fds == NULL) {
+ ERR("zmalloc returned NULL");
+ return -1;
+ }
+
+ conn_table = (struct ustcomm_connection **) zmalloc(n_fds * sizeof(struct ustcomm_connection *));
+ if(conn_table == NULL) {
+ ERR("zmalloc returned NULL");
+ retval = -1;
+ goto free_fds_return;
+ }
+
+ /* special idx 0 is for listening socket */
+ fds[idx].fd = server->listen_fd;
+ fds[idx].events = POLLIN;
+ idx++;
+
+ list_for_each_entry(conn, &server->connections, list) {
+ fds[idx].fd = conn->fd;
+ fds[idx].events = POLLIN;
+ conn_table[idx] = conn;
+ idx++;
+ }
+
+ result = poll(fds, n_fds, timeout);
+ if(result == -1 && errno == EINTR) {
+ /* That's ok. ustd receives signals to notify it must shutdown. */
+ retval = -1;
+ goto free_conn_table_return;
+ }
+ else if(result == -1) {
+ PERROR("poll");
+ retval = -1;
+ goto free_conn_table_return;
+ }
+ else if(result == 0) {
+ retval = 0;
+ goto free_conn_table_return;
+ }
+
+ if(fds[0].revents) {
+ struct ustcomm_connection *newconn;
+ int newfd;
+
+ result = newfd = accept(server->listen_fd, NULL, NULL);
+ if(result == -1) {
+ PERROR("accept");
+ retval = -1;
+ goto free_conn_table_return;
+ }
+
+ newconn = (struct ustcomm_connection *) zmalloc(sizeof(struct ustcomm_connection));
+ if(newconn == NULL) {
+ ERR("zmalloc returned NULL");
+ return -1;
+ }
+
+ ustcomm_init_connection(newconn);
+ newconn->fd = newfd;
+
+ list_add(&newconn->list, &server->connections);
+ }
+
+ for(idx=1; idx<n_fds; idx++) {
+ if(fds[idx].revents) {
+ retval = recv_message_conn(conn_table[idx], msg);
+ if(src)
+ src->fd = fds[idx].fd;
+
+ if(retval == 0) {
+ /* connection finished */
+ list_for_each_entry(conn, &server->connections, list) {
+ if(conn->fd == fds[idx].fd) {
+ ustcomm_close_app(conn);
+ list_del(&conn->list);
+ free(conn);
+ break;
+ }
+ }
+ }
+ else {
+ goto free_conn_table_return;
+ }
+ }
+ }
+
+ free(fds);
+ free(conn_table);
+ }
+
+free_conn_table_return:
+ free(conn_table);
+free_fds_return:
+ free(fds);
+ return retval;
+}
+
+int ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg, struct ustcomm_source *src, int timeout)
+{
+ return ustcomm_recv_message(&ustd->server, msg, src, timeout);
+}
+
+int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src, int timeout)
+{
+ return ustcomm_recv_message(&app->server, msg, src, timeout);
+}
+
+/* This removes src from the list of active connections of app.
+ */
+
+int ustcomm_app_detach_client(struct ustcomm_app *app, struct ustcomm_source *src)
+{
+ struct ustcomm_server *server = (struct ustcomm_server *)app;
+ struct ustcomm_connection *conn;
+
+ list_for_each_entry(conn, &server->connections, list) {
+ if(conn->fd == src->fd) {
+ list_del(&conn->list);
+ goto found;
+ }
+ }
+
+ return -1;
+found:
+ return src->fd;
+}
+
+static int init_named_socket(const char *name, char **path_out)
+{
+ int result;
+ int fd;
+