NNTPGrab

automation

base

client/gui

client/macosx

client/web/module

glue

gui_base

indexer

nntpgrab_core

plugins/decoder

plugins/jsonrpc

plugins/par2/par2cmdline

plugins/par2

plugins/unpack/file-roller/copy-n-paste

plugins/unpack/file-roller/nautilus

plugins/unpack/file-roller/src

plugins/unpack

server_qt

../deps/include

../deps/include/gio-unix-2.0

../deps/include/gio-unix-2.0/gio

../deps/include/glib-2.0

../deps/include/glib-2.0/gio

../deps/include/glib-2.0/glib

../deps/include/glib-2.0/gobject

../deps/include/libsoup-2.4

../deps/include/libsoup-2.4/libsoup

/usr/include

/usr/include/libxml2

/usr/include/glib-2.0

/usr/include/glib-2.0/glib

/usr/include/glib-2.0/gobject

/usr/include/glib-2.0/gthread

/usr/include/pango-1.0

/usr/include/pango-1.0/pango

/usr/include/gtk-2.0

/usr/include/gtk-2.0/gdk

/usr/include/gtk-2.0/gdk-pixbuf

/usr/include/gtk-2.0/gdk-pixbuf-xlib

/usr/include/libgtkhtml-3.14

/usr/include/libgtkhtml-3.14/editor

/usr/include/libgtkhtml-3.14/gtkhtml

/usr/lib64/glib-2.0/include

/*

    Copyright (C) 2005-2009  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

*/

#ifndef _DECODER_THREAD_H_

#define _DECODER_THREAD_H_

#include "collections.h"

#include "configuration.h"

typedef struct _decoder_data {

    NNTPCollection  *collection;

    NNTPFile        *file;

    //GThreadPool     *poolPar2verify;

    //GThreadPool     *poolPar2repair;

    //GThreadPool     *poolUnpack;

} DecoderData;

void         decoder_thread_initialize(Configuration *config);

void         decoder_thread_destroy(void);

gboolean     decoder_thread_start(void);

gboolean     decoder_thread_stop(void);

void         decoder_thread_push_task(DecoderData *task);

#endif /* _DECODER_THREAD_H_ */

/*

    Copyright (C) 2005-2009  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 _MSC_VER

#include "config.h.win32"

#else

#include "config.h"

#endif

#include 

#include "nntpgrab_types.h"

#include "collections.h"

#include "collection_alloc.h"

NNTPFile *

file_new(void)

{

    NNTPFile *file = g_slice_new0(NNTPFile);

    file_ref(file);

    return file;

}

void

file_ref(NNTPFile *file)

{

    g_atomic_int_inc(&file->refcount);

//    g_print("file_ref: refcount = %i for subject %s\n", file->refcount, file->subject);

}

void

file_unref(NNTPFile *file)

{

    GList *groups;

    GList *parts;

    if (!g_atomic_int_dec_and_test(&file->refcount)) {

//        g_print("file_unref: refcount = %i for subject %s\n", file->refcount, file->subject);

        return;

    }

//    g_print("file_unref: refcount = %i for subject %s\n", file->refcount, file->subject);

//    g_print("releasing file\n");

    groups = file->groups;

    while (groups) {

        g_free(groups->data);

        groups = g_list_next(groups);

    }

    g_list_free(file->groups);

    parts = file->parts;

    while (parts) {

        g_slice_free(NNTPPart, parts->data);

        parts = g_list_next(parts);

    }

    g_list_free(file->parts);

    g_slice_free(NNTPFile, file);

}

NNTPCollection *

collection_new(void)

{

    NNTPCollection *collection = g_slice_new0(NNTPCollection);

    g_static_mutex_init(&collection->mutex);

    collection_ref(collection);

    return collection;

}

void

collection_ref(NNTPCollection *collection)

{

    g_atomic_int_inc(&collection->refcount);

//    g_print("collection_ref: refcount = %i for collection %s\n", collection->refcount, collection->collection_name);

}

void

collection_unref(NNTPCollection *collection)

{

    GList *files;

    if (!g_atomic_int_dec_and_test(&collection->refcount)) {

//        g_print("collection_unref: refcount = %i for collection %s\n", collection->refcount, collection->collection_name);

        return;

    }

//    g_print("collection_unref: refcount = %i for collection %s\n", collection->refcount, collection->collection_name);

//    g_print("releasing collection\n");

    files = collection->files;

    while (files) {

        NNTPFile *file = (NNTPFile *) files->data;

        file_unref(file);

        files = g_list_next(files);

    }

    g_list_free(collection->files);

    g_list_free(collection->files_to_download);

    g_slice_free(NNTPCollection, collection);

}

/*

    Copyright (C) 2005-2009  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

*/

/* Parts of this file are taken from ProFTP's throttle.c */

#include 

#include 

#include 

#include 

#include 

#ifdef WIN32

#include 

#endif

#include "nntpgrab.h"

#include "nntpgrab_plugin.h"

/* This allows a throttling process to be killed by the admin */

#ifndef WIN32

static void

xfer_rate_sigmask(int block)

