pub, sub, quit + print subscriber

lib/process.c

@@ -37,7 +37,7 @@ void srv_process_free (struct process * p)
     free (p);
 }
 
-void process_print (struct process *p)
+void srv_process_print (struct process *p)
 {
     printf ("process %d : index %d\n", p->pid, p->index);
 }

lib/process.h

@@ -31,6 +31,6 @@ int srv_process_eq (const struct process *p1, const struct process *p2);
 void srv_process_gen (struct process *p
         , pid_t pid, unsigned int index, unsigned int version);
 
-void process_print (struct process *);
+void srv_process_print (struct process *);
 
 #endif

lib/pubsubd.c

@@ -27,6 +27,22 @@ pubsubd_channels_del (struct channels *chans, struct channel *c)
     }
 }
 
+void pubsubd_channels_del_all (struct channels *chans)
+{
+    if (!chans)
+        return;
+
+    struct channel *c;
+
+    while (!LIST_EMPTY(chans)) {
+        c = LIST_FIRST(chans);
+        LIST_REMOVE(c, entries);
+        pubsubd_channel_free (c);
+        free (c);
+        c = NULL;
+    }
+}
+
 struct channel * pubsubd_channel_copy (struct channel *c)
 {
     struct channel *copy;
@@ -59,6 +75,15 @@ pubsubd_channel_eq (const struct channel *c1, const struct channel *c2)
 
 void pubsubd_subscriber_init (struct app_list_head *chans) { LIST_INIT(chans); } 
 
+void pubsubd_app_list_elm_print (const struct app_list_elm *ale)
+{
+    printf ( "app_list_elm\n\t");
+    srv_process_print (ale->p);
+
+    printf ( "\tchan : %s\n", ale->chan);
+    printf ( "\taction : %d\n", (int) ale->action);
+}
+
 struct app_list_elm * pubsubd_app_list_elm_copy (const struct app_list_elm *ale)
 {
     if (ale == NULL)
@@ -218,8 +243,11 @@ int pubsubd_get_new_process (struct service *srv, struct app_list_elm *ale)
                          else if (strncmp("sub", token, 3) == 0) {
                              ale->action = 1;
                          }
-                         else {
-                             ale->action = 2; // both
+                         else if (strncmp("both", token, 4) == 0) {
+                             ale->action = 2;
+                         }
+                         else { // everything else is about killing the service
+                             ale->action = 3;
                          }
                      }
         }

lib/pubsubd.h

@@ -49,6 +49,8 @@ struct channel {
 void pubsubd_channels_init (struct channels *chans);
 struct channel * pubsubd_channel_copy (struct channel *c);
 struct channel * pubsubd_channel_get (struct channels *chans, struct channel *c);
+void pubsubd_channels_del (struct channels *chans, struct channel *c);
+void pubsubd_channels_del_all (struct channels *chans);
 
 void pubsubd_channel_free (struct channel *c);
 int pubsubd_channel_eq (const struct channel *c1, const struct channel *c2);
@@ -63,7 +65,7 @@ struct app_list_elm {
     struct process *p;
     char chan[BUFSIZ];
     size_t chanlen;
-    char action; // 0 : pub, 1 : sub, 2 : both
+    char action; // 0 : pub, 1 : sub, 2 : both, kill the service : 3
     LIST_ENTRY(app_list_elm) entries;
 };
 
@@ -79,5 +81,6 @@ void pubsubd_subscriber_del (struct app_list_head *al, struct app_list_elm *p);
 struct app_list_elm * pubsubd_app_list_elm_copy (const struct app_list_elm *ale);
 void pubsubd_app_list_elm_create (struct app_list_elm *ale, struct process *p);
 void pubsubd_app_list_elm_free (struct app_list_elm *todel);
+void pubsubd_app_list_elm_print (const struct app_list_elm *ale);
 
 #endif

pingpong/pingpong.c

@@ -28,7 +28,7 @@ void main_loop (const struct service *srv)
         }
 
         // printf ("before print\n");
-        process_print (&proc);
+        srv_process_print (&proc);
         // printf ("after print\n");
 
         // about the message

pubsub/pubsubd.c

@@ -26,14 +26,28 @@ main(int argc, char* argv[])
         // for each new process
         struct app_list_elm ale;
         pubsubd_get_new_process (&srv, &ale);
-        process_print (ale.p);
+        pubsubd_app_list_elm_print (&ale);
+
+        // stop the application ? (action 3)
+        if (ale.action == 3) {
+            pubsubd_channels_del_all (&chans);
+            printf ("Quitting ...\n");
+            exit (0);
+        }
+
+        // add the chan to the list
+
+        // each chan has a list of subscribers
+        // someone who only push a msg doesn't need to be registered
+
+        // 
 
         // TODO thread to handle multiple clients at a time
     }
 
     // the application will shut down, and remove the service named pipe
     if (srv_close (&srv))
-        ohshit(1, "service_close error");
+        ohshit (1, "service_close error");
 
     return EXIT_SUCCESS;
 }