bb_store_file.c File Reference

#include <errno.h>
#include <stdlib.h>
#include <stdio.h>
#include <time.h>
#include <string.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <signal.h>
#include "gwlib/gwlib.h"
#include "msg.h"
#include "bearerbox.h"
#include "sms.h"

Go to the source code of this file.

Functions
static void	write_msg (Msg *msg)
static int	read_msg (Msg **msg, Octstr *os, long *off)
static int	open_file (Octstr *name)
static int	rename_store (void)
static int	do_dump (void)
static void	store_dumper (void *arg)
static void	store_file_for_each_message (void(*callback_fn)(Msg *msg, void *data), void *data)
static long	store_file_messages (void)
static int	store_to_dict (Msg *msg)
static int	store_file_save (Msg *msg)
static int	store_file_save_ack (Msg *msg, ack_status_t status)
static int	store_file_load (void(*receive_msg)(Msg *))
static int	store_file_dump (void)
static void	store_file_shutdown (void)
int	store_file_init (const Octstr *fname, long dump_freq)

Variables
static FILE *	file = NULL
static Octstr *	filename = NULL
static Octstr *	newfile = NULL
static Octstr *	bakfile = NULL
static Mutex *	file_mutex = NULL
static long	cleanup_thread = -1
static long	dump_frequency = 0
static Dict *	sms_dict = NULL
static int	active = 1
static time_t	last_dict_mod = 0
static List *	loaded

Function Documentation

do_dump()

static int do_dump ( void  )

static

Definition at line 176 of file bb_store_file.c.

References dict_get(), dict_keys(), file, filename, gwlist_destroy(), gwlist_get(), gwlist_len(), msg, newfile, octstr_destroy_item(), open_file(), rename_store(), sms_dict, and write_msg().

Referenced by store_file_dump(), and store_file_load().

{
    Octstr *key;
    Msg *msg;
    List *sms_list;
    long l;

    if (filename == NULL)
        return 0;

    /* create a new store-file and save all non-acknowledged
     * messages into it
     */
    if (open_file(newfile)==-1)
        return -1;

    sms_list = dict_keys(sms_dict);
    for (l=0; l < gwlist_len(sms_list); l++) {
        key = gwlist_get(sms_list, l);
        msg = dict_get(sms_dict, key);
        if (msg != NULL)
            write_msg(msg);
    }
    fflush(file);
    gwlist_destroy(sms_list, octstr_destroy_item);

    /* rename old storefile as .bak, and then new as regular file
     * without .new ending */

    return rename_store();
}

open_file()

static int open_file ( Octstr *  name )

static

Definition at line 146 of file bb_store_file.c.

References error(), file, name, and octstr_get_cstr.

Referenced by do_dump(), and store_file_load().

{
    file = fopen(octstr_get_cstr(name), "w");
    if (file == NULL) {
        error(errno, "Failed to open '%s' for writing, cannot create store-file",
          octstr_get_cstr(name));
        return -1;
    }
    return 0;
}

read_msg()

static int read_msg ( Msg **  msg, Octstr *  os, long *  off  )

static

Definition at line 118 of file bb_store_file.c.

References decode_network_long(), error(), gw_assert(), msg, octstr_copy, octstr_destroy(), octstr_get_many_chars(), octstr_len(), and store_msg_unpack.

Referenced by store_file_load().

{
    unsigned char buf[4];
    long i;
    Octstr *pack;

    gw_assert(*off >= 0);
    if (*off + 4 > octstr_len(os)) {
        error(0, "Packet too short while unpacking Msg.");
        return -1;
    }

    octstr_get_many_chars((char*)buf, os, *off, 4);
    i = decode_network_long(buf);
    *off  +=  4;
    
    pack = octstr_copy(os, *off, i);
    *off += octstr_len(pack);
    *msg = store_msg_unpack(pack);
    octstr_destroy(pack);
    
    if (!*msg)
        return -1;
    
    return 0;
}

rename_store()

static int rename_store ( void  )

static

Definition at line 158 of file bb_store_file.c.