{

    static sigset_t sig_set;

    if (block) {

        sigemptyset(&sig_set);

        sigaddset(&sig_set, SIGCHLD);

        sigaddset(&sig_set, SIGUSR1);

        sigaddset(&sig_set, SIGINT);

        sigaddset(&sig_set, SIGQUIT);

#ifdef SIGIO

        sigaddset(&sig_set, SIGIO);

#endif /* SIGIO */

#ifdef SIGBUS

        sigaddset(&sig_set, SIGBUS);

#endif /* SIGBUS */

        sigaddset(&sig_set, SIGHUP);

        while (sigprocmask(SIG_BLOCK, &sig_set, NULL) < 0) {

            if (errno == EINTR) {

                continue;

            }

            break;

        }

    } else {

        while (sigprocmask(SIG_UNBLOCK, &sig_set, NULL) < 0) {

            if (errno == EINTR) {

                continue;

            }

            break;

        }

    }

}

#endif

/* Returns the difference, in milliseconds, between the given timeval and now */

static long

xfer_rate_since(struct timeval *then)

{

    struct timeval now;

    gettimeofday(&now, NULL);

    return (((now.tv_sec - then->tv_sec) * 1000L) + ((now.tv_usec - then->tv_usec) / 1000L));

}

void

throttle_pause(struct timeval start_time, off_t xferlen, int max_bandwidth)

{

    long double xfer_rate_bps;

    long ideal = 0, elapsed = 0;

    /* Calculate the time interval since the transfer of data started. */

    elapsed = xfer_rate_since(&start_time);

    xfer_rate_bps = max_bandwidth * 1024.0;

    ideal = (xferlen * 1000L) / xfer_rate_bps;

    if (ideal > elapsed) {

        struct timeval tv;

        /* Setup for the select.  We use select() instead of usleep() because it

         * seems to be far more portable across platforms.

         *

         * ideal and elapsed are in milleconds, but tv_usec will be microseconds,

         * so be sure to convert properly.

         */

        tv.tv_usec = (ideal - elapsed) * 1000;

        tv.tv_sec = tv.tv_usec / 1000000L;

        tv.tv_usec = tv.tv_usec % 1000000L;

        ng_plugin_emit_log_msg(NULL, NG_LOG_LEVEL_DEBUG, "transferring too fast, delaying %ld sec%s, %ld usecs",

            (long int) tv.tv_sec, tv.tv_sec == 1 ? "" : "s", (long int) tv.tv_usec);

#ifdef WIN32

        /* Stupid Win32 doesn't support the select() function with no sockets attached.. */

        Sleep(ideal - elapsed);

#else

        /* No interruptions, please... */

        xfer_rate_sigmask(TRUE);

        if (select(0, NULL, NULL, NULL, &tv) < 0) {

            ng_plugin_emit_log_msg(NULL, NG_LOG_LEVEL_WARNING, "warning: unable to throttle bandwidth: %s", strerror(errno));

        }

        xfer_rate_sigmask(FALSE);

#endif

    }

    return;

}

/*

    Copyright (C) 2005-2009  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

*/

#ifndef _COLLECTION_ALLOC_H_

#define _COLLECTION_ALLOC_H_

NNTPFile        *file_new(void);

void             file_ref(NNTPFile *file);

void             file_unref(NNTPFile *file);

NNTPCollection  *collection_new(void);

void             collection_ref(NNTPCollection *collection);

void             collection_unref(NNTPCollection *collection);

#endif /* _COLLECTION_ALLOC_H_ */

/*

    Copyright (C) 2005-2009  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 "nntpgrab.h"

#include "nntpgrab_internal.h"

#include "nntpgrab_plugin.h"

#include "nntpgrab_utils.h"

#include "nntpconnection.h"

#include "queue_sqlite.h"

#include "download_queue.h"

#include "collection_alloc.h"

#include "configuration.h"

#include "decoder_thread.h"

#include "download_thread.h"

static GStaticMutex  mutex = G_STATIC_MUTEX_INIT;

static GList        *download_queue = NULL;

static int           num_servers = 0;

static int           num_enabled_servers = 0;

static unsigned int  is_high_priority_server;  /* These are actually a bit-masked variable where every server has one bit */

static unsigned int  is_normal_priority_server;

static unsigned int  is_server_enabled;

static ConfigOpts    opts;

static void

rebuild_download_queue_list(NNTPCollection *collection)

{

    GList *list_file;

    // Clear out any old items in the list

    g_list_free(collection->files_to_download);

    collection->files_to_download = NULL;

    list_file = collection->files;

    while (list_file) {

        NNTPFile *file = list_file->data;

        if (!file->file_is_downloaded) {

            collection->files_to_download = g_list_append(collection->files_to_download, file);

        }

        list_file = g_list_next(list_file);

    }

}

static gboolean

get_bit_value(unsigned int param, int pos)

{

    unsigned int shifted = param >> pos;

    return ((shifted & 0x1) == 0x1);

}

static void

set_bit_value(unsigned int *param, int pos, gboolean value)

{

    unsigned int mask = 0x1 << pos;

    if (value) {

        *param = *param ^ mask;

    } else {

        mask = ~mask;

        *param = *param & mask;

    }

}

static gboolean

test_server_may_download_part(NNTPPart *part, int server_id)

