NNTPGrab

/* 

    Copyright (C) 2005-2012  Erik van Pienbroek

    This program is free software; you can redistribute it and/or modify

    it under the terms of the GNU General Public License as published by

    the Free Software Foundation; either version 2 of the License, or

    (at your option) any later version.

    This program is distributed in the hope that it will be useful,

    but WITHOUT ANY WARRANTY; without even the implied warranty of

    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License

    along with this program; if not, write to the Free Software

    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

*/

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include 

#include 

#include 

#include 

#include 

#include "nntpgrab.h"

#include "nntpgrab_utils.h"

#include "nntpgrab_internal.h"

#include "nntpgrab_glue.h"

#include "nntpgrab_glue_internal.h"

#include "json_object.h"

#include "json.h"

#include "jsonrpc.h"

static void process_foreach_collection_event(const char *msg, struct json_object *params);

static void process_foreach_file_event(const char *msg, struct json_object *params);

static void process_foreach_group_event(const char *msg, struct json_object *params);

static struct json_object *

get_json_param(const char *msg, struct json_object *params, const char *field, const char *event_name, enum json_type type)

{

    struct json_object *obj;

    obj = json_object_object_get(params, (char*) field);

    if (!obj) {

        g_print(__FILE__ ":%i field '%s' is missing for event or method '%s', msg = %s\n", __LINE__, field, event_name, msg);

        return NULL;

    }

    if (json_object_get_type(obj) != type) {

        g_print(__FILE__ ":%i field '%s' is of invalid type for event or method '%s', msg = %s\n", __LINE__, field, event_name, msg);

        return NULL;

    }

    return obj;

}