References bakfile, error(), filename, newfile, and octstr_get_cstr.

Referenced by do_dump().

{
    if (rename(octstr_get_cstr(filename), octstr_get_cstr(bakfile)) == -1) {
        if (errno != ENOENT) {
            error(errno, "Failed to rename old store '%s' as '%s'",
            octstr_get_cstr(filename), octstr_get_cstr(bakfile));
            return -1;
        }
    }
    if (rename(octstr_get_cstr(newfile), octstr_get_cstr(filename)) == -1) {
        error(errno, "Failed to rename new store '%s' as '%s'",
              octstr_get_cstr(newfile), octstr_get_cstr(filename));
        return -1;
    }
    return 0;
}

store_dumper()

static void store_dumper ( void *  arg )

static

Definition at line 213 of file bb_store_file.c.

References active, bakfile, dict_destroy(), dump_frequency, file, file_mutex, filename, gwthread_sleep(), last_dict_mod, mutex_destroy(), newfile, octstr_destroy(), sms_dict, and store_dump.

Referenced by store_file_load().

{
    time_t now;
    int busy = 0;

    while (active) {
        now = time(NULL);
        /*
         * write store to file up to each N. second, providing
         * that something happened or if we are constantly busy.
         */
        if (now - last_dict_mod > dump_frequency || busy) {
            store_dump();
            /* 
             * make sure that no new dump is done for a while unless
             * something happens. This moves the trigger in the future
             * and allows the if statement to pass if nothing happened
             * in the mean time while sleeping. The busy flag is needed
             * to garantee we do dump in case we are constantly busy
             * and hence the difference between now and last dict
             * operation is less then dump frequency, otherwise we
             * would never dump. This is for constant high load.
             */
            last_dict_mod = time(NULL) + 3600*24;
            busy = 0;
        } else {
            busy = (now - last_dict_mod) > 0;
        }
        gwthread_sleep(dump_frequency);
    }
    store_dump();
    if (file != NULL)
       fclose(file);
    octstr_destroy(filename);
    octstr_destroy(newfile);
    octstr_destroy(bakfile);
    mutex_destroy(file_mutex);

    dict_destroy(sms_dict);
    /* set all vars to NULL */
    filename = newfile = bakfile = NULL;
    file_mutex = NULL;
    sms_dict = NULL;
}

store_file_dump()

static int store_file_dump ( void  )

static

Definition at line 482 of file bb_store_file.c.

References debug(), dict_key_count(), do_dump(), file, file_mutex, mutex_lock, mutex_unlock, and sms_dict.

Referenced by store_file_init().

{
    int retval;

    debug("bb.store", 0, "Dumping %ld messages to store",
      dict_key_count(sms_dict));
    mutex_lock(file_mutex);
    if (file != NULL) {
        fclose(file);
        file = NULL;
    }
    retval = do_dump();
    mutex_unlock(file_mutex);

    return retval;
}

store_file_for_each_message()

static void store_file_for_each_message ( void(*)(Msg *msg, void *data)  callback_fn, void *  data  )

static

Definition at line 261 of file bb_store_file.c.

References status::callback_fn, status::data, dict_get(), dict_keys(), file_mutex, filename, gwlist_destroy(), gwlist_get(), gwlist_len(), msg, mutex_lock, mutex_unlock, octstr_destroy_item(), and sms_dict.

Referenced by store_file_init().

{
    List *keys;
    long l;
    Msg *msg;
    Octstr *key;

    /* if there is no store-file, then don't loop in sms_store */
    if (filename == NULL)
        return;

    mutex_lock(file_mutex);
    keys = dict_keys(sms_dict);
    for (l = 0; l < gwlist_len(keys); l++) {
        key = gwlist_get(keys, l);
        msg = dict_get(sms_dict, key);
        if (msg == NULL)
            continue;
        callback_fn(msg, data);
    }
    mutex_unlock(file_mutex);

    gwlist_destroy(keys, octstr_destroy_item);
}

store_file_init()

int store_file_init	(	const Octstr *	fname,
		long	dump_freq
	)