{

    if (!get_bit_value(is_server_enabled, server_id)) {

        return FALSE;

    }

    if (get_bit_value(part->servers_already_tried, server_id)) {

        return FALSE;

    }

    // Is this server one with the highest priority ?

    if (get_bit_value(is_high_priority_server, server_id)) {

        return TRUE;

    }

    // Are all the other high-priority servers already tried ?

    if ((is_high_priority_server & part->servers_already_tried) != is_high_priority_server) {

        return FALSE;

    }

    // All the high-priority servers are already tried, now test against the servers with the 'normal' priority

    if (get_bit_value(is_normal_priority_server, server_id)) {

        return TRUE;

    }

    if ((is_normal_priority_server & part->servers_already_tried) != is_normal_priority_server) {

        return FALSE;

    }

    // All the normal priority servers are also tried, allow the part

    return TRUE;

}

static gboolean

on_config_changed(Configuration *config, gpointer data)

{

    GList *list, *avail_servers;

    int i;

    g_static_mutex_lock(&mutex);

    opts = configuration_get_opts(config);

    is_high_priority_server = 0;

    is_normal_priority_server = 0;

    is_server_enabled = 0;

    num_enabled_servers = 0;

    num_servers = 0;

    avail_servers = configuration_get_avail_servers(config);

    list = avail_servers;

    i = 0;

    while (list) {

        char *servername = list->data;

        ConfigServer *server = configuration_get_server_info(config, servername);

        g_return_val_if_fail(server != NULL, FALSE);

        if (server->enabled) {

            num_enabled_servers++;

            if (server->priority == SERVER_PRIORITY_HIGH) {

                set_bit_value(&is_high_priority_server, i, TRUE);

                set_bit_value(&is_normal_priority_server, i, FALSE);

            } else if (server->priority == SERVER_PRIORITY_LOW) {

                set_bit_value(&is_high_priority_server, i, FALSE);

                set_bit_value(&is_normal_priority_server, i, FALSE);

            } else {

                set_bit_value(&is_high_priority_server, i, FALSE);

                set_bit_value(&is_normal_priority_server, i, TRUE);;

            }

        }

        set_bit_value(&is_server_enabled, i, server->enabled);

        i++;

        num_servers++;

        g_free(servername);

        g_slice_free(ConfigServer, server);

        list = g_list_next(list);

    }

    g_list_free(avail_servers);

    g_static_mutex_unlock(&mutex);

    return FALSE;

}

gboolean

download_queue_init(Configuration *config, char **errmsg)

{

    GList *list;

    g_return_val_if_fail(download_queue == NULL, FALSE);

    if (!queue_sqlite_load(&download_queue, errmsg)) {

        return FALSE;

    }

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

    on_config_changed(config, NULL);

    list = download_queue;

    while (list) {

        NNTPCollection *collection = list->data;

        rebuild_download_queue_list(collection);

        list = g_list_next(list);

    }

    return TRUE;

}

gboolean

download_queue_destroy(char **errmsg)

{

    g_static_mutex_lock(&mutex);

    if (!queue_sqlite_save(download_queue, errmsg)) {

        g_static_mutex_unlock(&mutex);

        return FALSE;

    }

    queue_sqlite_free(download_queue);

    download_queue = NULL;

    g_static_mutex_unlock(&mutex);

    return TRUE;

}

gboolean

download_queue_save(char **errmsg)

{

    gboolean retval;

    g_static_mutex_lock(&mutex);

    retval = queue_sqlite_save(download_queue, errmsg);

    g_static_mutex_unlock(&mutex);

    return retval;

}

gboolean

download_queue_get_next_part_to_download(NNTPCollection **collection, NNTPFile **file, NNTPPart **part, int server_id, gboolean *nothing_to_download)