#define get_param(msg, params, field, event_name, type, dest)                                           \

    {                                                                                                   \

        struct json_object *obj = get_json_param(msg, params, field, event_name, json_type_ ## type);   \

        if (!obj) {                                                                                     \

            return;                                                                                     \

        }                                                                                               \

        dest = json_object_get_ ## type (obj);                                                          \

    }

static void

emit_part_download_start(const char *msg, struct json_object *params)

{

    char *servername;

    int conn_id;

    char *collection_name;

    char *subject;

    int part_num;

    get_param(msg, params, "servername", "part_download_start", string, servername);

    get_param(msg, params, "conn_id", "part_download_start", int, conn_id);

    get_param(msg, params, "collection_name", "part_download_start", string, collection_name);

    get_param(msg, params, "subject", "part_download_start", string, subject);

    get_param(msg, params, "part_num", "part_download_start", int, part_num);

    nntpgrab_core_emit_part_download_start(FALSE, servername, conn_id, collection_name, subject, part_num);

}

static void

emit_part_done(const char *msg, struct json_object *params)

{

    char *servername;

    int conn_id;

    char *collection_name;

    char *subject;

    int part_num;

    int size;

    get_param(msg, params, "servername", "part_done", string, servername);

    get_param(msg, params, "conn_id", "part_done", int, conn_id);

    get_param(msg, params, "collection_name", "part_done", string, collection_name);

    get_param(msg, params, "subject", "part_done", string, subject);

    get_param(msg, params, "part_num", "part_done", int, part_num);

    get_param(msg, params, "size", "part_done", int, size);

    nntpgrab_core_emit_part_done(FALSE, servername, conn_id, collection_name, subject, part_num, size);

}

static void

emit_part_failed(const char *msg, struct json_object *params)

{

    char *servername;

    int conn_id;

    char *collection_name;

    char *subject;

    int part_num;

    int size;

    gboolean all_servers_tried;

    get_param(msg, params, "servername", "part_done", string, servername);

    get_param(msg, params, "conn_id", "part_done", int, conn_id);

    get_param(msg, params, "collection_name", "part_done", string, collection_name);

    get_param(msg, params, "subject", "part_done", string, subject);

    get_param(msg, params, "part_num", "part_done", int, part_num);

    get_param(msg, params, "size", "part_done", int, size);

    get_param(msg, params, "all_servers_tried", "part_done", boolean, all_servers_tried);

    nntpgrab_core_emit_part_failed(FALSE, servername, conn_id, collection_name, subject, part_num, size, all_servers_tried);

}

static void

emit_traffic_monitor_update(const char *msg, struct json_object *params)

{

    int bytes_received[10];

    int stamp;

    get_param(msg, params, "bytes_received1",  "traffic_monitor_update", int, bytes_received[0]);

    get_param(msg, params, "bytes_received2",  "traffic_monitor_update", int, bytes_received[1]);

    get_param(msg, params, "bytes_received3",  "traffic_monitor_update", int, bytes_received[2]);

    get_param(msg, params, "bytes_received4",  "traffic_monitor_update", int, bytes_received[3]);

    get_param(msg, params, "bytes_received5",  "traffic_monitor_update", int, bytes_received[4]);

    get_param(msg, params, "bytes_received6",  "traffic_monitor_update", int, bytes_received[5]);

    get_param(msg, params, "bytes_received7",  "traffic_monitor_update", int, bytes_received[6]);

    get_param(msg, params, "bytes_received8",  "traffic_monitor_update", int, bytes_received[7]);

    get_param(msg, params, "bytes_received9",  "traffic_monitor_update", int, bytes_received[8]);

    get_param(msg, params, "bytes_received10", "traffic_monitor_update", int, bytes_received[9]);

    get_param(msg, params, "stamp", "traffic_monitor_update", int, stamp);

    nntpgrab_core_emit_traffic_monitor_update(FALSE, bytes_received, stamp);

}

static void

emit_part_progress_update(const char *msg, struct json_object *params)

{

    char *servername;

    int conn_id;

    char *collection_name;

    char *subject;

    int part_num;

    int bytes_downloaded;

    int bytes_total;

    get_param(msg, params, "servername", "part_progress_update", string, servername);

    get_param(msg, params, "conn_id", "part_progress_update", int, conn_id);

    get_param(msg, params, "collection_name", "part_progress_update", string, collection_name);

    get_param(msg, params, "subject", "part_progress_update", string, subject);

    get_param(msg, params, "part_num", "part_progress_update", int, part_num);

    get_param(msg, params, "bytes_downloaded", "part_progress_update", int, bytes_downloaded);

    get_param(msg, params, "bytes_total", "part_progress_update", int, bytes_total);

    nntpgrab_core_emit_part_progress_update(FALSE, servername, conn_id, collection_name, subject, part_num, bytes_downloaded, bytes_total);

}

static void

emit_collection_added(const char *msg, struct json_object *params)

{

    char *collection_name;

    char *poster;

    get_param(msg, params, "collection_name", "collection_added", string, collection_name);

    get_param(msg, params, "poster", "collection_added", string, poster);

    nntpgrab_core_emit_collection_added(FALSE, collection_name, poster);

}

static void

emit_collection_removed(const char *msg, struct json_object *params)

{

    char *collection_name;

    get_param(msg, params, "collection_name", "collection_removed", string, collection_name);

    nntpgrab_core_emit_collection_removed(FALSE, collection_name);

}

static void

emit_collection_modified(const char *msg, struct json_object *params)

{

    char *collection_name;

    char *poster;

    get_param(msg, params, "collection_name", "collection_modified", string, collection_name);

    get_param(msg, params, "poster", "collection_modified", string, poster);

    nntpgrab_core_emit_collection_modified(FALSE, collection_name, poster);

}

static void

emit_file_added(const char *msg, struct json_object *params)

{

    struct json_object *obj;

    int len;

    int i;

    char *collection_name;

    char *subject;

    char *poster;

    int stamp;

    guint64 file_size;

    guint64 total_size;

    guint64 total_size_remaining;

    NGTaskState status;

    int num_parts;

    GList *groups;

    get_param(msg, params, "collection_name", "file_added", string, collection_name);

    get_param(msg, params, "subject", "file_added", string, subject);

    get_param(msg, params, "poster", "file_added", string, poster);

    get_param(msg, params, "stamp", "file_added", int, stamp);

    get_param(msg, params, "file_size", "file_added", int, file_size);

    get_param(msg, params, "total_size", "file_added", int, total_size);

    get_param(msg, params, "total_size_remaining", "file_added", int, total_size_remaining);

    get_param(msg, params, "status", "file_added", int, status);

    get_param(msg, params, "num_parts", "file_added", int, num_parts);

    obj = get_json_param(msg, params, "groups", "file_added", json_type_array);

    if (!obj) {

        return;

    }

    groups = NULL;

    len = json_object_array_length(obj);

    for (i = 0; i < len; i++) {

        struct json_object *group = json_object_array_get_idx(obj, i);

        char *groupname;

        g_return_if_fail(group != NULL);

        groupname = json_object_get_string(group);

        g_return_if_fail(groupname != NULL);

        groups = g_list_append(groups, groupname);

    }

    nntpgrab_core_emit_file_added(FALSE, collection_name, subject, poster, stamp, file_size * 1024, total_size * 1024, total_size_remaining * 1024, status, num_parts, groups);

    g_list_free(groups);

}

static void

emit_file_removed(const char *msg, struct json_object *params)

{

    char *collection_name;

    char *subject;

    guint64 total_size;

    guint64 total_size_remaining;

    get_param(msg, params, "collection_name", "file_removed", string, collection_name);

    get_param(msg, params, "subject", "file_removed", string, subject);

    get_param(msg, params, "total_size", "file_removed", int, total_size);

    get_param(msg, params, "total_size_remaining", "file_removed", int, total_size_remaining);

    nntpgrab_core_emit_file_removed(FALSE, collection_name, subject, total_size * 1024, total_size_remaining * 1024);

}

static void

emit_file_download_state_update(const char *msg, struct json_object *params)

{

    char *collection_name;

    char *subject;

    int num_parts_total;

    int num_parts_done;

    int num_parts_failed;

    guint64 file_size;

    guint64 file_size_remaining;

    guint64 total_size;

    guint64 total_size_remaining;

    get_param(msg, params, "collection_name", "file_download_state_update", string, collection_name);

    get_param(msg, params, "subject", "file_download_state_update", string, subject);

    get_param(msg, params, "num_parts_total", "file_download_state_update", int, num_parts_total);

    get_param(msg, params, "num_parts_done", "file_download_state_update", int, num_parts_done);

    get_param(msg, params, "num_parts_failed", "file_download_state_update", int, num_parts_failed);

    get_param(msg, params, "file_size", "file_download_state_update", int, file_size);

    get_param(msg, params, "file_size_remaining", "file_download_state_update", int, file_size_remaining);

    get_param(msg, params, "total_size", "file_download_state_update", int, total_size);

    get_param(msg, params, "total_size_remaining", "file_download_state_update", int, total_size_remaining);

    nntpgrab_core_emit_file_download_state_update(FALSE, collection_name, subject, num_parts_total, num_parts_done, num_parts_failed, file_size * 1024, file_size_remaining * 1024, total_size * 1024, total_size_remaining * 1024);

}

static void

emit_file_state_changed(const char *msg, struct json_object *params)

{

    char *collection_name;

    char *subject;

    char *real_filename;

    int old_state;

    int new_state;

    guint64 file_size_remaining;

    guint64 total_size;

    guint64 total_size_remaining;

    get_param(msg, params, "collection_name", "file_state_changed", string, collection_name);

    get_param(msg, params, "subject", "file_state_changed", string, subject);

    get_param(msg, params, "real_filename", "file_state_changed", string, real_filename);

    get_param(msg, params, "old_state", "file_state_changed", int, old_state);

    get_param(msg, params, "new_state", "file_state_changed", int, new_state);

    get_param(msg, params, "file_size_remaining", "file_state_changed", int, file_size_remaining);

    get_param(msg, params, "total_size", "file_state_changed", int, total_size);

    get_param(msg, params, "total_size_remaining", "file_state_changed", int, total_size_remaining);

    nntpgrab_core_emit_file_state_changed(FALSE, collection_name, subject, real_filename, old_state, new_state, file_size_remaining * 1024, total_size * 1024, total_size_remaining * 1024);

}

static void

emit_connection_connecting(const char *msg, struct json_object *params)

{

    char *servername;

    int conn_id;

    get_param(msg, params, "servername", "connection_connecting", string, servername);

    get_param(msg, params, "conn_id", "connection_connecting", int, conn_id);

    nntpgrab_core_emit_connecting(FALSE, servername, conn_id);

}

static void

emit_connection_connected(const char *msg, struct json_object *params)

{

    char *servername;

    int conn_id;

    char *welcome_msg;

    get_param(msg, params, "servername", "connection_connected", string, servername);

    get_param(msg, params, "conn_id", "connection_connected", int, conn_id);

    get_param(msg, params, "welcome_msg", "connection_connected", string, welcome_msg);

    nntpgrab_core_emit_connected(FALSE, servername, conn_id, welcome_msg);

}

static void

emit_connection_disconnect(const char *msg, struct json_object *params)

{

    char *servername;

    int conn_id;

    NNTPDisconnectType disconnect_type;

    char *reason;

    get_param(msg, params, "servername", "connection_disconnect", string, servername);

    get_param(msg, params, "conn_id", "connection_disconnect", int, conn_id);

    get_param(msg, params, "disconnect_type", "connection_disconnect", int, disconnect_type);

    get_param(msg, params, "reason", "connection_disconnect", string, reason);

    nntpgrab_core_emit_disconnect(FALSE, servername, conn_id, disconnect_type, reason);

}

static void

emit_schedular_state_changed(const char *msg, struct json_object *params)

{

    NGSchedularState new_state;

    char *reason;

    get_param(msg, params, "new_state", "schedular_state_changed", int, new_state);

    get_param(msg, params, "reason", "schedular_state_changed", string, reason);

    nntpgrab_core_emit_schedular_state_changed(FALSE, new_state, (strlen(reason) > 0 ? reason : NULL));

}

static void

emit_log_message(const char *msg, struct json_object *params)

{

    char *component;

    NGLogLevel log_level;

    char *message;

    get_param(msg, params, "component", "log_message", string, component);

    get_param(msg, params, "log_level", "log_message", int, log_level);

    get_param(msg, params, "msg", "log_message", string, message);

    nntpgrab_core_emit_log_message(FALSE, component, log_level, message, TRUE, FALSE);

}

static void

emit_task_moved(const char *msg, struct json_object *params)

{

    char *orig_collection_name;

    char *subject;

    char *new_collection_name;

    int old_position;

    int new_position;

    get_param(msg, params, "orig_collection_name", "task_moved", string, orig_collection_name);

    get_param(msg, params, "subject", "task_moved", string, subject);

    get_param(msg, params, "new_collection_name", "task_moved", string, new_collection_name);

    get_param(msg, params, "old_position", "task_moved", int, old_position);

    get_param(msg, params, "new_position", "task_moved", int, new_position);

    nntpgrab_core_emit_file_moved(FALSE, orig_collection_name, subject, new_collection_name, old_position, new_position);

}

static void

emit_collection_moved(const char *msg, struct json_object *params)

{

    char *collection_name;

    int old_position;

    int new_position;

    get_param(msg, params, "collection_name", "collection_moved", string, collection_name);

    get_param(msg, params, "old_position", "collection_moved", int, old_position);

    get_param(msg, params, "new_position", "collection_moved", int, new_position);

    nntpgrab_core_emit_collection_moved(FALSE, collection_name, old_position, new_position);

}

static void

emit_plugin_loaded(const char *msg, struct json_object *params)

{

    char *plugin_name;

    gboolean is_persistent;

    get_param(msg, params, "plugin_name", "plugin_loaded", string, plugin_name);

    get_param(msg, params, "is_persistent", "plugin_loaded", boolean, is_persistent);

    nntpgrab_core_emit_plugin_loaded(FALSE, plugin_name, is_persistent);

}

static void

emit_plugin_unloaded(const char *msg, struct json_object *params)

{

    char *plugin_name;

    get_param(msg, params, "plugin_name", "plugin_unloaded", string, plugin_name);

    nntpgrab_core_emit_plugin_unloaded(FALSE, plugin_name);

}

static void

emit_plugin_event(const char *msg, struct json_object *params)

{

    int i;

    int len;

    struct json_object *obj;

    char *plugin_name;

    char *event_name;

    char **values;

    get_param(msg, params, "plugin_name", "plugin_event", string, plugin_name);

    get_param(msg, params, "event_name", "plugin_event", string, event_name);

    obj = get_json_param(msg, params, "values", "plugin_event", json_type_array);

    if (!obj) {

        return;

    }

    len = json_object_array_length(obj);

    values = g_slice_alloc0(sizeof(char*) * (len + 1));

    for (i = 0; i < len; i++) {

        struct json_object *val_obj = json_object_array_get_idx(obj, i);

        char *val;

        g_return_if_fail(val_obj != NULL);

        val = json_object_get_string(val_obj);

        g_return_if_fail(val != NULL);

        values[i] = val;

    }

    nntpgrab_core_emit_plugin_event(FALSE, plugin_name, event_name, (const char **) values);

    g_slice_free1(sizeof(char*) * (len + 1), values);

}

/* This function used to be part of GLib, but because it behaviour got 

 * changed in GLib 2.27.4 we can't rely on it's value anymore for

 * hardcoded hash comparisons. This is the g_str_hash implementation

 * of before GLib 2.27.4 */

static guint

ng_str_hash (gconstpointer v)

{

    const signed char *p;

    guint32 h = 0;

    for (p = v; *p != '\0'; p++)

        h = (h << 5) - h + *p;

    return h;

}

static void

glue_process_jsonrpc_notification(const char *msg, struct json_object *obj)

{

    struct json_object *params;

    struct json_object *method;

    char *event_name;

    guint hash;

    params = json_object_object_get(obj, "params");

    if (!params) {

        g_print(__FILE__ ":%i JSON message lacks a 'params' field, ignoring: %s\n", __LINE__, msg);

        return;

    }

    if (json_object_get_type(params) != json_type_object) {

        g_print(__FILE__ ":%i JSON parameter is of invalid type, ignoring: %s\n", __LINE__, msg);

        return;

    }

    method = json_object_object_get(obj, "method");

    if (!method) {

        g_print(__FILE__ ":%i JSON message lacks a 'method' field, ignoring: %s\n", __LINE__, msg);

        return;

    }

    event_name = json_object_get_string(method);

    hash = ng_str_hash(event_name);

    switch (hash) {

        case 4148964023U:       /* config_changed */

            if (has_signal_handler_pending(CONFIG_CHANGED_SIGNAL)) {

                nntpgrab_core_emit_config_changed(FALSE);

            }

            break;

        case 718712759U:        /* part_download_start */

            if (has_signal_handler_pending(PART_DOWNLOAD_START_SIGNAL)) {

                emit_part_download_start(msg, params);

            }

            break;

        case 2485450830U:       /* part_done */

            if (has_signal_handler_pending(PART_DONE_SIGNAL)) {

                emit_part_done(msg, params);

            }

            break;

        case 560621065U:        /* part_failed */

            if (has_signal_handler_pending(PART_FAILED_SIGNAL)) {

                emit_part_failed(msg, params);

            }

            break;

        case 3851882960U:       /* traffic_monitor_update */

            if (has_signal_handler_pending(TRAFFIC_MONITOR_UPDATE_SIGNAL)) {

                emit_traffic_monitor_update(msg, params);

            }

            break;

        case 3667888847U:       /* part_progress_update */

            if (has_signal_handler_pending(PART_PROGRESS_UPDATE_SIGNAL)) {

                emit_part_progress_update(msg, params);

            }

            break;

        case 2792752159U:       /* collection_added */

            if (has_signal_handler_pending(COLLECTION_ADDED_SIGNAL)) {

                emit_collection_added(msg, params);

            }

            break;

        case 1720184767U:       /* collection_removed */

            if (has_signal_handler_pending(COLLECTION_REMOVED_SIGNAL)) {

                emit_collection_removed(msg, params);

            }

            break;

        case 1683429194U:       /* collection_modified */

            if (has_signal_handler_pending(COLLECTION_MODIFIED_SIGNAL)) {

                emit_collection_modified(msg, params);

            }

            break;

        case 2127244413U:       /* file_added */

            if (has_signal_handler_pending(FILE_ADDED_SIGNAL)) {

                emit_file_added(msg, params);

            }

            break;

        case 2117367965U:       /* file_removed */

            if (has_signal_handler_pending(FILE_REMOVED_SIGNAL)) {

                emit_file_removed(msg, params);

            }

            break;

        case 2548616203U:       /* file_download_state_update */

            if (has_signal_handler_pending(FILE_DOWNLOAD_STATE_UPDATE_SIGNAL)) {

                emit_file_download_state_update(msg, params);

            }

            break;

        case 3407424387U:       /* file_state_changed */

            if (has_signal_handler_pending(FILE_STATE_CHANGED_SIGNAL)) {

                emit_file_state_changed(msg, params);

            }

            break;

        case 1506868921U:       /* connection_connecting */

            if (has_signal_handler_pending(CONNECTION_CONNECTING_SIGNAL)) {

                emit_connection_connecting(msg, params);

            }

            break;

        case 2681007848U:       /* connection_connected */

            if (has_signal_handler_pending(CONNECTION_CONNECTED_SIGNAL)) {

                emit_connection_connected(msg, params);

            }

            break;

        case 2812926109U:       /* connection_disconnect */

            if (has_signal_handler_pending(CONNECTION_DISCONNECT_SIGNAL)) {

                emit_connection_disconnect(msg, params);

            }

            break;

        case 1376609542U:       /* schedular_state_changed */

            if (has_signal_handler_pending(SCHEDULAR_STATE_CHANGED_SIGNAL)) {

                emit_schedular_state_changed(msg, params);

            }

            break;

        case 843068174U:        /* foreach_collection_event */

            process_foreach_collection_event(msg, params);

            break;

        case 3826728748U:       /* foreach_file_event */

            process_foreach_file_event(msg, params);

            break;

        case 2102350437U:       /* foreach_group_event */

            process_foreach_group_event(msg, params);

            break;

        case 693552652U:        /* log_message */

            if (has_signal_handler_pending(LOG_MESSAGE_SIGNAL)) {

                emit_log_message(msg, params);

            }

            break;

        case 1307041657U:       /* task_moved */

            if (has_signal_handler_pending(TASK_MOVED_SIGNAL)) {

                emit_task_moved(msg, params);

            }

            break;

        case 2804179410U:       /* collection_moved */

            if (has_signal_handler_pending(COLLECTION_MOVED_SIGNAL)) {

                emit_collection_moved(msg, params);

            }

            break;

        case 2405579865U:       /* all_downloads_completed */

            if (has_signal_handler_pending(ALL_DOWNLOADS_COMPLETED_SIGNAL)) {

                nntpgrab_core_emit_all_downloads_completed(FALSE);

            }

            break;

        case 3241070225U:       /* plugin_loaded */

            if (has_signal_handler_pending(PLUGIN_LOADED_SIGNAL)) {

                emit_plugin_loaded(msg, params);

            }

            break;

        case 3055047530U:       /* plugin_unloaded */

            if (has_signal_handler_pending(PLUGIN_UNLOADED_SIGNAL)) {

                emit_plugin_unloaded(msg, params);

            }

            break;

        case 2037961358U:       /* plugin_event */

            if (has_signal_handler_pending(PLUGIN_EVENT_SIGNAL)) {

                emit_plugin_event(msg, params);

            }

            break;

        default:

            g_print(__FILE__ ":%i JSON notification with unknown event name received, event_name = %s, hash = %u, msg = %s\n", __LINE__, event_name, hash, msg);

            break;

    }

}

static GAsyncQueue *command_queue = NULL;

static GStaticMutex command_mutex = G_STATIC_MUTEX_INIT;

static GStaticMutex init_mutex = G_STATIC_MUTEX_INIT;

void

glue_process_jsonrpc_msg(const char *msg)

{

    struct json_object *obj;

    struct json_object *id;

    g_return_if_fail(msg != NULL);

    obj = json_tokener_parse((char*)msg);

    if (!obj) {

        g_print(__FILE__ ":%i Invalid message received from server, ignoring: %s\n", __LINE__, msg);

        return;

    }

    id = json_object_object_get(obj, "id");

    if (!id) {

        /* Notification received! */

        glue_process_jsonrpc_notification(msg, obj);

        json_object_put(obj);

        return;

    }

    g_print(__FILE__ ":%i JSON response received for msg: %s\n", __LINE__, msg);

    g_static_mutex_lock(&init_mutex);

    if (command_queue == NULL) {

        command_queue = g_async_queue_new();

    }

    g_static_mutex_unlock(&init_mutex);

    g_async_queue_push(command_queue, obj);

    /* No need to unref here as it should be done by the thread who pop's the queue */

}

/*************************************************************************************/

/* JSON-RPC methods */

/*************************************************************************************/

#define get_method_param(params, field, method_name, type, dest)                                        \

    {                                                                                                   \

        char *msg = json_object_get_string(params);                                                     \

        struct json_object *obj = get_json_param(msg, params, field, method_name, json_type_ ## type);  \

        if (!obj) {                                                                                     \

            json_object_put(params) ;                                                                   \

            return FALSE;                                                                               \

        }                                                                                               \

        dest = json_object_get_ ## type (obj);                                                          \

    }

static struct json_object *

create_new_json_request(const char *method, struct json_object *request_params)

{

    struct json_object *request;

    static int id = 0;

    if (!request_params) {

        request_params = json_object_new_array();

    }

    request = json_object_new_object();

    json_object_object_add(request, "method", json_object_new_string((char*) method));

    json_object_object_add(request, "id", json_object_new_int(++id));

    json_object_object_add(request, "params", request_params);

    return request;

}

static struct json_object *

send_request_and_wait_for_response(NntpgrabGlue *glue, struct json_object *request, char **errmsg)

{

    struct json_object *response;

    struct json_object *result;

    struct json_object *error;

    GTimeVal tv;

    int count;

    g_return_val_if_fail(glue != NULL, NULL);

    g_return_val_if_fail(request != NULL, NULL);

    g_return_val_if_fail(errmsg != NULL, NULL);

    g_static_mutex_lock(&command_mutex);

    if (!write_line(&glue->socket, "%s\r\n", json_object_to_json_string(request))) {

        g_static_mutex_unlock(&command_mutex);

        return NULL;

    }

    g_static_mutex_lock(&init_mutex);

    if (command_queue == NULL) {

        command_queue = g_async_queue_new();

    }

    g_static_mutex_unlock(&init_mutex);

    count = 0;

    while (count < READ_TIMEOUT_VALUE) {

        g_get_current_time(&tv);

        tv.tv_sec++;

        response = g_async_queue_timed_pop(command_queue, &tv);

        if (response) {

            break;

        } else if (glue->socket.socket_id <= 0) {

            *errmsg = g_strdup_printf(__FILE__ ":%i Connection got disconnected\n", __LINE__);

            nntpgrab_core_emit_log_message(FALSE, "Glue layer", NG_LOG_LEVEL_WARNING, *errmsg, TRUE, FALSE);

            g_static_mutex_unlock(&command_mutex);

            return NULL;

        }

        count++;

    }

    g_static_mutex_unlock(&command_mutex);

    if (!response) {

        *errmsg = g_strdup_printf(__FILE__ ":%i Timed out waiting for response from server, request = %s\n", __LINE__, json_object_to_json_string(request));

        nntpgrab_core_emit_log_message(FALSE, "Glue layer", NG_LOG_LEVEL_WARNING, *errmsg, TRUE, FALSE);

        return NULL;

    }

    error = json_object_object_get(response, "error");

    if (error) {

        *errmsg = g_strdup(json_object_get_string(error));

        json_object_put(response);

        return NULL;

    }

    result = json_object_object_get(response, "result");

    if (!result) {

        *errmsg = g_strdup_printf(__FILE__ ":%i Response from server lacks a 'result' field, request = %s, response = %s\n", __LINE__, json_object_to_json_string(request), json_object_to_json_string(response));

        nntpgrab_core_emit_log_message(FALSE, "Glue layer", NG_LOG_LEVEL_WARNING, *errmsg, TRUE, FALSE);

        json_object_put(response);

        return NULL;

    }

    /* We're only interested in the result, so we add a reference to that and unref the response */

    json_object_get(result);

    json_object_put(response);

    return result;

}

void

nntpgrab_glue_internal_kill_server(NntpgrabGlue *obj)

{

    NntpgrabGlue *glue = NNTPGRAB_GLUE(obj);

    struct json_object *request;

    struct json_object *response_result;

    char *errmsg = NULL;

    request = create_new_json_request("nntpgrab_server_request_quit", NULL);

    if (!(response_result = send_request_and_wait_for_response(glue, request, &errmsg))) {

        g_print(__FILE__ ":%i nntpgrab_server_request_quit FAILED: %s\n", __LINE__, errmsg);

        g_free(errmsg);

        json_object_put(request);

        return;

    }

    json_object_put(response_result);

    json_object_put(request);

}

NGList *

nntpgrab_glue_internal_config_get_avail_servers(NntpgrabGlue *obj)

{

    NntpgrabGlue *glue = NNTPGRAB_GLUE(obj);

    struct json_object *request;

    struct json_object *response_result;

    NGList *ret = NULL;

    char *errmsg = NULL;

    int i;

    int len;

    request = create_new_json_request("nntpgrab_config_get_avail_servers", NULL);

    if (!(response_result = send_request_and_wait_for_response(glue, request, &errmsg))) {

        g_print(__FILE__ ":%i nntpgrab_config_get_avail_servers FAILED: %s\n", __LINE__, errmsg);

        g_free(errmsg);

        json_object_put(request);

        return NULL;

    }

    len = json_object_array_length(response_result);

    for (i = 0; i < len; i++) {

        struct json_object *obj = json_object_array_get_idx(response_result, i);

        char *servername;

        g_return_val_if_fail(obj != NULL, NULL);

        servername = json_object_get_string(obj);

        g_return_val_if_fail(servername != NULL, NULL);

        ret = ng_list_append(ret, g_strdup(servername));

    }

    json_object_put(request);

    json_object_put(response_result);

    return ret;

}

gboolean

nntpgrab_glue_internal_config_get_server_info(NntpgrabGlue *obj, const char *servername, NGConfigServer *ret)

{

    NntpgrabGlue *glue = NNTPGRAB_GLUE(obj);

    struct json_object *request;

    struct json_object *request_params;

    struct json_object *response_result;

    char *errmsg = NULL;

    char *servername_received;

    char *hostname;

    int port;

    char *username;

    char *password;

    int max_threads;