Definition at line 515 of file bb_store_file.c.

References active, bakfile, BB_STORE_DEFAULT_DUMP_FREQ, dict_create(), dump_frequency, file_mutex, filename, gwlist_add_producer(), gwlist_create, loaded, msg_destroy_item(), mutex_create, newfile, octstr_duplicate, octstr_format(), octstr_get_cstr, octstr_len(), panic, sms_dict, store_dump, store_file_dump(), store_file_for_each_message(), store_file_load(), store_file_messages(), store_file_save(), store_file_save_ack(), store_file_shutdown(), store_for_each_message, store_load, store_messages, store_save, store_save_ack, and store_shutdown.

Referenced by store_init().

{
    /* Initialize function pointers */
    store_messages = store_file_messages;
    store_save = store_file_save;
    store_save_ack = store_file_save_ack;
    store_load = store_file_load;
    store_dump = store_file_dump;
    store_shutdown = store_file_shutdown;
    store_for_each_message = store_file_for_each_message;

    if (fname == NULL)
        return 0; /* we are done */

    if (octstr_len(fname) > (FILENAME_MAX-5))
        panic(0, "Store file filename too long: `%s', failed to init.",
          octstr_get_cstr(fname));

    filename = octstr_duplicate(fname);
    newfile = octstr_format("%s.new", octstr_get_cstr(filename));
    bakfile = octstr_format("%s.bak", octstr_get_cstr(filename));

    sms_dict = dict_create(1024, msg_destroy_item);

    if (dump_freq > 0)
        dump_frequency = dump_freq;
    else
        dump_frequency = BB_STORE_DEFAULT_DUMP_FREQ;

    file_mutex = mutex_create();
    active = 1;

    loaded = gwlist_create();
    gwlist_add_producer(loaded);

    return 0;
}

store_file_load()

static int store_file_load ( void(*)(Msg *)  receive_msg )

static

Definition at line 383 of file bb_store_file.c.

References bakfile, cleanup_thread, dict_key_count(), dict_keys(), dict_remove(), do_dump(), error(), file, file_mutex, filename, gwlist_destroy(), gwlist_extract_first(), gwlist_remove_producer(), gwthread_create, info(), loaded, msg, msg_destroy(), msg_dump(), mutex_lock, mutex_unlock, newfile, octstr_destroy(), octstr_destroy_item(), octstr_get_cstr, octstr_len(), octstr_read_file(), open_file(), panic, read_msg(), sms_dict, store_dumper(), store_to_dict(), and warning().

Referenced by store_file_init().

{
    List *keys;
    Octstr *store_file, *key;
    Msg *msg;
    int retval, msgs;
    long end, pos;

    if (filename == NULL)
        return 0;

    mutex_lock(file_mutex);
    if (file != NULL) {
        fclose(file);
        file = NULL;
    }

    store_file = octstr_read_file(octstr_get_cstr(filename));
    if (store_file != NULL)
        info(0, "Loading store file `%s'", octstr_get_cstr(filename));
    else {
        store_file = octstr_read_file(octstr_get_cstr(newfile));
        if (store_file != NULL)
            info(0, "Loading store file `%s'", octstr_get_cstr(newfile));
        else {
            store_file = octstr_read_file(octstr_get_cstr(bakfile));
            if (store_file != NULL)
                   info(0, "Loading store file `%s'", octstr_get_cstr(bakfile));
            else {
                info(0, "Cannot open any store file, starting a new one");
                retval = open_file(filename);
                goto end;
            }
        }
    }

    info(0, "Store-file size %ld, starting to unpack%s", octstr_len(store_file),
        octstr_len(store_file) > 10000 ? " (may take awhile)" : "");


    pos = 0;
    msgs = 0;
    end = octstr_len(store_file);
    
    while (pos < end) {
        if (read_msg(&msg, store_file, &pos) == -1) {
            error(0, "Garbage at store-file, skipped.");
            continue;
        }
        if (msg_type(msg) == sms) {
            store_to_dict(msg);
            msgs++;
        } else if (msg_type(msg) == ack) {
            store_to_dict(msg);
        } else {
            warning(0, "Strange message in store-file, discarded, "
                "dump follows:");
            msg_dump(msg, 0);
        }
        msg_destroy(msg);
    }
    octstr_destroy(store_file);

    info(0, "Retrieved %d messages, non-acknowledged messages: %ld",
        msgs, dict_key_count(sms_dict));

    /* now create a new sms_store out of messages left */

    keys = dict_keys(sms_dict);
    while ((key = gwlist_extract_first(keys)) != NULL) {
        msg = dict_remove(sms_dict, key);
        if (store_to_dict(msg) != -1) {
            receive_msg(msg);
        } else {
            error(0, "Found unknown message type in store file.");
            msg_dump(msg, 0);
            msg_destroy(msg);
        }
        octstr_destroy(key);
    }
    gwlist_destroy(keys, octstr_destroy_item);

    /* Finally, generate new store file out of left messages */
    retval = do_dump();

end:
    mutex_unlock(file_mutex);

    /* allow using of store */
    gwlist_remove_producer(loaded);

    /* start dumper thread */
    if ((cleanup_thread = gwthread_create(store_dumper, NULL))==-1)
        panic(0, "Failed to create a cleanup thread!");

    return retval;
}