{

    GList *list;

    g_return_val_if_fail(collection != NULL, FALSE);

    g_return_val_if_fail(file != NULL, FALSE);

    g_return_val_if_fail(part != NULL, FALSE);

    g_return_val_if_fail(server_id >= 0, FALSE);

    g_return_val_if_fail(nothing_to_download != NULL, FALSE);

    ng_plugin_emit_log_msg(NULL, NG_LOG_LEVEL_INFO, "download_queue_get_next_part_to_download for server_id %i", server_id);

    *nothing_to_download = TRUE;

    if (num_enabled_servers == 0) {

        /* No enabled servers found, don't even bother checking the queue */

        return FALSE;

    }

    g_static_mutex_lock(&mutex);

    list = download_queue;

    while (list) {

        GList *list2;

        *collection = (NNTPCollection *) list->data;

        // Is this collection being removed ?

        if ((*collection)->stop_flag) {

            list = g_list_next(list);

            continue;

        }

        list2 = (*collection)->files_to_download;

        while (list2) {

            GList *list3;

            gboolean complete_flag;

            *file = (NNTPFile *) list2->data;

            // Is this file already downloaded?

            if ((*file)->file_is_downloaded) {

                // Yes, remove it from the files_to_download list and start over

                (*collection)->files_to_download = g_list_remove((*collection)->files_to_download, list2->data);

                list2 = (*collection)->files_to_download;

                continue;

            }

            if ((*file)->now_decoding) {

                list2 = g_list_next(list2);

                continue;

            }

            // Is this file being removed?

            if ((*file)->stop_flag) {

                list2 = g_list_next(list2);

                continue;

            }

            complete_flag = TRUE;

            list3 = (*file)->parts;

            while (list3) {

                *part = (NNTPPart *) list3->data;

                if ((*part)->now_downloading) {

                    complete_flag = FALSE;

                    *nothing_to_download = FALSE;

                }

                if (!(*part)->downloaded) {

                    if (!(*part)->now_downloading && test_server_may_download_part(*part, server_id)) {

                        (*part)->now_downloading = TRUE;

                        (*file)->file_is_downloading = TRUE;

                        g_static_mutex_unlock(&mutex);

                        *nothing_to_download = FALSE;

                        ng_plugin_emit_log_msg(NULL, NG_LOG_LEVEL_DEBUG, "Returned part %i from file '%s' from collection '%s'", (*part)->part_num, (*file)->subject, (*collection)->collection_name);

                        return TRUE;

                    }

                    // Are all the servers already tried ?

                    if ((is_server_enabled & (*part)->servers_already_tried) == is_server_enabled) {

                        complete_flag = TRUE;

                    } else {

                        complete_flag = FALSE;

                        *nothing_to_download = FALSE;

                    }

                }

                list3 = g_list_next(list3);

            }

            // All the parts are already downloaded?

            if (complete_flag && !(*file)->now_decoding) {

                DecoderData *decoder_data;

                ng_plugin_emit_log_msg(NULL, NG_LOG_LEVEL_DEBUG, "File '%s' from collection '%s' is now ready to be decoded", (*file)->subject, (*collection)->collection_name);

                (*file)->now_decoding = TRUE;

                nntpgrab_core_emit_file_state_changed(FALSE, (*collection)->collection_name, (*file)->subject, NULL, (*file)->status, TASK_STATE_WAITING_FOR_DECODE);

                (*file)->status = TASK_STATE_WAITING_FOR_DECODE;

                decoder_data = g_slice_new0(DecoderData);

                decoder_data->collection = (*collection);

                decoder_data->file = (*file);

                //decoder_data->poolPar2verify = poolPar2verify;

                //decoder_data->poolPar2repair = poolPar2repair;

                //decoder_data->poolUnpack = poolUnpack;

                file_ref(*file);

                collection_ref(*collection);

                decoder_thread_push_task(decoder_data);

            }

            list2 = g_list_next(list2);

        }

        list = g_list_next(list);

    }

    *collection = NULL;

    *file = NULL;

    *part = NULL;

    g_static_mutex_unlock(&mutex);

    return FALSE;

}

void

download_queue_update_part_status(NNTPConnectionInfo *conn, NNTPCollection *collection, NNTPFile *file, NNTPPart *part, int server_id, gboolean success, gboolean reset_to_zero, gboolean dont_retry_anymore)

{

    GList *list;

    g_return_if_fail(conn != NULL);

    g_return_if_fail(collection != NULL);

    g_return_if_fail(file != NULL);

    g_return_if_fail(part != NULL);

    g_return_if_fail(server_id >= 0);

    g_static_mutex_lock(&collection->mutex);

    part->now_downloading = FALSE;

    part->downloaded = success;

    file->file_is_downloading = FALSE;

    if (!file->file_is_new) {

        file->file_is_changed = TRUE;

    }

    if (success) {

        if (file->file_size_remaining < part->size) {

            g_print("Invalid file size remaining!\n");

            g_print("file->file_size_remaining = %"G_GUINT64_FORMAT"\n", file->file_size_remaining);

            g_print("part->size = %i\n", part->size);

            file->file_size_remaining = part->size;

        }

        if (collection->total_size_remaining < part->size) {

            g_print("Invalid collection size remaining!\n");

            g_print("collection->total_size_remaining = %"G_GUINT64_FORMAT"\n", collection->total_size_remaining);

            g_print("part->size = %i\n", part->size);

            collection->total_size_remaining = part->size;

        }

        file->file_size_remaining -= part->size;

        collection->total_size_remaining -= part->size;

        file->num_parts_downloaded++;

        nntpgrab_core_emit_part_done(FALSE, conn->server_info.servername, conn->poll_fd.fd, collection->collection_name, file->subject, part->part_num, part->size);

        nntpgrab_core_emit_file_download_state_update(FALSE, collection->collection_name, file->subject, file->num_parts, file->num_parts_downloaded, file->num_parts_failed, file->file_size, file->file_size_remaining, collection->total_size, collection->total_size_remaining);

    } else if (!success && dont_retry_anymore) {

        set_bit_value(&part->servers_already_tried, server_id, TRUE);

    }

    // Are other threads downloading parts of this file ?

    list = file->parts;

    while (list) {

        NNTPPart *part2 = list->data;

        if (part2->now_downloading) {

            file->file_is_downloading = TRUE;

        }

        list = g_list_next(list);

    }

    // Are all the servers already tried ?

    if (!success && dont_retry_anymore) {

        gboolean all_servers_tried = TRUE;

        if ((is_server_enabled & part->servers_already_tried) == is_server_enabled) {

            all_servers_tried = TRUE;

        } else {

            all_servers_tried = FALSE;

        }

        nntpgrab_core_emit_part_failed(FALSE, conn->server_info.servername, conn->poll_fd.fd, collection->collection_name, file->subject, part->part_num, part->size, all_servers_tried);

        if (all_servers_tried && !reset_to_zero) {

            if (file->file_size_remaining < part->size) {

                g_print("Invalid file size remaining!\n");

                g_print("file->file_size_remaining = %"G_GUINT64_FORMAT"\n", file->file_size_remaining);

                g_print("part->size = %i\n", part->size);

                file->file_size_remaining = part->size;

            }

            if (collection->total_size_remaining < part->size) {

                g_print("Invalid collection size remaining!\n");

                g_print("collection->total_size_remaining = %"G_GUINT64_FORMAT"\n", collection->total_size_remaining);

                g_print("part->size = %i\n", part->size);

                collection->total_size_remaining = part->size;

            }

            /* Assume the part was downloaded as all servers were tried and no one had the part we requested */

            part->downloaded = TRUE;

            file->num_parts_failed++;

            file->file_size_remaining -= part->size;

            collection->total_size_remaining -= part->size;

            nntpgrab_core_emit_file_download_state_update(FALSE, collection->collection_name, file->subject, file->num_parts, file->num_parts_downloaded, file->num_parts_failed, file->file_size, file->file_size_remaining, collection->total_size, collection->total_size_remaining);

        }

    }

    if (reset_to_zero) {

        file->num_parts_downloaded = 0;

        file->num_parts_failed = 0;

        nntpgrab_core_emit_file_download_state_update(FALSE, collection->collection_name, file->subject, file->num_parts, file->num_parts_downloaded, file->num_parts_failed, file->file_size, file->file_size_remaining, collection->total_size, collection->total_size_remaining);

    }

    g_static_mutex_unlock(&collection->mutex);

}

