schedular.c

Go to the documentation of this file.

/*
 Copyright (C) 2005-2007 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
*/

#include 
#include 
#include 
#include 
#include 

#include "configuration.h"
#include "schedular.h"
#include "plugin_schedular.h"
#include "nntpgrab_plugin_schedular.h"
#include "nntpconnection.h"
#include "decoder.h"
#include "nntpgrab_internal.h"

static PluginSchedularExportedFuncs plugin_funcs;
static GModule *plugin = NULL;
static Schedular *schedular = NULL;

typedef struct SchedularClass SchedularClass;

static GType schedular_get_type (void);

static void task_added_callback(const NNTPCollection *collection, const NNTPFile *file);
static void task_removed_callback(const char *collection_name, const char *subject, guint64 total_file_size, guint64 total_file_size_remaining);
static void task_state_changed_callback(const char *collection_name, const char *subject, TaskState old_state, TaskState new_state);
static void collection_added_callback(const char *collection_name);
static void collection_removed_callback(const char *collection_name);
static GList *config_get_avail_servers(void);
static ConfigServer *config_get_server_info(const char *servername);
static ConfigOpts config_get_opts (void);
static gpointer nntp_connection_initialize (const char *servername, NNTPGrabErrCode *err);
static void nntp_connection_destroy (gpointer obj);
static NNTPGrabErrCode nntp_connection_get_part (gpointer obj, const char *collection_name, const NNTPFile *file, const NNTPPart *part, char **reason);
static gboolean decode_file (const char *collection_name, const NNTPFile *file, int *saved_errno);
static void notify_schedular_state_changed (SchedularState new_state, const char *reason);
static void emit_fatal_error (const char *errmsg);
static void emit_part_download_complete (const char *servername, gpointer connection, const char *collection_name, const char *subject, int partnum, int size);
static void emit_part_download_failed (const char *servername, gpointer connection, const char *collection_name, const char *subject, int partnum, int size, gboolean all_servers_tried);
static void emit_task_moved (const char *orig_collection_name, const char *subject, const char *new_collection_name, int old_position, int new_position);
static void emit_collection_moved (const char *collection_name, int old_position, int new_position);
static void emit_file_download_state_update (const char *collection_name, const char *subject, int num_parts_total, int num_parts_done, int num_parts_failed, guint64 file_size, guint64 file_size_remaining, guint64 total_file_size, guint64 total_file_size_remaining);
static void config_changed_callback(Configuration *config, ConfigChangedFlag flag, gpointer data);

struct Schedular
{
    GObject parent;

    GStaticRWLock rwlock;
    Configuration *config;
    ConfigOpts opts;
    Decoder *decoder;
};

struct SchedularClass
{
    GObjectClass parent;
};

G_DEFINE_TYPE(Schedular, schedular, G_TYPE_OBJECT);

static void
schedular_init (Schedular *obj)
{
    g_static_rw_lock_init(&obj->rwlock);
}

static void
schedular_finalize (GObject *obj)
{
    Schedular *schedular = SCHEDULAR(obj);

    if (plugin_funcs.destroy) {
        plugin_funcs.destroy();
    }

    g_module_close(plugin);

    g_static_rw_lock_writer_lock(&schedular->rwlock);

    g_object_unref(schedular->config);
    decoder_destroy(schedular->decoder);

    g_static_rw_lock_writer_unlock(&schedular->rwlock);

    g_static_rw_lock_free(&schedular->rwlock);
}

static void
schedular_class_init (SchedularClass *klass)
{
    GObjectClass *gobject_class = G_OBJECT_CLASS (klass);

    gobject_class->finalize = schedular_finalize;
}