store_file_messages()

static long store_file_messages ( void  )

static

Definition at line 287 of file bb_store_file.c.

References dict_key_count(), and sms_dict.

Referenced by store_file_init().

{
    return (sms_dict ? dict_key_count(sms_dict) : -1);
}

store_file_save()

static int store_file_save ( Msg *  msg )

static

Definition at line 332 of file bb_store_file.c.

References file, file_mutex, filename, gwlist_consume(), loaded, msg, mutex_lock, mutex_unlock, store_to_dict(), and write_msg().

Referenced by store_file_init().

{
    if (filename == NULL)
        return 0;

    /* block here until store not loaded */
    gwlist_consume(loaded);

    /* lock file_mutex in order to have dict and file in sync */
    mutex_lock(file_mutex);
    if (store_to_dict(msg) == -1) {
        mutex_unlock(file_mutex);
        return -1;
    }
    
    /* write to file, too */
    write_msg(msg);
    fflush(file);
    mutex_unlock(file_mutex);

    return 0;
}

store_file_save_ack()

static int store_file_save_ack ( Msg *  msg, ack_status_t  status  )

static

Definition at line 356 of file bb_store_file.c.

References filename, msg, msg_create, msg_destroy(), status::status, store_save, and uuid_copy().

Referenced by store_file_init().

{
    Msg *mack;
    int ret;

    /* only sms are handled */
    if (!msg || msg_type(msg) != sms)
        return -1;

    if (filename == NULL)
        return 0;

    mack = msg_create(ack);
    if (!mack)
        return -1;

    mack->ack.time = msg->sms.time;
    uuid_copy(mack->ack.id, msg->sms.id);
    mack->ack.nack = status;

    ret = store_save(mack);
    msg_destroy(mack);

    return ret;
}

store_file_shutdown()

static void store_file_shutdown ( void  )

static

Definition at line 500 of file bb_store_file.c.

References active, cleanup_thread, filename, gwlist_destroy(), gwthread_join(), gwthread_wakeup(), and loaded.

Referenced by store_file_init().

{
    if (filename == NULL)
        return;

    active = 0;
    gwthread_wakeup(cleanup_thread);
    /* wait for cleanup thread */
    if (cleanup_thread != -1)
        gwthread_join(cleanup_thread);

    gwlist_destroy(loaded, NULL);
}

store_to_dict()

static int store_to_dict ( Msg *  msg )

static

Definition at line 293 of file bb_store_file.c.

References dict_put(), dict_remove(), last_dict_mod, msg, msg_destroy(), msg_duplicate(), MSG_PARAM_UNDEFINED, octstr_create, octstr_destroy(), sms_dict, uuid_generate(), uuid_is_null(), UUID_STR_LEN, uuid_unparse(), and warning().

Referenced by store_file_load(), and store_file_save().