static gboolean

test_is_subject_already_in_collection(NNTPCollection *collection, const char *subject)

{

    GList *list;

    list = collection->files;

    while (list) {

        NNTPFile *file = (NNTPFile *) list->data;

        if (!strcmp(file->subject, subject)) {

            return TRUE;

        }

        list = g_list_next(list);

    }

    return FALSE;

}

static gboolean

test_if_collection_contains_same_poster(NNTPCollection *collection, const char *poster)

{

    GList *list;

    list = collection->files;

    while (list) {

        NNTPFile *file = (NNTPFile *) list->data;

        if (strcmp(file->poster, poster)) {

            /* Different poster found */

            return FALSE;

        }

        list = g_list_next(list);

    }

    return TRUE;

}

gboolean

download_queue_add_file_to_queue_internal(const char *collection_name, NNTPFile *file, char **errmsg)

{

    GList *list;

    NNTPCollection *collection;

    g_static_mutex_lock(&mutex);

    // Find out if the collection_name already exists in the download queue

    list = download_queue;

    while (list) {

        collection = (NNTPCollection *) list->data;

        if (!strcmp(collection->collection_name, collection_name)) {

            // Collection found!

            // Does the subject already exists in this collection ?

            if (test_is_subject_already_in_collection(collection, file->subject)) {

                if (errmsg) {

                    *errmsg = g_strdup_printf(_("There already is a file with subject '%s' part of collection '%s'"), file->subject, collection_name);

                }

                g_static_mutex_unlock(&mutex);

                return FALSE;

            }

            // Add the file to this collection

            collection->files = g_list_append(collection->files, file);

            collection->files_to_download = g_list_append(collection->files_to_download, file);

            collection->total_size += file->file_size;

            collection->total_size_remaining += file->file_size;

            file->file_size_remaining = file->file_size;

            file->position = g_list_index(collection->files, file) + 1;

            file->file_is_new = TRUE;

            nntpgrab_core_emit_file_added(FALSE, collection->collection_name, file->subject, file->poster, file->stamp, file->file_size, collection->total_size, collection->total_size_remaining, file->status, file->num_parts, file->groups);

            // Is this the same poster as all the other files in this collection?

            if (test_if_collection_contains_same_poster(collection, file->poster)) {

                if (strcmp(collection->poster, file->poster)) {

                    // The poster wasn't set for this collection yet

                    strncpy(collection->poster, file->poster, sizeof(collection->poster) - 1);

                    nntpgrab_core_emit_collection_modified(FALSE, collection_name, collection->poster);

                    if (!collection->collection_is_new) {

                        collection->collection_is_changed = TRUE;

                    }

                }

            } else {

                // Some other file contained a different poster, reset to empty

                if (strcmp(collection->poster, file->poster)) {

                    memset(collection->poster, 0, sizeof(collection->poster));

                    nntpgrab_core_emit_collection_modified(FALSE, collection_name, collection->poster);

                    if (!collection->collection_is_new) {

                        collection->collection_is_changed = TRUE;

                    }

                }

            }

            g_static_mutex_unlock(&mutex);

            download_thread_awaken_if_needed();

            return TRUE;

        }

        list = g_list_next(list);

    }

    // Collection doesn't exist yet, add a new one

    collection = collection_new();

    strncpy((char *) collection->collection_name, collection_name, sizeof(collection->collection_name));

    strncpy((char *) collection->poster, file->poster, sizeof(collection->poster));

    download_queue = g_list_append(download_queue, collection);

    nntpgrab_core_emit_collection_added(FALSE, collection_name, collection->poster);

    // Finally add the task to the collection

    collection->files = g_list_append(collection->files, file);

    collection->files_to_download = g_list_append(collection->files_to_download, file);

    collection->total_size = file->file_size;

    collection->total_size_remaining = file->file_size;

    collection->collection_is_new = TRUE;

    file->file_is_new = TRUE;

    file->position = 1;

    nntpgrab_core_emit_file_added(FALSE, collection->collection_name, file->subject, file->poster, file->stamp, file->file_size, collection->total_size, collection->total_size_remaining, file->status, file->num_parts, file->groups);

    g_static_mutex_unlock(&mutex);

    download_thread_awaken_if_needed();

    return TRUE;

}

