TT#170203 unify acc record fetching

Instead of using two lists to keep the acc records (based on where they
were retrieved from) use only a single list for all acc records.

Make sure the list is only appended to in the functions doing the record
retrieval.

Since the list of acc records needs to be sorted only when records were
retrieved from Redis, change the return value of the retrieval functions
to indiciate whether this needs to be done, or -1 for error.

Change-Id: Ie61c054b430cb5d390b1f2b742c64be1df831fd4

mediator.c

@@ -185,29 +185,12 @@ static uint64_t mediator_calc_runtime(struct timeval *tv_start, struct timeval *
                (tv_start->tv_sec * 1000000 + tv_start->tv_usec)));
 }
 
-/**********************************************************************/
-// appends the entire `two` list to the end of `one`, resetting `two` to empty
-static void mediator_splice_gqueue(GQueue *one, GQueue *two) {
-    if (!two->head)
-        return;
-    if (!one->tail) {
-        *one = *two;
-        g_queue_init(two);
-        return;
-    }
-    one->tail->next = two->head;
-    two->head->prev = one->tail;
-    one->length += two->length;
-    g_queue_init(two);
-}
-
 /**********************************************************************/
 int main(int argc, char **argv)
 {
     GQueue mysql_callids = G_QUEUE_INIT;
     GQueue redis_callids = G_QUEUE_INIT;
-    GQueue mysql_records = G_QUEUE_INIT;
-    GQueue redis_records = G_QUEUE_INIT;
+    GQueue acc_records = G_QUEUE_INIT;
     uint64_t cdr_count, last_count;
     int maprefresh;
     struct medmysql_batches *batches;
@@ -332,8 +315,7 @@ int main(int argc, char **argv)
 
         g_queue_clear_full(&mysql_callids, g_free);
         g_queue_clear_full(&redis_callids, g_free);
-        g_queue_clear_full(&mysql_records, med_entry_free);
-        g_queue_clear_full(&redis_records, med_entry_free);
+        g_queue_clear_full(&acc_records, med_entry_free);
         cdr_count = 0;
         last_count = mediator_count;
 
@@ -370,30 +352,34 @@ int main(int argc, char **argv)
             gettimeofday(&tv_start, NULL);
 #endif
 
-            if(medmysql_fetch_records(mysql_callid, &mysql_records, 1) != 0)
+            int ret = medmysql_fetch_records(mysql_callid, &acc_records, 1);
+            if (ret < 0)
                 goto out;
+            int must_sort = (ret > 0);
 
-            if(medredis_fetch_records(mysql_callid, &redis_records) == 0)
-            {
-                mediator_splice_gqueue(&mysql_records, &redis_records);
+            ret = medredis_fetch_records(mysql_callid, &acc_records);
+            if (ret > 0)
+                must_sort = 1;
 
-		// only re-sort if records from Redis were added, as MySQL already does the sorting
-                records_sort(&mysql_records);
+            if (must_sort)
+            {
+                // only re-sort if records from Redis were added, as MySQL already does the sorting
+                records_sort(&acc_records);
             }
 
-            int are_records_complete = records_complete(&mysql_records);
+            int are_records_complete = records_complete(&acc_records);
 
             if (!are_records_complete && !do_intermediate)
             {
                 L_DEBUG("Found incomplete call with cid '%s', skipping...\n", mysql_callid);
-                g_queue_clear_full(&mysql_records, med_entry_free);
+                g_queue_clear_full(&acc_records, med_entry_free);
                 continue;
             }
 
-            if(cdr_process_records(&mysql_records, &cdr_count, batches, do_intermediate) != 0)
+            if(cdr_process_records(&acc_records, &cdr_count, batches, do_intermediate) != 0)
                 goto out;
 
-            g_queue_clear_full(&mysql_records, med_entry_free);
+            g_queue_clear_full(&acc_records, med_entry_free);
 
             mediator_count += cdr_count;
 
@@ -416,34 +402,31 @@ int main(int argc, char **argv)
             gettimeofday(&tv_start, NULL);
 #endif
 
-            if(medredis_fetch_records(redis_callid, &redis_records) != 0)
+            if(medredis_fetch_records(redis_callid, &acc_records) < 0)
                 goto out;
 
-            if(medmysql_fetch_records(redis_callid, &mysql_records, 0) == 0)
-            {
-                mediator_splice_gqueue(&redis_records, &mysql_records);
-            }
+            medmysql_fetch_records(redis_callid, &acc_records, 0);
 
-	    // always sort records from Redis, regardless of whether records from MySQL were merged
-            records_sort(&redis_records);
+            // always sort records from Redis, regardless of whether records from MySQL were merged
+            records_sort(&acc_records);
 
-            int are_records_complete = records_complete(&redis_records);
+            int are_records_complete = records_complete(&acc_records);
 
             if (!are_records_complete && !do_intermediate)
             {
                 L_DEBUG("Found incomplete call with cid '%s', skipping...\n", redis_callid);
-                g_queue_clear_full(&redis_records, med_entry_free);
+                g_queue_clear_full(&acc_records, med_entry_free);
                 continue;
             }
 
-            L_DEBUG("process cdr with cid '%s' and %u records\n", redis_callid, redis_records.length);
+            L_DEBUG("process cdr with cid '%s' and %u records\n", redis_callid, acc_records.length);
 
-            if (redis_records.length) {
-                if(cdr_process_records(&redis_records, &cdr_count, batches, do_intermediate) != 0) {
-                    g_queue_clear_full(&redis_records, med_entry_free);
+            if (acc_records.length) {
+                if(cdr_process_records(&acc_records, &cdr_count, batches, do_intermediate) != 0) {
+                    g_queue_clear_full(&acc_records, med_entry_free);
                     goto out;
                 }
-                g_queue_clear_full(&redis_records, med_entry_free);
+                g_queue_clear_full(&acc_records, med_entry_free);
 
                 mediator_count += cdr_count;
             }
@@ -484,6 +467,7 @@ out:
     L_INFO("Shutting down.");
     sd_notify(0, "STOPPING=1\n");
 
+    g_queue_clear_full(&acc_records, med_entry_free);
     g_queue_clear_full(&mysql_callids, g_free);
     g_queue_clear_full(&redis_callids, g_free);
     mediator_destroy_maps();

medmysql.c

@@ -715,6 +715,7 @@ int medmysql_fetch_records(char *callid,
         goto out;
     }
 
+    ret = 0;
     while((row = mysql_fetch_row(res)) != NULL)
     {
         med_entry_t *e = g_slice_alloc0(sizeof(*e));

medredis.c

@@ -681,8 +681,6 @@ int medredis_fetch_records(char *callid,
     cids[2] = g_strdup_printf("acc:cid::%s%s", callid, XFERSUFFIX);
     cids[3] = g_strdup_printf("acc:cid::%s%s%s", callid, PBXSUFFIX, XFERSUFFIX);
 
-    g_queue_clear_full(entries, med_entry_free);
-
     L_DEBUG("Fetching records from redis\n");
 
     for (i = 0; i < G_N_ELEMENTS(cids); ++i) {
@@ -729,6 +727,8 @@ int medredis_fetch_records(char *callid,
     L_DEBUG("Appending all keys to redis command\n");
     g_list_foreach(keys, medredis_append_key, NULL);
 
+    int ret = 0;
+
     for (GList *l = keys; l; l = l->next) {
         med_entry_t *e;
         char *key = (char*)l->data;
@@ -751,13 +751,14 @@ int medredis_fetch_records(char *callid,
         }
         medredis_free_reply(&reply);
         g_queue_push_tail(entries, e);
+        ret = 1;
     }
 
     g_list_free_full(keys, medredis_free_keys_list);
 
     medredis_consume_replies();
 
-    return 0;
+    return ret;
 
 
 err:
@@ -766,7 +767,6 @@ err:
     for (i = 0; i < G_N_ELEMENTS(cids); ++i) {
         g_free(cids[i]);
     }
-    g_queue_clear_full(entries, med_entry_free);
     medredis_consume_replies();
     return -1;