{
    Msg *copy;
    Octstr *uuid_os;
    char id[UUID_STR_LEN + 1];
    
    /* always set msg id and timestamp */
    if (msg_type(msg) == sms && uuid_is_null(msg->sms.id))
        uuid_generate(msg->sms.id);

    if (msg_type(msg) == sms && msg->sms.time == MSG_PARAM_UNDEFINED)
        time(&msg->sms.time);

    if (msg_type(msg) == sms) {
        copy = msg_duplicate(msg);
        
        uuid_unparse(copy->sms.id, id);
        uuid_os = octstr_create(id);
        
        dict_put(sms_dict, uuid_os, copy);
        octstr_destroy(uuid_os);
        last_dict_mod = time(NULL);
    } else if (msg_type(msg) == ack) {
        uuid_unparse(msg->ack.id, id);
        uuid_os = octstr_create(id);
        copy = dict_remove(sms_dict, uuid_os);
        octstr_destroy(uuid_os);
        if (copy == NULL) {
            warning(0, "bb_store: get ACK of message not found "
                   "from store, strange?");
        } else {
            msg_destroy(copy);
            last_dict_mod = time(NULL);
        }
    } else
        return -1;
    return 0;
}

write_msg()

static void write_msg ( Msg *  msg )

static

Definition at line 102 of file bb_store_file.c.

References encode_network_long(), file, msg, octstr_destroy(), octstr_insert_data(), octstr_len(), octstr_print(), and store_msg_pack.

Referenced by do_dump(), and store_file_save().

{
    Octstr *pack;
    unsigned char buf[4];
    
    pack = store_msg_pack(msg);
    encode_network_long(buf, octstr_len(pack));
    octstr_insert_data(pack, 0, (char*)buf, 4);

    octstr_print(file, pack);
    fflush(file);

    octstr_destroy(pack);
}

Variable Documentation

active

int active = 1

static

Definition at line 97 of file bb_store_file.c.

Referenced by store_dumper(), store_file_init(), and store_file_shutdown().

bakfile

Octstr* bakfile = NULL

static

Definition at line 90 of file bb_store_file.c.

Referenced by rename_store(), store_dumper(), store_file_init(), and store_file_load().

cleanup_thread

long cleanup_thread = -1

static

Definition at line 92 of file bb_store_file.c.

Referenced by store_file_load(), and store_file_shutdown().

dump_frequency

long dump_frequency = 0

static

Definition at line 93 of file bb_store_file.c.

Referenced by store_dumper(), and store_file_init().

file

FILE* file = NULL

static

Definition at line 87 of file bb_store_file.c.

Referenced by do_dump(), open_file(), store_dumper(), store_file_dump(), store_file_load(), store_file_save(), and write_msg().

file_mutex

Mutex* file_mutex = NULL

static

Definition at line 91 of file bb_store_file.c.

Referenced by store_dumper(), store_file_dump(), store_file_for_each_message(), store_file_init(), store_file_load(), and store_file_save().

filename

Octstr* filename = NULL

static

Definition at line 88 of file bb_store_file.c.

Referenced by do_dump(), rename_store(), store_dumper(), store_file_for_each_message(), store_file_init(), store_file_load(), store_file_save(), store_file_save_ack(), and store_file_shutdown().

last_dict_mod

time_t last_dict_mod = 0

static

Definition at line 98 of file bb_store_file.c.

Referenced by store_dumper(), and store_to_dict().

loaded

List* loaded

static

Definition at line 99 of file bb_store_file.c.

Referenced by store_file_init(), store_file_load(), store_file_save(), and store_file_shutdown().

newfile

Octstr* newfile = NULL

static

Definition at line 89 of file bb_store_file.c.

Referenced by do_dump(), rename_store(), store_dumper(), store_file_init(), and store_file_load().

sms_dict

Dict* sms_dict = NULL

static

Definition at line 95 of file bb_store_file.c.

Referenced by do_dump(), store_dumper(), store_file_dump(), store_file_for_each_message(), store_file_init(), store_file_load(), store_file_messages(), and store_to_dict().