Schedular *
schedular_new(Configuration *config, char **errmsg)
{
    static GStaticMutex mutex = G_STATIC_MUTEX_INIT;

    g_static_mutex_lock(&mutex);

    g_assert(schedular == NULL);

    if (!plugin) {
        char *msg;
        PluginSchedularImportedFuncs imported_funcs;
        plugin_schedular_init_func init_func;
        char *filename;

#ifdef WIN32
#define EXT "-0.dll"
#else
#define EXT ".so"
#endif

       if (g_getenv("NNTPGRAB_LIBDIR")) {
            filename = g_strdup_printf("%s/libnntpgrab_plugin_schedular%s", g_getenv("NNTPGRAB_LIBDIR"), EXT);
       } else {
            filename = g_strdup_printf("%s/libnntpgrab_plugin_schedular%s", PLUGIN_DIR, EXT);
       }

        plugin = g_module_open(filename, G_MODULE_BIND_LOCAL);

        g_free(filename);

        if (!plugin) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        imported_funcs.version = NNTPGRAB_PLUGIN_API_VERSION;
        imported_funcs.task_added = task_added_callback;
        imported_funcs.task_removed = task_removed_callback;
        imported_funcs.task_state_changed = task_state_changed_callback;
        imported_funcs.collection_added = collection_added_callback;
        imported_funcs.collection_removed = collection_removed_callback;
        imported_funcs.config_get_avail_servers = config_get_avail_servers;
        imported_funcs.config_get_server_info = config_get_server_info;
        imported_funcs.config_get_opts = config_get_opts;
        imported_funcs.nntp_connection_initialize = nntp_connection_initialize;
        imported_funcs.nntp_connection_destroy = nntp_connection_destroy;
        imported_funcs.nntp_connection_get_part = nntp_connection_get_part;
        imported_funcs.decode_file = decode_file;
        imported_funcs.notify_schedular_state_changed = notify_schedular_state_changed;
        imported_funcs.emit_fatal_error = emit_fatal_error;
        imported_funcs.emit_part_download_complete = emit_part_download_complete;
        imported_funcs.emit_part_download_failed = emit_part_download_failed;
        imported_funcs.emit_task_moved = emit_task_moved;
        imported_funcs.emit_collection_moved = emit_collection_moved;
        imported_funcs.emit_file_download_state_update = emit_file_download_state_update;

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_get_version", (gpointer *) &plugin_funcs.get_version)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        // Check if the API version matches
        if (plugin_funcs.get_version() != NNTPGRAB_PLUGIN_API_VERSION) {
            if (errmsg) {
                *errmsg = g_strdup_printf(_("Schedular Plugin API mismatch (Plugin API version = %i, expected = %i)"), plugin_funcs.get_version(), NNTPGRAB_PLUGIN_API_VERSION);
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_initialize", (gpointer *) &init_func)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_destroy", (gpointer *) &plugin_funcs.destroy)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_start", (gpointer *) &plugin_funcs.start)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_stop", (gpointer *) &plugin_funcs.stop)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_get_state", (gpointer *) &plugin_funcs.get_state)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_add_task_to_queue", (gpointer *) &plugin_funcs.add_task_to_queue)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_del_task_from_queue", (gpointer *) &plugin_funcs.del_task_from_queue)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_restart_task", (gpointer *) &plugin_funcs.restart_task)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_save_queue", (gpointer *) &plugin_funcs.save_queue)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_foreach_task", (gpointer *) &plugin_funcs.foreach_task)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_move_task", (gpointer *) &plugin_funcs.move_task)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        if (!g_module_symbol(plugin, "nntpgrab_plugin_schedular_move_collection", (gpointer *) &plugin_funcs.move_collection)) {
            if (errmsg) {
                *errmsg = g_strdup(g_module_error());
            }

            return NULL;
        }

        /* All required methods are found, lets start */
        if (!init_func(imported_funcs, &msg)) {
            if (errmsg) {
                *errmsg = g_strdup_printf(_("Schedular Plugin initialization failed!\n%s"), msg);
            }

            g_free(msg);

            return NULL;
        }

        nntpgrab_core_emit_debug_message(_("Schedular plugin initialized"));
    }

    g_object_ref(config);

    schedular = g_object_new(SCHEDULAR_TYPE_OBJECT, NULL);
    schedular->config = config;
    schedular->decoder = decoder_new(config);
    schedular->opts = configuration_get_opts(schedular->config);

    g_signal_connect(config, "config_changed", G_CALLBACK(config_changed_callback), NULL);

    g_static_mutex_unlock(&mutex);

    return schedular;
}

