TT#170203 support filter function when fetching records

This allows us to fetch only specific records that we're interested in,
based on context, instead of all of them and then having to do a second
pass over them.

Change-Id: I5e314fa633f57c79db85476e347a3305b5f585e9

mediator.c

@@ -352,12 +352,12 @@ int main(int argc, char **argv)
             gettimeofday(&tv_start, NULL);
 #endif
 
-            int ret = medmysql_fetch_records(mysql_callid, &acc_records, 1);
+            int ret = medmysql_fetch_records(mysql_callid, &acc_records, 1, NULL, NULL);
             if (ret < 0)
                 goto out;
             int must_sort = (ret > 0);
 
-            ret = medredis_fetch_records(mysql_callid, &acc_records);
+            ret = medredis_fetch_records(mysql_callid, &acc_records, NULL, NULL);
             if (ret > 0)
                 must_sort = 1;
 
@@ -402,10 +402,10 @@ int main(int argc, char **argv)
             gettimeofday(&tv_start, NULL);
 #endif
 
-            if(medredis_fetch_records(redis_callid, &acc_records) < 0)
+            if(medredis_fetch_records(redis_callid, &acc_records, NULL, NULL) < 0)
                 goto out;
 
-            medmysql_fetch_records(redis_callid, &acc_records, 0);
+            medmysql_fetch_records(redis_callid, &acc_records, 0, NULL, NULL);
 
             // always sort records from Redis, regardless of whether records from MySQL were merged
             records_sort(&acc_records);

mediator.h

@@ -100,6 +100,8 @@ typedef struct {
     time_t created;
 } med_cache_entry_t;
 
+typedef gboolean (*records_filter_func)(med_entry_t *, void *data);
+
 extern GHashTable *med_peer_host_table;
 extern GHashTable *med_peer_ip_table;
 extern GHashTable *med_peer_id_table;

medmysql.c

@@ -673,7 +673,8 @@ gboolean medmysql_fetch_callids(GQueue *output)
 
 /**********************************************************************/
 int medmysql_fetch_records(char *callid,
-        GQueue *entries, int warn_empty)
+        GQueue *entries, int warn_empty,
+        records_filter_func filter, void *filter_data)
 {
     MYSQL_RES *res;
     MYSQL_ROW row;
@@ -734,7 +735,10 @@ int medmysql_fetch_records(char *callid,
 
         cdr_parse_entry(e);
 
-        g_queue_push_tail(entries, e);
+        if (!filter || filter(e, filter_data))
+            g_queue_push_tail(entries, e);
+        else
+            med_entry_free(e);
 
         if (check_shutdown())
             return -1;

medmysql.h

@@ -53,7 +53,7 @@ struct medmysql_call_stat_info_t {
 int medmysql_init(void);
 void medmysql_cleanup(void);
 gboolean medmysql_fetch_callids(GQueue *output);
-int medmysql_fetch_records(char *callid, GQueue *entries, int warn_empty);
+int medmysql_fetch_records(char *callid, GQueue *entries, int warn_empty, records_filter_func, void *filter_data);
 int medmysql_trash_entries(const char *callid, struct medmysql_batches *);
 int medmysql_backup_entries(const char *callid, struct medmysql_batches *);
 int medmysql_delete_entries(const char *callid, struct medmysql_batches *);

medredis.c

@@ -643,7 +643,9 @@ static void medredis_free_keys_list(gpointer data) {
 
 /**********************************************************************/
 int medredis_fetch_records(char *callid,
-        GQueue *entries) {
+        GQueue *entries,
+        records_filter_func filter, void *filter_data)
+{
 
 
     /*
@@ -746,8 +748,14 @@ int medredis_fetch_records(char *callid,
             continue;
         }
         medredis_free_reply(&reply);
-        g_queue_push_tail(entries, e);
-        ret = 1;
+
+        if (!filter || filter(e, filter_data))
+        {
+            g_queue_push_tail(entries, e);
+            ret = 1;
+        }
+        else
+            med_entry_free(e);
     }
 
     g_list_free_full(keys, medredis_free_keys_list);

medredis.h

@@ -9,7 +9,7 @@
 int medredis_init(void);
 void medredis_cleanup(void);
 gboolean medredis_fetch_callids(GQueue *output);
-int medredis_fetch_records(char *callid, GQueue *entries);
+int medredis_fetch_records(char *callid, GQueue *entries, records_filter_func, void *filter_data);
 int medredis_trash_entries(GQueue *records);
 int medredis_backup_entries(GQueue *records);