first working vterm

configure.ac

@@ -14,6 +14,21 @@ AC_PROG_LIBTOOL
 AC_DEFINE([_GNU_SOURCE], [1],[Use GNU source])
 # AC_CANONICAL_HOST is needed to access the 'host_os' variable
 
+has_antd=no
+# check for lib antd
+AC_CHECK_HEADER([antd/plugin.h],[
+    has_antd=yes
+    # check if the library exists
+],[
+    AC_MSG_ERROR([Unable to find antd, please install it first])
+])
+AC_CHECK_LIB([antd],[antd_send],[],[
+    if test "$has_antd" = "yes"; then
+        AC_MSG_ERROR([Unable to find antd shared library, please install it first])
+    fi
+])
+
+
 # debug option
 AC_ARG_ENABLE([debug],
     [  --enable-debug    Turn on debugging],

log.h

@@ -9,11 +9,11 @@
     } while(0)
 
 #ifdef DEBUG
-	#define LOG(m, a,...) syslog ((LOG_NOTICE),m "_log@[%s: %d]: " a "\n", __FILE__, \
+	#define M_LOG(m, a,...) syslog ((LOG_NOTICE),m "_log@[%s: %d]: " a "\n", __FILE__, \
 		__LINE__, ##__VA_ARGS__)
 #else
-    #define LOG(m, a,...) do{}while(0)
+    #define M_LOG(m, a,...) do{}while(0)
 #endif
-#define ERROR(m, a,...) syslog ((LOG_ERR),m "_error@[%s: %d]: " a "\n", __FILE__, \
+#define M_ERROR(m, a,...) syslog ((LOG_ERR),m "_error@[%s: %d]: " a "\n", __FILE__, \
 		__LINE__, ##__VA_ARGS__)
 #endif

tunnel.c