void
schedular_destroy(Schedular *obj)
{
    g_object_unref(obj);
}

gboolean
schedular_start(Schedular *obj)
{
    g_assert(plugin_funcs.start);

    return plugin_funcs.start();
}

gboolean
schedular_stop(Schedular *obj)
{
    g_assert(plugin_funcs.stop);

    return plugin_funcs.stop(NULL);
}

SchedularState
schedular_get_state (Schedular *obj)
{
    g_assert(plugin_funcs.get_state);

    return plugin_funcs.get_state();
}

gboolean
schedular_add_task_to_queue(Schedular *obj, const char *collection_name, const NNTPFile *file, char **errmsg)
{
    char *msg;

    g_assert(plugin_funcs.add_task_to_queue);

    msg = g_strdup_printf(_("New task added to queue, collection_name = '%s', subject = '%s'"), collection_name, file->subject);
    nntpgrab_core_emit_debug_message(msg);
    g_free(msg);

    return plugin_funcs.add_task_to_queue(collection_name, file, errmsg);
}

gboolean
schedular_del_task_from_queue(Schedular *obj, const char *collection_name, const char *subject, char **errmsg)
{
    char *msg;

    g_assert(plugin_funcs.del_task_from_queue);

    if (!subject) {
        msg = g_strdup_printf(_("Entire collection removed from queue, collection_name = '%s'"), collection_name);
    } else {
        msg = g_strdup_printf(_("Task removed from queue, collection_name = '%s', subject = '%s'"), collection_name, subject);
    }
    nntpgrab_core_emit_debug_message(msg);
    g_free(msg);

    return plugin_funcs.del_task_from_queue(collection_name, subject, errmsg);
}

gboolean
schedular_restart_task(Schedular *obj, const char *collection_name, const char *subject, char **errmsg)
{
    char *msg;

    g_assert(plugin_funcs.restart_task);

    if (!subject) {
        msg = g_strdup_printf(_("Entire collection restarted, collection_name = '%s'"), collection_name);
    } else {
        msg = g_strdup_printf(_("Task restarted, collection_name = '%s', subject = '%s'"), collection_name, subject);
    }
    nntpgrab_core_emit_debug_message(msg);
    g_free(msg);

    return plugin_funcs.restart_task(collection_name, subject, errmsg);
}

gboolean
schedular_save_queue(Schedular *obj, char **errmsg)
{
    g_assert(plugin_funcs.save_queue);

    return plugin_funcs.save_queue(errmsg);
}

struct _foreach_data
{
    ForeachCollectionFunc collection_func;
    ForeachFileFunc file_func;
    ForeachGroupFunc group_func;
    gpointer data;
};

static void
foreach_func(gpointer data, gpointer user_data)
{
    struct _foreach_data *foreach_data = (struct _foreach_data *) user_data;
    NNTPCollection *collection = (NNTPCollection *) data;
    GList *list, *list2;

    if (foreach_data->collection_func) {
        foreach_data->collection_func(collection->collection_name, collection->total_size, collection->total_size_remaining, collection->position, foreach_data->data);
    }

    if (!foreach_data->group_func && !foreach_data->file_func) {
        return;
    }

    list = collection->files;
    while (list) {
        NNTPFile *file = (NNTPFile *) list->data;

        if (foreach_data->file_func) {
            foreach_data->file_func(collection->collection_name, file->subject, file->poster, (gint64) file->stamp, file->file_size, file->file_size_remaining, file->position, g_list_length(file->parts), file->num_parts_downloaded, file->num_parts_failed, file->status, file->real_filename, foreach_data->data);
        }

        if (foreach_data->group_func) {
            list2 = file->groups;
            while (list2) {
                const char *group = (const char *) list2->data;

                foreach_data->group_func(collection->collection_name, file->subject, group, foreach_data->data);

                list2 = g_list_next(list2);
            }
        }

        list = g_list_next(list);
    }
}