gboolean

download_queue_add_file_to_queue(const char *collection_name, const char *subject, const char *poster, time_t stamp, nguint64 file_size, NGList *groups, NGList *parts, char **errmsg)

{

    NNTPFile *file;

    GList *list;

    char *tmp;

    char *tmp2;

    GHashTable *hashtable;

    char collection_name_copy[256];          /* This parameter needs to be of the same size as defined in base/collections.h! */

    g_return_val_if_fail(collection_name != NULL, FALSE);

    g_return_val_if_fail(subject != NULL, FALSE);

    g_return_val_if_fail(poster != NULL, FALSE);

    g_return_val_if_fail(file_size >= 0, FALSE);

    g_return_val_if_fail(groups != NULL, FALSE);

    g_return_val_if_fail(parts != NULL, FALSE);

    memset(&collection_name_copy, 0, sizeof(collection_name_copy));

    strncpy(collection_name_copy, collection_name, sizeof(collection_name_copy) - 1);

    file = file_new();

    strncpy((char *) file->subject, subject, sizeof(file->subject));

    strncpy((char *) file->poster, poster, sizeof(file->poster));

    file->stamp = stamp;

    file->file_size = file_size;

    file->file_size_remaining = file_size;

    tmp = tempnam(opts.temp_directory, NULL);

    tmp2 = g_path_get_basename(tmp);

    strncpy((char *) file->tmp_filename, tmp2, sizeof(file->tmp_filename));

    g_free(tmp2);

    free(tmp);                  // Don't use g_free here as tempnam is a libc function!

    // Detect the file type and the par2 blocks

    if (!nntpgrab_utils_strip_subject(file->subject, NULL, NULL, NULL, NULL, NULL, &file->file_type, &file->par2_start_num, &file->par2_num_blocks, NULL, &file->num_parts)) {

        // Sanitize values

        file->file_type = NNTP_FILE_TYPE_UNKNOWN;

        file->par2_start_num = -1;

        file->par2_num_blocks =-1;

        file->num_parts = g_list_length((GList*) parts);

    }

    file->status = TASK_STATE_WAITING_FOR_DOWNLOAD;

    if (file->par2_start_num == -1) {

        file->par2_end_num = -1;

    } else {

        file->par2_end_num = file->par2_start_num + file->par2_num_blocks;

    }

    list = (GList*) groups;

    while (list) {

        const char *group = list->data;

        file->groups = g_list_append(file->groups, g_strdup(group));

        list = g_list_next(list);

    }

    // When adding parts, prevent the adding of duplicate part numbers

    hashtable = g_hash_table_new(g_direct_hash, g_direct_equal);

    list = (GList*) parts;

    while (list) {

        NNTPGrabPart *part = list->data;

        NNTPPart *new_part;

        new_part = g_hash_table_lookup(hashtable, GINT_TO_POINTER(part->part_num));

        if (new_part) {

            // Hey, we've seen this part before!

            // Only keep the newest only

            memset(new_part, 0, sizeof(NNTPPart));

        } else {

            // We've never seen this part before

            new_part = g_slice_new0(NNTPPart);

            file->parts = g_list_append(file->parts, new_part);

            g_hash_table_insert(hashtable, GINT_TO_POINTER(part->part_num), new_part);

        }

        strncpy((char *) new_part->message_id, part->message_id, sizeof(new_part->message_id));

        new_part->size = part->size;

        new_part->part_num = part->part_num;

        list = g_list_next(list);

    }

    g_hash_table_destroy(hashtable);

    return download_queue_add_file_to_queue_internal(collection_name_copy, file, errmsg);

}

gboolean

download_queue_del_file_from_queue(const char *collection_name, const char *subject, char **errmsg)