@@ -17,12 +17,12 @@ static int msg_check_number(int fd, int number)
     int value;
     if(read(fd,&value,sizeof(value)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read integer value: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to read integer value: %s", strerror(errno));
         return -1;
     }
     if(number != value)
     {
-        ERROR(MODULE_NAME, "Value mismatches: %04X, expected %04X", value, number);
+        M_ERROR(MODULE_NAME, "Value mismatches: %04X, expected %04X", value, number);
         return -1;
     }
     return 0;
@@ -32,17 +32,17 @@ static int msg_read_string(int fd, char* buffer, uint8_t max_length)
     uint8_t size;
     if(read(fd,&size,sizeof(size)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read string size: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to read string size: %s", strerror(errno));
         return -1;
     }
     if(size > max_length)
     {
-        ERROR(MODULE_NAME, "String length exceeds the maximal value of %d", max_length);
+        M_ERROR(MODULE_NAME, "String length exceeds the maximal value of %d", max_length);
         return -1;
     }
     if(read(fd,buffer,size) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read string to buffer: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to read string to buffer: %s", strerror(errno));
         return -1;
     }
     return 0;
@@ -53,7 +53,7 @@ static uint8_t* msg_read_payload(int fd, int* size)
     uint8_t* data;
     if(read(fd,size,sizeof(*size)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read payload data size: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to read payload data size: %s", strerror(errno));
         return NULL;
     }
     if(*size <= 0)
@@ -64,12 +64,12 @@ static uint8_t* msg_read_payload(int fd, int* size)
     data = (uint8_t*) malloc(*size);
     if(data == NULL)
     {
-        ERROR(MODULE_NAME, "Unable to allocate memory for payload data: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to allocate memory for payload data: %s", strerror(errno));
         return NULL;
     }
     if(read(fd,data,*size) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read payload data to buffer: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to read payload data to buffer: %s", strerror(errno));
         free(data);
         return NULL;
     }
@@ -86,15 +86,15 @@ int open_unix_socket(char* path)
     int fd = socket(AF_UNIX, SOCK_STREAM, 0);
     if(fd == -1)
     {
-        ERROR(MODULE_NAME, "Unable to create Unix domain socket: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to create Unix domain socket: %s", strerror(errno));
         return -1;
     }
     if(connect(fd, (struct sockaddr*)(&address), sizeof(address)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to connect to socket '%s': %s", address.sun_path, strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to connect to socket '%s': %s", address.sun_path, strerror(errno));
         return -1;
     }
-    LOG(MODULE_NAME, "Socket %s is created successfully", path);
+    M_LOG(MODULE_NAME, "Socket %s is created successfully", path);
     return fd;
 }
 
@@ -103,32 +103,32 @@ int msg_read(int fd, tunnel_msg_t* msg)
 {
     if(msg_check_number(fd, MSG_MAGIC_BEGIN) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to check begin magic number");
+        M_ERROR(MODULE_NAME, "Unable to check begin magic number");
         return -1;
     }
     if(read(fd,&msg->header.type,sizeof(msg->header.type)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read msg type: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to read msg type: %s", strerror(errno));
         return -1;
     }
-    if(msg->header.type > 0x6)
+    if(msg->header.type > 0x7)
     {
-        ERROR(MODULE_NAME, "Unknown msg type: %d", msg->header.type);
+        M_ERROR(MODULE_NAME, "Unknown msg type: %d", msg->header.type);
         return -1;
     }
     if(read(fd, &msg->header.channel_id, sizeof(msg->header.channel_id)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read msg channel id");
+        M_ERROR(MODULE_NAME, "Unable to read msg channel id");
         return -1;
     }
     if(read(fd, &msg->header.client_id, sizeof(msg->header.client_id)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read msg client id");
+        M_ERROR(MODULE_NAME, "Unable to read msg client id");
         return -1;
     }
     if((msg->data = msg_read_payload(fd, &msg->header.size)) == NULL && msg->header.size != 0)
     {
-        ERROR(MODULE_NAME, "Unable to read msg payload data");
+        M_ERROR(MODULE_NAME, "Unable to read msg payload data");
         return -1;
     }
     if(msg_check_number(fd, MSG_MAGIC_END) == -1)
@@ -137,7 +137,7 @@ int msg_read(int fd, tunnel_msg_t* msg)
         {
             free(msg->data);
         }
-        ERROR(MODULE_NAME, "Unable to check end magic number");
+        M_ERROR(MODULE_NAME, "Unable to check end magic number");
         return -1;
     }
     return 0;
@@ -149,32 +149,32 @@ int msg_write(int fd, tunnel_msg_t* msg)
     int number = MSG_MAGIC_BEGIN;
     if(write(fd,&number, sizeof(number)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to write begin magic number: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to write begin magic number: %s", strerror(errno));
         return -1;
     }
     // write type
     if(write(fd,&msg->header.type, sizeof(msg->header.type)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to write msg type: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to write msg type: %s", strerror(errno));
         return -1;
     }
     // write channel id
     if(write(fd,&msg->header.channel_id, sizeof(msg->header.channel_id)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to write msg channel id: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to write msg channel id: %s", strerror(errno));
         return -1;
     }
     //write client id
     if(write(fd,&msg->header.client_id, sizeof(msg->header.client_id)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to write msg client id: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to write msg client id: %s", strerror(errno));
         return -1;
     }
     // write payload len
     
     if(write(fd,&msg->header.size, sizeof(msg->header.size)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to write msg payload length: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to write msg payload length: %s", strerror(errno));
         return -1;
     }
     // write payload data
@@ -182,14 +182,14 @@ int msg_write(int fd, tunnel_msg_t* msg)
     {
         if(write(fd,msg->data, msg->header.size) == -1)
         {
-            ERROR(MODULE_NAME, "Unable to write msg payload: %s", strerror(errno));
+            M_ERROR(MODULE_NAME, "Unable to write msg payload: %s", strerror(errno));
             return -1;
         }
     }
     number = MSG_MAGIC_END;
     if(write(fd,&number, sizeof(number)) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to write end magic number: %s", strerror(errno));
+        M_ERROR(MODULE_NAME, "Unable to write end magic number: %s", strerror(errno));
         return -1;
     }
     return 0;

tunnel.h

@@ -16,6 +16,7 @@
 #define    CHANNEL_DATA             (uint8_t)0x6
 #define    CHANNEL_UNSUBSCRIBE      (uint8_t)0x3
 #define    CHANNEL_SUBSCRIBE        (uint8_t)0x2
+#define    CHANNEL_CTRL             (uint8_t)0x7
 
 typedef struct {
     uint8_t type;

vterm/vterm.c

@@ -5,26 +5,308 @@
 #include <sys/socket.h>
 #include <signal.h>
 #include <errno.h>
+#include <fcntl.h>
+#include <termios.h>
+#include <sys/select.h>
+#include <sys/ioctl.h>
+#include <sys/wait.h>
+
+#include <antd/list.h>
+#include <antd/bst.h>
+
 #include "tunnel.h"
 
 #define MODULE_NAME     "vterm"
 
+typedef struct{
+	int     fdm;
+	pid_t   pid;
+	int     cid;
+} vterm_proc_t;
+
+static bst_node_t* processes = NULL;
+
 static volatile int running = 1;
 
-void int_handler(int dummy) {
+static void int_handler(int dummy) {
     (void) dummy;
     running = 0;
 }
 
+static vterm_proc_t* terminal_new(void)
+{
+    int fdm, fds, rc;
+    pid_t pid;
+    vterm_proc_t* proc = NULL;
+    // Check arguments
+    fdm = posix_openpt(O_RDWR);
+    if (fdm < 0)
+    {
+        M_LOG(MODULE_NAME, "Error on posix_openpt(): %s\n", strerror(errno));
+        return NULL;
+    }
+    
+    rc = grantpt(fdm);
+    if (rc != 0)
+    {
+        M_LOG(MODULE_NAME, "Error on grantpt(): %s\n", strerror(errno));
+        return NULL;
+    }
+    rc = unlockpt(fdm);
+    if (rc != 0)
+    {
+        M_LOG(MODULE_NAME, "Error on unlockpt(): %s\n", strerror(errno));
+        return NULL;
+    }
+    
+    // Open the slave side ot the PTY
+    fds = open(ptsname(fdm), O_RDWR);
+    
+    // Create the child process
+    pid = fork();
+    if (pid)
+    {
+        // parent
+        proc = (vterm_proc_t*)malloc(sizeof(vterm_proc_t));
+        proc->fdm = fdm;
+        proc->pid = pid;
+        return proc;
+    }
+    else
+    {
+        //struct termios slave_orig_term_settings; // Saved terminal settings
+        //struct termios new_term_settings; // Current terminal settings
+        
+        // CHILD
+        
+        // Close the master side of the PTY
+        close(fdm);
+        
+        // Save the defaults parameters of the slave side of the PTY
+        //rc = tcgetattr(fds, &slave_orig_term_settings);
+        
+        // Set RAW mode on slave side of PTY
+        //new_term_settings = slave_orig_term_settings;
+        //cfmakeraw (&new_term_settings);
+        //tcsetattr (fds, TCSANOW, &new_term_settings);
+        
+        // The slave side of the PTY becomes the standard input and outputs of the child process
+        // we use cook mode here
+        close(0); // Close standard input (current terminal)
+        close(1); // Close standard output (current terminal)
+        close(2); // Close standard error (current terminal)
+        
+        rc = dup(fds); // PTY becomes standard input (0)
+        rc = dup(fds); // PTY becomes standard output (1)
+        rc = dup(fds); // PTY becomes standard error (2)
+        
+        // Now the original file descriptor is useless
+        close(fds);
+        
+        // Make the current process a new session leader
+        setsid();
+        
+        // As the child is a session leader, set the controlling terminal to be the slave side of the PTY
+        // (Mandatory for programs like the shell to make them manage correctly their outputs)
+        ioctl(0, TIOCSCTTY, 1);
+        
+        //system("/bin/bash");
+        rc = system("TERM=linux login");
+        //M_LOG("%s\n","Terminal exit");
+        _exit(1);
+    }
+}
+
+static void terminal_kill(int client_id, int should_delete)
+{
+    // find the proc
+    bst_node_t* node = bst_find(processes, client_id);
+    vterm_proc_t* proc;
+    if(node != NULL)
+    {
+        proc = (vterm_proc_t*)node->data;
+        if(proc != NULL)
+        {
+            (void) close(proc->fdm);
+            M_LOG(MODULE_NAME, "Kill the process %d", proc->pid);
+            if(kill(proc->pid, SIGKILL) == - 1)
+            {
+                M_ERROR(MODULE_NAME, "Unable to kill process %d: %s", proc->pid, strerror(errno));
+            }
+            free(node->data);
+            if(should_delete)
+                processes = bst_delete(processes, node->key);
+        }
+    }
+}
+
+static int terminal_write(tunnel_msg_t* msg)
+{
+    // TODO: control frame e.g. for window resize
+     bst_node_t* node = bst_find(processes, msg->header.client_id);
+    vterm_proc_t* proc;
+    if(node != NULL)
+    {
+        proc = (vterm_proc_t*)node->data;
+        if(proc != NULL)
+        {
+            if(write(proc->fdm, msg->data, msg->header.size) == -1)
+            {
+                M_ERROR(MODULE_NAME, "Unable to write data to the terminal corresponding to client %d", msg->header.client_id);
+                return -1;
+            }
+        }
+        else
+        {
+            M_ERROR(MODULE_NAME, "Unable to find the process linked to client %d", msg->header.client_id);
+            return -1;
+        }
+    }
+    else
+    {
+        M_ERROR(MODULE_NAME, "Unable to find the process from processes list for %d", msg->header.client_id);
+        return -1;
+    }
+    return 0;
+}
+
+static void unsubscribe(bst_node_t* node, void** args, int argc)
+{
+    (void) argc;
+    tunnel_msg_t msg;
+    int* ufd = (int*) args[0];
+    vterm_proc_t* proc = (vterm_proc_t*) node->data;
+    if(proc != NULL)
+    {
+        msg.header.type = CHANNEL_UNSUBSCRIBE;
+		msg.header.client_id = proc->cid;
+		msg.header.size = 0;
+		terminal_kill(proc->cid, 0);
+		if(msg_write(*ufd, &msg) == -1)
+		{
+		    M_ERROR(MODULE_NAME, "Unable to request unsubscribe to client %d", proc->cid);
+		}
+    }
+}
+
+static void set_sock_fd(bst_node_t* node, void** args, int argc)
+{
+    (void) argc;
+    tunnel_msg_t msg;
+    pid_t wpid;
+    fd_set* fd_in = (fd_set*) args[1];
+    int* max_fd = (int*)args[2];
+    list_t* list_p = (list_t*) args[3];
+    int* ufd = (int*) args[0];
+    
+    vterm_proc_t* proc = (vterm_proc_t*) node->data;
+     
+    if(proc != NULL)
+    {
+        // monitor the pid
+        wpid = waitpid(proc->pid, NULL, WNOHANG);
+    	if(wpid == -1 || wpid > 0)
+    	{
+    		// child exits
+    		M_LOG(MODULE_NAME, "Terminal linked to client %d exits\n", proc->cid);
+    		unsubscribe(node, args, argc);
+    		list_put_ptr(list_p, node);
+    	}
+    	else
+    	{
+    	    FD_SET(proc->fdm, fd_in);
+            if(*max_fd < proc->fdm)
+            {
+                *max_fd = proc->fdm;
+            }
+    	}
+    }
+}
+
+static void terminal_monitor(bst_node_t* node, void** args, int argc)
+{
+    (void) argc;
+    int* ufd = (int*) args[0];
+    fd_set* fd_in = (fd_set*) args[1];
+    list_t* list = (list_t*) args[3];
+    char buff[BUFFLEN];
+    tunnel_msg_t msg;
+    int rc;
+    vterm_proc_t* proc = (vterm_proc_t*) node->data;
+     
+    if(proc != NULL && FD_ISSET(proc->fdm, fd_in))
+    {
+        if ((rc = read(proc->fdm, buff, BUFFLEN)) > 0)
+        {
+            // Send data to client
+            msg.header.client_id = node->key;
+            msg.header.type = CHANNEL_DATA;
+            msg.header.size = rc;
+            msg.data = buff;
+            if(msg_write(*ufd, &msg) == -1)
+            {
+                terminal_kill(node->key, 0);
+                M_ERROR(MODULE_NAME,"Unable to send data to client %d", msg.header.client_id);
+                list_put_ptr(list, node);
+            }
+        }
+        else
+        {
+            if (rc < 0)
+            {
+            	M_LOG(MODULE_NAME, "Error on read standard input: %s\n", strerror(errno));
+            	terminal_kill(node->key, 0);
+            	list_put_ptr(list, node);
+            }
+        }
+    }
+}
+
+static void terminal_resize(int cid, int col, int row)
+{
+    struct winsize win = {0, 0, 0, 0};
+    bst_node_t* node = bst_find(processes, cid);
+    vterm_proc_t* proc;
+    if(node != NULL)
+    {
+        proc = (vterm_proc_t*) node->data;
+        if (ioctl(proc->fdm, TIOCGWINSZ, &win) != 0)
+        {
+            if (errno != EINVAL)
+            {
+                M_ERROR(MODULE_NAME, "Unable to get terminal winsize setting: %s", strerror(errno));
+                return;
+            }
+            memset(&win, 0, sizeof(win));
+        }
+        //printf("Setting winsize\n");
+        if (row >= 0)
+            win.ws_row = (unsigned short)row;
+        if (col >= 0)
+            win.ws_col = (unsigned short)col;
+
+        if (ioctl(proc->fdm, TIOCSWINSZ, (char *)&win) != 0)
+            M_ERROR(MODULE_NAME, "Unable to set terminal window size process linked to client %d: %s", cid, strerror(errno));
+        }
+    else
+    {
+        M_ERROR(MODULE_NAME, "Unable to find the terminal process linked to client %d", cid);
+    }
+}
 
 int main(int argc, char** argv)
 {
     int fd;
     tunnel_msg_t msg;
     fd_set fd_in;
-    int status;
+    int status, maxfd;
     struct timeval timeout;
     char buff[MAX_CHANNEL_NAME+1];
+    void *args[4];
+    list_t list;
+    item_t item;
+    int ncol, nrow;
+    
     LOG_INIT(MODULE_NAME);
     if(argc != 2)
     {
@@ -32,38 +314,38 @@ int main(int argc, char** argv)
         return -1;
     }
     signal(SIGINT, int_handler);
-    LOG(MODULE_NAME, "Hotline is: %s", argv[1]);
+    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)
     {
-        ERROR(MODULE_NAME, "Unable to open the hotline: %s", argv[1]);
+        M_ERROR(MODULE_NAME, "Unable to open the hotline: %s", argv[1]);
         return -1;
     }
     msg.header.type = CHANNEL_OPEN;
     msg.header.channel_id = 0;
     msg.header.client_id = 0;
-    LOG(MODULE_NAME, "Request to open the channel %s", MODULE_NAME);
+    M_LOG(MODULE_NAME, "Request to open the channel %s", MODULE_NAME);
     (void)strncpy(buff, MODULE_NAME,MAX_CHANNEL_NAME);
     msg.header.size = strlen(buff);
     msg.data = (uint8_t*) buff;
     if(msg_write(fd, &msg) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to write message to hotline");
+        M_ERROR(MODULE_NAME, "Unable to write message to hotline");
         (void) close(fd);
         return -1;
     }
-    LOG(MODULE_NAME, "Wait for comfirm creation of %s", MODULE_NAME);
+    M_LOG(MODULE_NAME, "Wait for comfirm creation of %s", MODULE_NAME);
     // now wait for message
     if(msg_read(fd, &msg) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to read message from hotline");
+        M_ERROR(MODULE_NAME, "Unable to read message from hotline");
         (void) close(fd);
         return -1;
     }
     if(msg.header.type == CHANNEL_OK)
     {
-        LOG(MODULE_NAME, "Channel created: %s", MODULE_NAME);
+        M_LOG(MODULE_NAME, "Channel created: %s", MODULE_NAME);
         if(msg.data)
             free(msg.data);
     }
@@ -75,12 +357,28 @@ int main(int argc, char** argv)
 	    timeout.tv_usec = 500;
         FD_ZERO(&fd_in);
         FD_SET(fd, &fd_in);
-        status = select(fd + 1, &fd_in, NULL, NULL, &timeout);
+        maxfd = fd;
+        
+        // monitor processes
+        list = list_init();
+        args[1] = (void*) &fd_in;
+        args[2] = (void*) &maxfd;
+        args[3] = (void*) &list;
+        args[0] = (void*) &fd;
+        bst_for_each(processes, set_sock_fd, args, 4);
+        list_for_each(item, list)
+        {
+            processes = bst_delete(processes, ((bst_node_t*)(item->value.ptr))->key);
+            item->value.ptr = NULL;
+        }
+        list_free(&list);
+        
+        status = select(maxfd + 1, &fd_in, NULL, NULL, &timeout);
         
         switch (status)
 	    {
             case -1:
-                LOG(MODULE_NAME, "Error %d on select()\n", errno);
+                M_LOG(MODULE_NAME, "Error %d on select()\n", errno);
                 running = 0;
                 break;
             case 0:
@@ -90,47 +388,118 @@ int main(int argc, char** argv)
                 break;
             // we have data
             default:
-                if(msg_read(fd, &msg) == -1)
+                if (FD_ISSET(fd, &fd_in))
                 {
-                    ERROR(MODULE_NAME, "Unable to read message from channel. quit");
+                    if(msg_read(fd, &msg) == -1)
-                    (void) close(fd);
+                    {
-                    running = 0;
+                        M_ERROR(MODULE_NAME, "Unable to read message from channel. quit");
+                        (void) close(fd);
+                        running = 0;
+                    }
+                    else
+                    {
+                        switch (msg.header.type)
+                        {
+                        case CHANNEL_SUBSCRIBE:
+                            M_LOG(MODULE_NAME, "Client %d subscribes to the chanel", msg.header.client_id);
+                            // create new process
+                            vterm_proc_t* proc = terminal_new();
+                            if(proc == NULL)
+                            {
+                                M_ERROR(MODULE_NAME, "Unable to create new terminal for client %d", msg.header.client_id);
+                                // unsubscribe client
+                                msg.header.type = CHANNEL_UNSUBSCRIBE;
+                                msg.header.size = 0;
+                                if(msg_write(fd, &msg) == -1)
+                                {
+                                    M_LOG(MODULE_NAME,"Unable to request unsubscribe client %d", msg.header.client_id);
+                                }
+                            }
+                            else
+                            {
+                                proc->cid = msg.header.client_id;
+                                // insert new terminal to the list
+                                processes = bst_insert(processes, msg.header.client_id, proc);
+                            }
+                            break;
+                        
+                        case CHANNEL_UNSUBSCRIBE:
+                            M_LOG(MODULE_NAME, "Client %d unsubscribes to the chanel", msg.header.client_id);
+                            terminal_kill(msg.header.client_id, 1);
+                            break;
+                        
+                        case CHANNEL_CTRL:
+                            if(msg.header.size == 8)
+                            {
+                                (void)memcpy(&ncol, msg.data, sizeof(ncol));
+                                (void)memcpy(&nrow, msg.data + sizeof(ncol), sizeof(nrow));
+                                M_LOG(MODULE_NAME, "Client %d request terminal window resize of (%d,%d)", msg.header.client_id, ncol, nrow);
+                                terminal_resize(msg.header.client_id, ncol, nrow);
+
+                            }
+                            else
+                            {
+                                M_ERROR(MODULE_NAME, "Invalid control message size: %d from client %d, expected 8", msg.header.size, msg.header.client_id);
+                            }
+
+                            break;
+
+                        case CHANNEL_DATA:
+                            if(terminal_write(&msg) == -1)
+                            {
+                                M_ERROR(MODULE_NAME, "Unable to write data to terminal corresponding to client %d", msg.header.client_id);
+                                terminal_kill(msg.header.client_id, 1);
+                                msg.header.type = CHANNEL_UNSUBSCRIBE;
+                                msg.header.size = 0;
+                                if(msg_write(fd, &msg) == -1)
+                                {
+                                    M_LOG(MODULE_NAME,"Unable to request unsubscribe client %d", msg.header.client_id);
+                                }
+                            }
+                            break;
+                        
+                        default:
+                            M_LOG(MODULE_NAME, "Client %d send message of type %d",
+                                msg.header.client_id, msg.header.type);
+                            break;
+                        }
+                        if(msg.data)
+                        {
+                            free(msg.data);
+                        }
+                    }
                 }
                 else
                 {
-                    switch (msg.header.type)
+                    // on the processes side
+                    list = list_init();
+                    bst_for_each(processes, terminal_monitor, args, 4);
+                    list_for_each(item, list)
                     {
-                    case CHANNEL_SUBSCRIBE:
+                        processes = bst_delete(processes, ((bst_node_t*)(item->value.ptr))->key);
-                        LOG(MODULE_NAME, "Client %d subscribes to the chanel", msg.header.client_id);
+                        item->value.ptr = NULL;
-                        break;
-                    
-                    case CHANNEL_UNSUBSCRIBE:
-                        LOG(MODULE_NAME, "Client %d unsubscribes to the chanel", msg.header.client_id);
-                        break;
-                    case CHANNEL_DATA:
-                        LOG(MODULE_NAME, "Got data");
-                        if(msg_write(fd, &msg) == -1)
-                        {
-                            LOG(MODULE_NAME,"Unable to write data back");
-                        }
-                        break;
-                    default:
-                        LOG(MODULE_NAME, "Client %d send message of type %d", msg.header.client_id, msg.header.type);
-                        break;
                     }
+                    list_free(&list);
                 }
+        
         }
     }
 
+    // unsubscribe all clients
+    args[0] = (void*) &fd;
+    bst_for_each(processes, unsubscribe, args, 1);
+    (void)bst_free(processes);
     // close the channel
-    LOG(MODULE_NAME, "Close the channel %s (%d)", MODULE_NAME, fd);
+    M_LOG(MODULE_NAME, "Close the channel %s (%d)", MODULE_NAME, fd);
     msg.header.type = CHANNEL_CLOSE;
     msg.header.size = 0;
     msg.data = NULL;
     if( msg_write(fd, &msg) == -1)
     {
-        ERROR(MODULE_NAME, "Unable to request channel close");
+        M_ERROR(MODULE_NAME, "Unable to request channel close");
     }
+    // close all opened terminal
+    
     (void)msg_read(fd, &msg);
     (void) close(fd);
     return 0;