void
schedular_foreach_task(Schedular *obj, ForeachCollectionFunc collection_func, ForeachFileFunc file_func, ForeachGroupFunc group_func, gpointer data)
{
    struct _foreach_data foreach_data;

    g_assert(plugin_funcs.foreach_task);

    foreach_data.collection_func = collection_func;
    foreach_data.file_func = file_func;
    foreach_data.group_func = group_func;
    foreach_data.data = data;

    plugin_funcs.foreach_task(foreach_func, &foreach_data);
}

gboolean
schedular_move_task(Schedular *obj, const char *collection_name_src, const char *subject_src, const char *collection_name_dest, int position_dest)
{
    return plugin_funcs.move_task(collection_name_src, subject_src, collection_name_dest, position_dest);
}

gboolean
schedular_move_collection(Schedular *obj, const char *collection_name, int new_position)
{
    return plugin_funcs.move_collection(collection_name, new_position);
}

static void
task_added_callback(const NNTPCollection *collection, const NNTPFile *file)
{
    nntpgrab_core_emit_file_added(collection, file);
}

static void
task_removed_callback(const char *collection_name, const char *subject, guint64 total_file_size, guint64 total_file_size_remaining)
{
    nntpgrab_core_emit_file_removed(collection_name, subject, total_file_size, total_file_size_remaining);
}

static void
task_state_changed_callback(const char *collection_name, const char *subject, TaskState old_state, TaskState new_state)
{
    nntpgrab_core_emit_file_state_changed(collection_name, subject, old_state, new_state);
}

static void
collection_added_callback(const char *collection_name)
{
    nntpgrab_core_emit_collection_added(collection_name);
}

static void
collection_removed_callback(const char *collection_name)
{
    nntpgrab_core_emit_collection_removed(collection_name);
}

static GList *
config_get_avail_servers(void)
{
    g_assert(schedular);

    return configuration_get_avail_servers(schedular->config);
}

static ConfigServer *
config_get_server_info(const char *servername)
{
    g_assert(schedular);

    return configuration_get_server_info(schedular->config, servername);
}

static ConfigOpts
config_get_opts (void)
{
    g_assert(schedular);

    return configuration_get_opts(schedular->config);
}

static gpointer
nntp_connection_initialize (const char *servername, NNTPGrabErrCode *err)
{
    NNTPConnection *conn;
    ConfigServer *server;

    server = configuration_get_server_info(schedular->config, servername);
    if (!server) {
        return NULL;
    }

    conn = nntpconnection_new(schedular->config, *server);
    g_slice_free(ConfigServer, server);

    if ((*err = nntpconnection_connect(conn)) != NNTP_ERROR_NONE) {
        nntpconnection_destroy(conn);
        return NULL;
    }

    return (gpointer) conn;
}

static void
nntp_connection_destroy (gpointer obj)
{
    g_assert(obj != NULL);

    if (nntpconnection_get_is_connected(obj)) {
        nntpconnection_disconnect(obj);
    }

    nntpconnection_destroy(obj);
}