{

    GList *list;

    NNTPCollection *collection;

    int i;

    sqlite3 *db;

    char collection_name_copy[256];          /* This parameter needs to be of the same size as defined in base/collections.h! */

    g_return_val_if_fail(collection_name != NULL, FALSE);

    /* subject can be NULL */

    memset(&collection_name_copy, 0, sizeof(collection_name_copy));

    strncpy(collection_name_copy, collection_name, sizeof(collection_name_copy) - 1);

    g_static_mutex_lock(&mutex);

    if (!(db = queue_sqlite_open(errmsg))) {

        g_static_mutex_unlock(&mutex);

        return FALSE;

    }

    if (!queue_sqlite_begin_transaction(db, errmsg)) {

        g_static_mutex_unlock(&mutex);

        queue_sqlite_close(db);

        return FALSE;

    }

    // Find out if the collection_name exists in the download queue

    list = download_queue;

    while (list) {

        collection = (NNTPCollection *) list->data;

        if (!strcmp(collection->collection_name, collection_name_copy)) {

            // Collection found!

            GList *list2;

            gboolean delete_flag = FALSE;

            // Do we have a subject?

            if (!subject) {

                // Delete the whole collection

                delete_flag = TRUE;

                // Mark the whole collection as being removed

                collection->stop_flag = TRUE;

            }

            // Try to find the subject

            list2 = collection->files;

            while (list2) {

                NNTPFile *file = (NNTPFile *) list2->data;

                list2 = g_list_next(list2);

                if (delete_flag || !strcmp(file->subject, subject)) {

                    // Bingo!

                    // Is this file being downloaded ?

                    while (file->file_is_downloading || file->now_decoding) {

                        // Mark this file as being removed

                        file->stop_flag = TRUE;

                        // Release the lock and wait for the download to complete

                        g_static_mutex_unlock(&mutex);

                        g_usleep(G_USEC_PER_SEC / 10);

                        g_static_mutex_lock(&mutex);

                        // Make sure that when all parts are downloaded, but the decoder isn't triggered, we reset the flag

                        if (file->file_is_downloading && !file->now_decoding) {

                            GList *list3 = file->parts;

                            gboolean is_downloading_flag = FALSE;

                            while (list3) {

                                NNTPPart *part = (NNTPPart*) list3->data;

                                if (part->now_downloading) {

                                    is_downloading_flag = TRUE;

                                }

                                list3 = g_list_next(list3);

                            }

                            file->file_is_downloading = is_downloading_flag;

                        }

                    }

                    // Remove the entry from the database

                    if (!queue_sqlite_remove_file_from_db(db, collection, file, TRUE, errmsg)) {

                        g_static_mutex_unlock(&mutex);

                        queue_sqlite_close(db);

                        return FALSE;

                    }

                    // Unlink any possible downloaded parts

                    for (i = 0; i < file->num_parts; i++) {

                        char *filename = g_strdup_printf("%s%s%s.%i", opts.temp_directory, G_DIR_SEPARATOR_S, file->tmp_filename, i + 1);

                        g_unlink(filename);

                        g_free(filename);

                    }

                    // Update the file size

                    collection->total_size -= file->file_size;

                    collection->total_size_remaining -= file->file_size_remaining;

                    // Remove it from the list

                    collection->files = g_list_remove(collection->files, file);

                    collection->files_to_download = g_list_remove(collection->files_to_download, file);

                    nntpgrab_core_emit_file_removed(FALSE, collection_name_copy, file->subject, collection->total_size, collection->total_size_remaining);

                    file_unref(file);

                    // Does the collection still got any items in it ?

                    if (g_list_length(collection->files) == 0) {

                        // Yes, remove the entire collection

                        if (!queue_sqlite_remove_collection_from_db(db, collection, errmsg)) {

                            g_static_mutex_unlock(&mutex);

                            queue_sqlite_close(db);

                            return FALSE;

                        }

                        download_queue = g_list_remove(download_queue, collection);

                        nntpgrab_core_emit_collection_removed(FALSE, collection_name_copy);

                        collection_unref(collection);

                        g_static_mutex_unlock(&mutex);

                        if (!queue_sqlite_end_transaction(db, errmsg)) {

                            queue_sqlite_close(db);

                            return FALSE;

                        }

                        queue_sqlite_close(db);

                        return TRUE;

                    }

                    if (!delete_flag) {

                        // Is this the same poster as all the other files in this collection?

                        file = collection->files->data;

                        if (test_if_collection_contains_same_poster(collection, file->poster)) {

                            if (strcmp(collection->poster, file->poster)) {

                                // The poster wasn't set for this collection yet

                                strncpy(collection->poster, file->poster, sizeof(collection->poster) - 1);

                                nntpgrab_core_emit_collection_modified(FALSE, collection_name_copy, collection->poster);

                                if (!collection->collection_is_new) {

                                    collection->collection_is_changed = TRUE;

                                }

                            }

                        } else {

                            // Some other file contained a different poster, reset to empty

                            if (strcmp(collection->poster, file->poster)) {

                                memset(collection->poster, 0, sizeof(collection->poster));

                                nntpgrab_core_emit_collection_modified(FALSE, collection_name_copy, collection->poster);

                                if (!collection->collection_is_new) {

                                    collection->collection_is_changed = TRUE;

                                }

                            }

                        }

                        if (!queue_sqlite_end_transaction(db, errmsg)) {

                            g_static_mutex_unlock(&mutex);

                            queue_sqlite_close(db);

                            return FALSE;

                        }

                        g_static_mutex_unlock(&mutex);

                        queue_sqlite_close(db);

                        return TRUE;

                    }

                }

            }

            if (delete_flag) {

                g_static_mutex_unlock(&mutex);

                if (!queue_sqlite_end_transaction(db, errmsg)) {

                    return FALSE;

                }

                queue_sqlite_close(db);

                return TRUE;

            }

        }

        list = g_list_next(list);

    }

    if (errmsg) {

        *errmsg = g_strdup_printf(_("Subject '%s' was not found in collection '%s'"), subject, collection_name_copy);

    }

    g_static_mutex_unlock(&mutex);

    queue_sqlite_close(db);

    return FALSE;

}

