lib: Manage multiple flows with a single CDAP instance

You can now add multiple flows to a CDAP instance. This will simplify
sending messages to different peers (e.g. for syncing the RIB). A
request will now return an array of keys terminated by
CDAP_KEY_INVALID. Removes the enum from the CDAP proto file to just
take the opcode as an integer.

1 files changed, 37 insertions, 11 deletions

diff --git a/src/ipcpd/normal/enroll.c b/src/ipcpd/normal/enroll.c
index 680cfbba..4e510038 100644
--- a/src/ipcpd/normal/enroll.c
+++ b/src/ipcpd/normal/enroll.c
@@ -78,13 +78,20 @@ static void * enroll_handle(void * o)
                         continue;
                 }
 
-                cdap = cdap_create(conn.flow_info.fd);
+                cdap = cdap_create();
                 if (cdap == NULL) {
                         log_err("Failed to instantiate CDAP.");
                         flow_dealloc(conn.flow_info.fd);
                         continue;
                 }
 
+                if (cdap_add_flow(cdap, conn.flow_info.fd)) {
+                        log_warn("Failed to add flow to CDAP.");
+                        cdap_destroy(cdap);
+                        flow_dealloc(conn.flow_info.fd);
+                        continue;
+                }
+
                 while (!(boot_r && members_r && dif_name_r)) {
                         key = cdap_request_wait(cdap, &oc, &name, &data,
                                                 (size_t *) &len , &flags);
@@ -167,7 +174,7 @@ static void * enroll_handle(void * o)
 int enroll_boot(char * dst_name)
 {
         struct cdap * cdap;
-        cdap_key_t    key;
+        cdap_key_t *  key;
         uint8_t *     data;
         size_t        len;
         struct conn   conn;
@@ -186,31 +193,41 @@ int enroll_boot(char * dst_name)
                 return -1;
         }
 
-        cdap = cdap_create(conn.flow_info.fd);
+        cdap = cdap_create();
         if (cdap == NULL) {
                 log_err("Failed to instantiate CDAP.");
                 return -1;
         }
 
+        if (cdap_add_flow(cdap, conn.flow_info.fd)) {
+                log_warn("Failed to add flow to CDAP.");
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
+                return -1;
+        }
+
         log_dbg("Getting boot information from %s.", dst_name);
 
         clock_gettime(CLOCK_REALTIME, &t0);
 
         key = cdap_request_send(cdap, CDAP_READ, TIME_PATH, NULL, 0, 0);
-        if (key < 0) {
+        if (key == NULL) {
                 log_err("Failed to send CDAP request.");
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(cdap, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key[0], &data, &len)) {
                 log_err("Failed to get CDAP reply.");
+                free(key);
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
+        free(key);
+
         clock_gettime(CLOCK_REALTIME, &rtt);
 
         delta_t = ts_diff_ms(&t0, &rtt);
@@ -226,20 +243,23 @@ int enroll_boot(char * dst_name)
         free(data);
 
         key = cdap_request_send(cdap, CDAP_READ, boot_ro, NULL, 0, 0);
-        if (key < 0) {
+        if (key == NULL) {
                 log_err("Failed to send CDAP request.");
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(cdap, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key[0], &data, &len)) {
                 log_err("Failed to get CDAP reply.");
+                free(key);
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
+        free(key);
+
         log_dbg("Packed information received (%zu bytes).", len);
 
         if (rib_unpack(data, len, UNPACK_CREATE)) {
@@ -254,20 +274,23 @@ int enroll_boot(char * dst_name)
         log_dbg("Packed information inserted into RIB.");
 
         key = cdap_request_send(cdap, CDAP_READ, members_ro, NULL, 0, 0);
-        if (key < 0) {
+        if (key == NULL) {
                 log_err("Failed to send CDAP request.");
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(cdap, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key[0], &data, &len)) {
                 log_err("Failed to get CDAP reply.");
+                free(key);
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
+        free(key);
+
         log_dbg("Packed information received (%zu bytes).", len);
 
         if (rib_unpack(data, len, UNPACK_CREATE)) {
@@ -282,20 +305,23 @@ int enroll_boot(char * dst_name)
         log_dbg("Packed information inserted into RIB.");
 
         key = cdap_request_send(cdap, CDAP_READ, dif_ro, NULL, 0, 0);
-        if (key < 0) {
+        if (key == NULL) {
                 log_err("Failed to send CDAP request.");
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(cdap, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key[0], &data, &len)) {
                 log_err("Failed to get CDAP reply.");
+                free(key);
                 cdap_destroy(cdap);
                 flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
+        free(key);
+
         log_dbg("Packed information received (%zu bytes).", len);
 
         if (rib_unpack(data, len, UNPACK_CREATE)) {