static NNTPGrabErrCode
nntp_connection_get_part (gpointer obj, const char *collection_name, const NNTPFile *file, const NNTPPart *part, char **reason)
{
    char *filename;
    NNTPGrabErrCode ret;

    g_assert(obj != NULL);
    g_assert(schedular != NULL);

    // Is the connection still alive?
    if (!nntpconnection_get_is_connected(obj)) {
        // Nope, reconnect first
        nntpgrab_core_emit_debug_message(_("Connection lost. Trying to reconnect"));
        if ((ret = nntpconnection_connect(obj)) != NNTP_ERROR_NONE) {
            nntpgrab_core_emit_debug_message(_("Reconnect failed"));
            nntpconnection_disconnect(obj);
            return ret;
        }
    }

    g_static_rw_lock_reader_lock(&schedular->rwlock);
    filename = g_strdup_printf("%s%s%s.%i", schedular->opts.temp_directory, G_DIR_SEPARATOR_S, file->tmp_filename, part->partnum);
    g_static_rw_lock_reader_unlock(&schedular->rwlock);

    if ((ret = nntpconnection_get_part(obj, collection_name, file->subject, part, filename)) != NNTP_ERROR_NONE) {
        if (reason) {
            switch (ret) {
                case NNTP_ERROR_UNABLE_TO_SAVE_PART:
                    *reason = g_strdup_printf(_("Unable to save part to file '%s'"), filename);
                    break;
                default:
                    *reason = NULL;
            }
        }

        g_free(filename);
        return ret;
    }

    g_free(filename);

    nntpgrab_core_emit_debug_message(_("Saving of part SUCCEEDED"));

    return NNTP_ERROR_NONE;
}

static gboolean
decode_file (const char *collection_name, const NNTPFile *file, int *saved_errno)
{
    return decoder_decode_file(schedular->decoder, collection_name, file, saved_errno);
}

static void
notify_schedular_state_changed (SchedularState new_state, const char *reason)
{
    nntpgrab_core_emit_schedular_state_changed(new_state, reason);
}

static void
emit_fatal_error (const char *errmsg)
{
    nntpgrab_core_emit_fatal_error(errmsg);
}

static void
emit_part_download_complete (const char *servername, gpointer connection, const char *collection_name, const char *subject, int partnum, int size)
{
    int conn_id = -1;

    if (connection) {
        conn_id = nntpconnection_get_conn_id((NNTPConnection*) connection);
    }

    nntpgrab_core_emit_part_done(servername, conn_id, collection_name, subject, partnum, size);
}

static void
emit_part_download_failed (const char *servername, gpointer connection, const char *collection_name, const char *subject, int partnum, int size, gboolean all_servers_tried)
{
    int conn_id = -1;

    if (connection) {
        conn_id = nntpconnection_get_conn_id((NNTPConnection*) connection);
    }

    nntpgrab_core_emit_part_failed(servername, conn_id, collection_name, subject, partnum, size, all_servers_tried);
}

static void
emit_task_moved (const char *orig_collection_name, const char *subject, const char *new_collection_name, int old_position, int new_position)
{
    nntpgrab_core_emit_task_moved(orig_collection_name, subject, new_collection_name, old_position, new_position);
}

static void
emit_collection_moved (const char *collection_name, int old_position, int new_position)
{
    nntpgrab_core_emit_collection_moved(collection_name, old_position, new_position);
}

static void
emit_file_download_state_update (const char *collection_name, const char *subject, int num_parts_total, int num_parts_done, int num_parts_failed, guint64 file_size, guint64 file_size_remaining, guint64 total_file_size, guint64 total_file_size_remaining)
{
    nntpgrab_core_emit_file_download_state_update(collection_name, subject, num_parts_total, num_parts_done, num_parts_failed, file_size, file_size_remaining, total_file_size, total_file_size_remaining);
}

static void
config_changed_callback (Configuration *config, ConfigChangedFlag flag, gpointer data)
{
    g_assert(schedular);

    nntpgrab_core_emit_debug_message(_("Schedular: Received config changed signal"));

    if (flag == CONFIG_CHANGED_OPTS_CHANGED) {
        g_static_rw_lock_writer_lock(&schedular->rwlock);
        schedular->opts = configuration_get_opts(config);
        g_static_rw_lock_writer_unlock(&schedular->rwlock);
    }

    if (flag == CONFIG_CHANGED_SERVER_ADDED || flag == CONFIG_CHANGED_SERVER_DELETED) {
        if (schedular_get_state(schedular) == SCHEDULAR_STATE_RUNNING) {
            // Restart the schedular to apply the new settings
            schedular_stop(schedular);
            schedular_start(schedular);
        }
    }
}

---