gboolean

download_queue_restart_file(const char *collection_name, const char *subject, char **errmsg)

{

    GList *list;

    int pos_last_file_to_download;

    NNTPCollection *collection;

    char collection_name_copy[256];          /* This parameter needs to be of the same size as defined in base/collections.h! */

    g_return_val_if_fail(collection_name != NULL, FALSE);

    /* subject can be NULL */

    memset(&collection_name_copy, 0, sizeof(collection_name_copy));

    strncpy(collection_name_copy, collection_name, sizeof(collection_name_copy) - 1);

    g_static_mutex_lock(&mutex);

    // Find out if the collection_name already exists in the download queue

    list = download_queue;

    while (list) {

        collection = (NNTPCollection *) list->data;

        pos_last_file_to_download = 0;

        if (!strcmp(collection->collection_name, collection_name_copy)) {

            // Collection found!

            GList *list2;

            gboolean restart_flag = FALSE;

            // Do we have a subject?

            if (!subject) {

                // Restart the whole collection

                restart_flag = TRUE;

                // Clear out the old list with items which still need to be downloaded

                g_list_free(collection->files_to_download);

                collection->files_to_download = NULL;

            }

            list2 = collection->files;

            while (list2) {

                int pos;

                NNTPFile *file = list2->data;

                if (restart_flag || !strcmp(file->subject, subject)) {

                    // Subject found!

                    GList *list3;

                    gboolean file_is_downloading = FALSE;

                    int filesize_remaining_active = 0;

                    list3 = file->parts;

                    while (list3) {

                        NNTPPart *part = list3->data;

                        if (part->now_downloading) {

                            list3 = g_list_next(list3);

                            file_is_downloading = TRUE;

                            filesize_remaining_active += part->size;

                            continue;

                        }

                        part->downloaded = FALSE;

                        part->now_downloading = FALSE;

                        part->servers_already_tried = 0;

                        list3 = g_list_next(list3);

                    }

                    file->file_is_downloading = file_is_downloading;

                    file->file_is_downloaded = FALSE;

                    file->now_decoding = FALSE;

                    file->num_parts_downloaded = 0;

                    file->num_parts_failed = 0;

                    if (restart_flag) {

                        // If the whole collection is restarted, don't bother with merging at the right position in the files_to_download list

                        collection->files_to_download = g_list_append(collection->files_to_download, file);

                    } else if (!file_is_downloading) {

                        // Only a part of the collection has been restarted

                        // The position of the previous item which still needs to be downloaded

                        // has already been saved in list_last_file_to_download

                        if (pos_last_file_to_download > 0) {

                            pos_last_file_to_download++;

                        }

                        collection->files_to_download = g_list_insert(collection->files_to_download, file,  pos_last_file_to_download);

                    }

                    collection->total_size_remaining += file->file_size - file->file_size_remaining - filesize_remaining_active;

                    file->file_size_remaining = file->file_size - filesize_remaining_active;

                    if (!file->file_is_new) {

                        file->file_is_changed = TRUE;

                    }

                    nntpgrab_core_emit_file_state_changed(FALSE, collection_name_copy, file->subject, NULL, file->status, TASK_STATE_WAITING_FOR_DOWNLOAD);

                    file->status = TASK_STATE_WAITING_FOR_DOWNLOAD;

                    if (!restart_flag) {

                        g_static_mutex_unlock(&mutex);

                        download_thread_awaken_if_needed();

                        return TRUE;

                    }

                }

                pos = g_list_index(collection->files_to_download, file);

                if (pos > 0) {

                    pos_last_file_to_download = pos;

                }

                list2 = g_list_next(list2);

            }

            if (restart_flag) {

                g_static_mutex_unlock(&mutex);

                download_thread_awaken_if_needed();

                return TRUE;

            }

        }

        list = g_list_next(list);

    }

    g_static_mutex_unlock(&mutex);

    if (errmsg) {

        *errmsg = g_strdup_printf(_("Subject '%s' was not found in collection '%s'"), subject, collection_name_copy);

    }

    return FALSE;

}

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->poster, 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

download_queue_foreach_file(ForeachCollectionFunc collection_func, ForeachFileFunc file_func, ForeachGroupFunc group_func, gpointer data)

{

    struct _foreach_data foreach_data;

    foreach_data.collection_func = collection_func;

    foreach_data.file_func = file_func;

    foreach_data.group_func = group_func;

    foreach_data.data = data;

    g_static_mutex_lock(&mutex);

    g_list_foreach(download_queue, foreach_func, &foreach_data);

    g_static_mutex_unlock(&mutex);

}

gboolean

download_queue_move_file(const char *collection_name_src, const char *subject_src, const char *collection_name_dest, int position_dest)

{

    GList *list;

    NNTPCollection *collection_src = NULL;

    NNTPCollection *collection_dest = NULL;

    NNTPFile *file = NULL;

    gboolean found = FALSE;

    int old_position = 0;

    int i;

    guint hash_collection_name_src;

    guint hash_subject_src;