Re: [PATCH] msg storage abstraction to instances
Alejandro Guerrieri <[email protected]>
| Newsgroups | gmane.comp.mobile.kannel.devel |
|---|---|
| Message-ID | <[email protected]> |
Wow, this looks promising :) I'll take a look asap. Regards, -- Alejandro Guerrieri [email protected] On 01/03/2010, at 12:46, Stipe Tolj wrote: > Hi list, > > I have attached a patch against CVS HEAD, that abstracts our current code in > > gw/bb_store.[ch] > gw/bb_store_[file|spool].c > > to allow INSTANTIATION of a message storage by a context handler. I moved the > code to > > gwlib/msg_store[_p].[ch] > gwlib/msg_store_[file|spool].c > > and changed the store_foobar(...) references to msg_store_foobar(msg_store,...) > in the bearerbox specific files > > gw/bb_[boxc|http|smsconn].c > gw/bearerbox.[ch] > > to comply with the existing behavior. Hence the patch is a NLC (no logic > change), but allows using several message storage instances/contexts in any daemon. > > Example: We could implement a SQL msg storage implementation > (gwlib/msg_store_mysql.c) or on the "other" side, implement a DLR storage type > of a msg store. (gw/dlr_msg_store.c). > > Kindly have a look and vote for commitment. If there are no objections to have > the message storage more abstracted this way, I will commit the code and remove > the "old" static module approach. > > Stipe > > -- > ------------------------------------------------------------------- > Kölner Landstrasse 419 > 40589 Düsseldorf, NRW, Germany > > tolj.org system architecture Kannel Software Foundation (KSF) > http://www.tolj.org/ http://www.kannel.org/ > > mailto:st_{at}_tolj.org mailto:stolj_{at}_kannel.org > ------------------------------------------------------------------- > ### Eclipse Workspace Patch 1.0 > #P gateway-cvs-head > Index: gw/bearerbox.h > =================================================================== > RCS file: /home/cvs/gateway/gw/bearerbox.h,v > retrieving revision 1.37 > diff -u -r1.37 bearerbox.h > --- gw/bearerbox.h 31 Aug 2009 10:54:07 -0000 1.37 > +++ gw/bearerbox.h 1 Mar 2010 11:29:07 -0000 > @@ -64,6 +64,7 @@ > #include "msg.h" > #include "smscconn.h" > #include "bb_store.h" > +#include "gwlib/msg_store.h" > > /* Default outgoing queue length */ > #define DEFAULT_OUTGOING_SMS_QLENGTH 1000000 > Index: gw/bb_http.c > =================================================================== > RCS file: /home/cvs/gateway/gw/bb_http.c,v > retrieving revision 1.54 > diff -u -r1.54 bb_http.c > --- gw/bb_http.c 8 Jul 2009 14:29:39 -0000 1.54 > +++ gw/bb_http.c 1 Mar 2010 11:29:07 -0000 > @@ -73,6 +73,7 @@ > > /* passed from bearerbox core */ > > +extern MsgStore *msg_store; > extern volatile sig_atomic_t bb_status; > > /* our own thingies */ > @@ -148,7 +149,7 @@ > { > Octstr *reply; > if ((reply = httpd_check_authorization(cgivars, 1))!= NULL) return reply; > - return store_status(status_type); > + return msg_store_status(msg_store, status_type); > } > > static Octstr *httpd_loglevel(List *cgivars, int status_type) > Index: gw/bb_boxc.c > =================================================================== > RCS file: /home/cvs/gateway/gw/bb_boxc.c,v > retrieving revision 1.95 > diff -u -r1.95 bb_boxc.c > --- gw/bb_boxc.c 21 Jan 2010 16:53:02 -0000 1.95 > +++ gw/bb_boxc.c 1 Mar 2010 11:29:07 -0000 > @@ -97,6 +97,7 @@ > /* incoming/outgoing sms queue control */ > extern long max_incoming_sms_qlength; > > +extern MsgStore *msg_store; > > /* our own thingies */ > > @@ -223,7 +224,7 @@ > uuid_copy(mack->ack.id, msg->sms.id); > mack->ack.time = msg->sms.time; > > - store_save(msg); > + msg_store_save(msg_store, msg); > > rc = smsc2_rout(msg, 0); > switch (rc) { > @@ -240,7 +241,7 @@ > warning(0, "Message rejected by bearerbox, no router!"); > > /* > - * we don't store_save_ack() here, since the call to > + * we don't msg_store_save_ack() here, since the call to > * bb_smscconn_send_failed() within smsc2_route() did > * it already. > */ > @@ -258,7 +259,7 @@ > * message from store-file. > */ > mack->ack.nack = ack_failed_tmp; > - store_save_ack(msg, ack_failed_tmp); > + msg_store_save_ack(msg_store, msg, ack_failed_tmp); > > /* destroy original message */ > msg_destroy(msg); > @@ -347,7 +348,7 @@ > gwlist_append(conn->retry, orig); > } else { > boxc_sent_pop(conn, msg, NULL); > - store_save(msg); > + msg_store_save(msg_store, msg); > } > debug("bb.boxc", 0, "boxc_receiver: got ack"); > } > Index: gw/bb_smscconn.c > =================================================================== > RCS file: /home/cvs/gateway/gw/bb_smscconn.c,v > retrieving revision 1.108 > diff -u -r1.108 bb_smscconn.c > --- gw/bb_smscconn.c 15 Nov 2009 14:12:28 -0000 1.108 > +++ gw/bb_smscconn.c 1 Mar 2010 11:29:07 -0000 > @@ -122,6 +122,8 @@ > /* configuration filename */ > extern Octstr *cfg_filename; > > +extern MsgStore *msg_store; > + > /* our own thingies */ > > static volatile sig_atomic_t smsc_running; > @@ -267,7 +269,7 @@ > } > > /* write ACK to store file */ > - store_save_ack(sms, ack_success); > + msg_store_save_ack(msg_store, sms, ack_success); > > if (sms->sms.sms_type != report_mt) { > bb_alog_sms(conn, sms, "Sent SMS"); > @@ -339,7 +341,7 @@ > > default: > /* write NACK to store file */ > - store_save_ack(sms, ack_failed); > + msg_store_save_ack(msg_store, sms, ack_failed); > > if (conn) counter_increase(conn->failed); > if (reason == SMSCCONN_FAILED_DISCARDED) > @@ -436,7 +438,7 @@ > sms->sms.sms_type = mo; > > /* write to store (if enabled) */ > - if (store_save(sms) == -1) { > + if (msg_store_save(msg_store, sms) == -1) { > msg_destroy(sms); > return SMSCCONN_FAILED_TEMPORARILY; > } > @@ -497,7 +499,7 @@ > bb_alog_sms(conn, sms, "DROPPED Received SMS"); > > /* put nack into store-file */ > - store_save_ack(sms, ack_failed); > + msg_store_save_ack(msg_store, sms, ack_failed); > > msg_destroy(copy); > msg_destroy(sms); > @@ -1335,18 +1337,18 @@ > */ > if (conn->reroute) { > /* change message direction */ > - store_save_ack(msg, ack_success); > + msg_store_save_ack(msg_store, msg, ack_success); > msg->sms.sms_type = mt_push; > - store_save(msg); > + msg_store_save(msg_store, msg); > /* drop into outbound queue again for routing */ > return smsc2_rout(msg, 0); > } > > if (conn->reroute_to_smsc) { > /* change message direction */ > - store_save_ack(msg, ack_success); > + msg_store_save_ack(msg_store, msg, ack_success); > msg->sms.sms_type = mt_push; > - store_save(msg); > + msg_store_save(msg_store, msg); > /* apply directly to the given smsc-id for MT traffic */ > octstr_destroy(msg->sms.smsc_id); > msg->sms.smsc_id = octstr_duplicate(conn->reroute_to_smsc); > @@ -1356,9 +1358,9 @@ > if (conn->reroute_by_receiver && msg->sms.receiver && > (smsc = dict_get(conn->reroute_by_receiver, msg->sms.receiver))) { > /* change message direction */ > - store_save_ack(msg, ack_success); > + msg_store_save_ack(msg_store, msg, ack_success); > msg->sms.sms_type = mt_push; > - store_save(msg); > + msg_store_save(msg_store, msg); > /* route by receiver number */ > /* XXX implement wildcard matching too! */ > octstr_destroy(msg->sms.smsc_id); > @@ -1397,7 +1399,7 @@ > gw_assert(msg); > for (i = 0; i < msg->total_parts; i++) { > if (msg->parts[i]) { > - store_save_ack(msg->parts[i], msg->ack); > + msg_store_save_ack(msg_store, msg->parts[i], msg->ack); > msg_destroy(msg->parts[i]); > } > } > @@ -1462,7 +1464,7 @@ > if (x->parts[i] == NULL) > continue; > msg = msg_duplicate(x->parts[i]); > - store_save_ack(x->parts[i], ack_success); > + msg_store_save_ack(msg_store, x->parts[i], ack_success); > switch(bb_smscconn_receive(NULL, msg)) { > case SMSCCONN_FAILED_REJECTED: > case SMSCCONN_SUCCESS: > @@ -1474,7 +1476,7 @@ > case SMSCCONN_FAILED_QFULL: > default: > /* oops put it back into dict and retry on next run */ > - store_save(x->parts[i]); > + msg_store_save(msg_store, x->parts[i]); > destroy = 0; > break; > } > @@ -1517,7 +1519,7 @@ > /* Checks if message is concatenated. Returns: > * - returns concat_complete if no concat parts, or message complete > * - returns concat_pending (and sets *pmsg to NULL) if parts pending > - * - returns concat_error if store_save fails > + * - returns concat_error if msg_store_save() fails > */ > static int check_concatenation(Msg **pmsg, Octstr *smscid) > { > @@ -1588,7 +1590,7 @@ > if (cmsg->parts[part - 1] != NULL) { > warning(0, "Duplicate message part %d, ref %d, from %s, to %s. Discarded!", > part, refnum, octstr_get_cstr(msg->sms.sender), octstr_get_cstr(msg->sms.receiver)); > - store_save_ack(msg, ack_success); > + msg_store_save_ack(msg_store, msg, ack_success); > msg_destroy(msg); > *pmsg = msg = NULL; > } else { > @@ -1615,7 +1617,7 @@ > octstr_append(msg->sms.msgdata, cmsg->parts[i]->sms.msgdata); > > /* Attempt to save the new one, if that fails, then reply with fail. */ > - if (store_save(msg) == -1) { > + if (msg_store_save(msg_store, msg) == -1) { > mutex_unlock(concat_lock); > msg_destroy(msg); > *pmsg = msg = NULL; > Index: gw/bearerbox.c > =================================================================== > RCS file: /home/cvs/gateway/gw/bearerbox.c,v > retrieving revision 1.175 > diff -u -r1.175 bearerbox.c > --- gw/bearerbox.c 28 Feb 2010 22:43:32 -0000 1.175 > +++ gw/bearerbox.c 1 Mar 2010 11:29:07 -0000 > @@ -129,6 +129,9 @@ > /* configuration filename */ > Octstr *cfg_filename; > > +/* Main message store context. */ > +MsgStore *msg_store; > + > volatile sig_atomic_t bb_status; > > /* > @@ -428,8 +431,19 @@ > log = cfg_get(grp, octstr_imm("store-location")); > val = cfg_get(grp, octstr_imm("store-type")); > } > - if (store_init(val, log, store_dump_freq, msg_pack, msg_unpack_wrapper) == -1) > - panic(0, "Could not start with store init failed."); > + if (val && octstr_compare(val, octstr_imm("file")) == 0) { > + msg_store = msg_store_init(STORE_FILE, msg_pack, msg_unpack_wrapper, > + log, (long) store_dump_freq); > + if (msg_store == NULL) > + panic(0, "Could not start with store init failed."); > + } else if (val && octstr_compare(val, octstr_imm("spool")) == 0) { > + msg_store = msg_store_init(STORE_SPOOL, msg_pack, msg_unpack_wrapper, log); > + if (msg_store == NULL) > + panic(0, "Could not start with store init failed."); > + } else if (val) { > + panic(0, "Directive 'store-type = %s' not supported!", > + octstr_get_cstr(val)); > + } > octstr_destroy(val); > octstr_destroy(log); > > @@ -666,7 +680,7 @@ > > gwthread_sleep(5.0); /* give time to threads to register themselves */ > > - if (store_load(dispatch_into_queue) == -1) > + if (msg_store_load(msg_store, dispatch_into_queue) == -1) > panic(0, "Cannot start with store-file failing"); > > info(0, "MAIN: Start-up done, entering mainloop"); > @@ -728,7 +742,7 @@ > > boxc_cleanup(); > smsc2_cleanup(); > - store_shutdown(); > + msg_store_shutdown(msg_store); > empty_msg_lists(); > gwlist_destroy(flow_threads, NULL); > gwlist_destroy(suspended, NULL); > @@ -972,7 +986,7 @@ > counter_value(outgoing_wdp_counter), gwlist_len(outgoing_wdp) + udp_outgoing_queue(), > counter_value(incoming_sms_counter), gwlist_len(incoming_sms), > counter_value(outgoing_sms_counter), gwlist_len(outgoing_sms), > - store_messages(), > + msg_store_messages(msg_store), > load_get(incoming_sms_load,0), load_get(incoming_sms_load,1), load_get(incoming_sms_load,2), > load_get(outgoing_sms_load,0), load_get(outgoing_sms_load,1), load_get(outgoing_sms_load,2), > counter_value(incoming_dlr_counter), counter_value(outgoing_dlr_counter), > Index: gwlib/msg_store_file.c > =================================================================== > RCS file: gwlib/msg_store_file.c > diff -N gwlib/msg_store_file.c > --- /dev/null 1 Jan 1970 00:00:00 -0000 > +++ gwlib/msg_store_file.c 1 Jan 1970 00:00:00 -0000 > @@ -0,0 +1,649 @@ > +/* ==================================================================== > + * The Kannel Software License, Version 1.0 > + * > + * Copyright (c) 2001-2009 Kannel Group > + * Copyright (c) 1998-2001 WapIT Ltd. > + * All rights reserved. > + * > + * Redistribution and use in source and binary forms, with or without > + * modification, are permitted provided that the following conditions > + * are met: > + * > + * 1. Redistributions of source code must retain the above copyright > + * notice, this list of conditions and the following disclaimer. > + * > + * 2. Redistributions in binary form must reproduce the above copyright > + * notice, this list of conditions and the following disclaimer in > + * the documentation and/or other materials provided with the > + * distribution. > + * > + * 3. The end-user documentation included with the redistribution, > + * if any, must include the following acknowledgment: > + * "This product includes software developed by the > + * Kannel Group (http://www.kannel.org/)." > + * Alternately, this acknowledgment may appear in the software itself, > + * if and wherever such third-party acknowledgments normally appear. > + * > + * 4. The names "Kannel" and "Kannel Group" must not be used to > + * endorse or promote products derived from this software without > + * prior written permission. For written permission, please > + * contact [email protected]. > + * > + * 5. Products derived from this software may not be called "Kannel", > + * nor may "Kannel" appear in their name, without prior written > + * permission of the Kannel Group. > + * > + * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED > + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES > + * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE > + * DISCLAIMED. IN NO EVENT SHALL THE KANNEL GROUP OR ITS CONTRIBUTORS > + * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, > + * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT > + * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR > + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, > + * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE > + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, > + * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. > + * ==================================================================== > + * > + * This software consists of voluntary contributions made by many > + * individuals on behalf of the Kannel Group. For more information on > + * the Kannel Group, please see <http://www.kannel.org/>. > + * > + * Portions of this software are based upon software originally written at > + * WapIT Ltd., Helsinki, Finland for the Kannel project. > + */ > + > +/* > + * msg_store_file.c - MsgStore type with a single sotrage file > + * > + * Based on the gw/bb_store.c from Kalle Marjola in 2001. > + */ > + > +#ifdef DO_MSG_STORE > + > +#include "sms.h" > + > + > +struct msg_store_file { > + FILE *file; > + Octstr *filename; > + Octstr *newfile; > + Octstr *bakfile; > + Mutex *file_mutex; > + long cleanup_thread; > + long dump_frequency; > + > + Dict *sms_dict; > + > + int active; > + time_t last_dict_mod; > + List *loaded; > + > + /* pointer back to the MsgStore */ > + MsgStore *func; > +}; > + > +typedef struct msg_store_file msg_store_file; > + > +static int store_file_dump(void *arg); > + > + > +static void write_msg(msg_store_file *context, Msg *msg) > +{ > + Octstr *pack; > + unsigned char buf[4]; > + > + pack = context->func->store_msg_pack(msg); > + encode_network_long(buf, octstr_len(pack)); > + octstr_insert_data(pack, 0, (char*)buf, 4); > + > + octstr_print(context->file, pack); > + fflush(context->file); > + > + octstr_destroy(pack); > +} > + > + > +static int read_msg(msg_store_file *context, Msg **msg, Octstr *os, long *off) > +{ > + 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 = context->func->store_msg_unpack(pack); > + octstr_destroy(pack); > + > + if (!*msg) > + return -1; > + > + return 0; > +} > + > + > +static int open_file(msg_store_file *context) > +{ > + context->file = fopen(octstr_get_cstr(context->newfile), "w"); > + if (context->file == NULL) { > + error(errno, "Failed to open '%s' for writing, cannot create store-file", > + octstr_get_cstr(context->newfile)); > + return -1; > + } > + return 0; > +} > + > + > +static int rename_store(msg_store_file *context) > +{ > + if (rename(octstr_get_cstr(context->filename), octstr_get_cstr(context->bakfile)) == -1) { > + if (errno != ENOENT) { > + error(errno, "Failed to rename old store '%s' as '%s'", > + octstr_get_cstr(context->filename), octstr_get_cstr(context->bakfile)); > + return -1; > + } > + } > + if (rename(octstr_get_cstr(context->newfile), octstr_get_cstr(context->filename)) == -1) { > + error(errno, "Failed to rename new store '%s' as '%s'", > + octstr_get_cstr(context->newfile), octstr_get_cstr(context->filename)); > + return -1; > + } > + return 0; > +} > + > + > +static int do_dump(msg_store_file *context) > +{ > + Octstr *key; > + Msg *msg; > + List *sms_list; > + long l; > + > + if (context->filename == NULL) > + return 0; > + > + /* > + * create a new store-file and save all non-acknowledged > + * messages into it > + */ > + if (open_file(context) == -1) > + return -1; > + > + sms_list = dict_keys(context->sms_dict); > + for (l = 0; l < gwlist_len(sms_list); l++) { > + key = gwlist_get(sms_list, l); > + msg = dict_get(context->sms_dict, key); > + if (msg != NULL) > + write_msg(context, msg); > + } > + fflush(context->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(context); > +} > + > + > +/* > + * Thread to write current store to file now and then, to prevent > + * it from becoming far too big (slows startup) > + */ > +static void store_dumper(void *arg) > +{ > + msg_store_file *context = arg; > + > + time_t now; > + int busy = 0; > + > + while (context->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 - context->last_dict_mod > context->dump_frequency || busy) { > + store_file_dump(context); > + /* > + * 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. > + */ > + context->last_dict_mod = time(NULL) + 3600 * 24; > + busy = 0; > + } else { > + busy = (now - context->last_dict_mod) > 0; > + } > + gwthread_sleep(context->dump_frequency); > + } > + store_file_dump(context); > + if (context->file != NULL) > + fclose(context->file); > + octstr_destroy(context->filename); > + octstr_destroy(context->newfile); > + octstr_destroy(context->bakfile); > + mutex_destroy(context->file_mutex); > + > + dict_destroy(context->sms_dict); > + /* set all vars to NULL */ > + context->filename = context->newfile = context->bakfile = NULL; > + context->file_mutex = NULL; > + context->sms_dict = NULL; > +} > + > + > +/*------------------------------------------------------*/ > + > +static Octstr *store_file_status(void *arg, int status_type) > +{ > + msg_store_file *context = arg; > + char *frmt; > + Octstr *ret, *key; > + unsigned long l; > + struct tm tm; > + Msg *msg; > + List *keys; > + char id[UUID_STR_LEN + 1]; > + > + ret = octstr_create(""); > + > + /* set the type based header */ > + if (status_type == MSG_STORE_STATUS_HTML) { > + octstr_append_cstr(ret, "<table border=1>\n" > + "<tr><td>SMS ID</td><td>Type</td><td>Time</td><td>Sender</td><td>Receiver</td>" > + "<td>SMSC ID</td><td>BOX ID</td><td>UDH</td><td>Message</td>" > + "</tr>\n"); > + } else if (status_type == MSG_STORE_STATUS_TEXT) { > + octstr_append_cstr(ret, "[SMS ID] [Type] [Time] [Sender] [Receiver] [SMSC ID] [BOX ID] [UDH] [Message]\n"); > + } > + > + /* if there is no store-file, then don't loop in sms_store */ > + if (context->filename == NULL) > + goto finish; > + > + keys = dict_keys(context->sms_dict); > + > + for (l = 0; l < gwlist_len(keys); l++) { > + key = gwlist_get(keys, l); > + msg = dict_get(context->sms_dict, key); > + if (msg == NULL) > + continue; > + > + if (msg_type(msg) == sms) { > + > + if (status_type == MSG_STORE_STATUS_HTML) { > + frmt = "<tr><td>%s</td><td>%s</td>" > + "<td>%04d-%02d-%02d %02d:%02d:%02d</td>" > + "<td>%s</td><td>%s</td><td>%s</td>" > + "<td>%s</td><td>%s</td><td>%s</td></tr>\n"; > + } else if (status_type == MSG_STORE_STATUS_XML) { > + frmt = "<message>\n\t<id>%s</id>\n\t<type>%s</type>\n\t" > + "<time>%04d-%02d-%02d %02d:%02d:%02d</time>\n\t" > + "<sender>%s</sender>\n\t" > + "<receiver>%s</receiver>\n\t<smsc-id>%s</smsc-id>\n\t" > + "<box-id>%s</box-id>\n\t" > + "<udh-data>%s</udh-data>\n\t<msg-data>%s</msg-data>\n\t" > + "</message>\n"; > + } else { > + frmt = "[%s] [%s] [%04d-%02d-%02d %02d:%02d:%02d] [%s] [%s] [%s] [%s] [%s] [%s]\n"; > + } > + > + /* transform the time value */ > +#if LOG_TIMESTAMP_LOCALTIME > + tm = gw_localtime(msg->sms.time); > +#else > + tm = gw_gmtime(msg->sms.time); > +#endif > + if (msg->sms.udhdata) > + octstr_binary_to_hex(msg->sms.udhdata, 1); > + if (msg->sms.msgdata && > + (msg->sms.coding == DC_8BIT || msg->sms.coding == DC_UCS2 || > + (msg->sms.coding == DC_UNDEF && msg->sms.udhdata))) > + octstr_binary_to_hex(msg->sms.msgdata, 1); > + > + uuid_unparse(msg->sms.id, id); > + > + octstr_format_append(ret, frmt, id, > + (msg->sms.sms_type == mo ? "MO" : > + msg->sms.sms_type == mt_push ? "MT-PUSH" : > + msg->sms.sms_type == mt_reply ? "MT-REPLY" : > + msg->sms.sms_type == report_mo ? "DLR-MO" : > + msg->sms.sms_type == report_mt ? "DLR-MT" : ""), > + tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, > + tm.tm_hour, tm.tm_min, tm.tm_sec, > + (msg->sms.sender ? octstr_get_cstr(msg->sms.sender) : ""), > + (msg->sms.receiver ? octstr_get_cstr(msg->sms.receiver) : ""), > + (msg->sms.smsc_id ? octstr_get_cstr(msg->sms.smsc_id) : ""), > + (msg->sms.boxc_id ? octstr_get_cstr(msg->sms.boxc_id) : ""), > + (msg->sms.udhdata ? octstr_get_cstr(msg->sms.udhdata) : ""), > + (msg->sms.msgdata ? octstr_get_cstr(msg->sms.msgdata) : "")); > + > + if (msg->sms.udhdata) > + octstr_hex_to_binary(msg->sms.udhdata); > + if (msg->sms.msgdata && > + (msg->sms.coding == DC_8BIT || msg->sms.coding == DC_UCS2 || > + (msg->sms.coding == DC_UNDEF && msg->sms.udhdata))) > + octstr_hex_to_binary(msg->sms.msgdata); > + } > + } > + gwlist_destroy(keys, octstr_destroy_item); > + > +finish: > + /* set the type based footer */ > + if (status_type == MSG_STORE_STATUS_HTML) { > + octstr_append_cstr(ret,"</table>"); > + } > + > + return ret; > +} > + > + > +static long store_file_messages(void *arg) > +{ > + msg_store_file *context = arg; > + > + return (context->sms_dict ? dict_key_count(context->sms_dict) : -1); > +} > + > + > +static int store_to_dict(msg_store_file *context, Msg *msg) > +{ > + 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(context->sms_dict, uuid_os, copy); > + octstr_destroy(uuid_os); > + context->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(context->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); > + context->last_dict_mod = time(NULL); > + } > + } else > + return -1; > + return 0; > +} > + > +static int store_file_save(void *arg, Msg *msg) > +{ > + msg_store_file *context = arg; > + > + if (context->filename == NULL) > + return 0; > + > + /* block here until store not loaded */ > + gwlist_consume(context->loaded); > + > + /* lock file_mutex in order to have dict and file in sync */ > + mutex_lock(context->file_mutex); > + if (store_to_dict(context, msg) == -1) { > + mutex_unlock(context->file_mutex); > + return -1; > + } > + > + /* write to file, too */ > + write_msg(context, msg); > + fflush(context->file); > + mutex_unlock(context->file_mutex); > + > + return 0; > +} > + > + > +static int store_file_save_ack(void *arg, Msg *msg, ack_status_t status) > +{ > + msg_store_file *context = arg; > + Msg *mack; > + int ret; > + > + /* only sms are handled */ > + if (!msg || msg_type(msg) != sms) > + return -1; > + > + if (context->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_file_save(context, mack); > + msg_destroy(mack); > + > + return ret; > +} > + > + > +static int store_file_load(void *arg, void (*receive_msg)(Msg*)) > +{ > + msg_store_file *context = arg; > + List *keys; > + Octstr *store_file, *key; > + Msg *msg; > + int retval, msgs; > + long end, pos; > + > + gw_assert(context != NULL); > + > + if (context->filename == NULL) > + return 0; > + > + mutex_lock(context->file_mutex); > + if (context->file != NULL) { > + fclose(context->file); > + context->file = NULL; > + } > + > + store_file = octstr_read_file(octstr_get_cstr(context->filename)); > + if (store_file != NULL) > + info(0, "Loading store file `%s'", octstr_get_cstr(context->filename)); > + else { > + store_file = octstr_read_file(octstr_get_cstr(context->newfile)); > + if (store_file != NULL) > + info(0, "Loading store file `%s'", octstr_get_cstr(context->newfile)); > + else { > + store_file = octstr_read_file(octstr_get_cstr(context->bakfile)); > + if (store_file != NULL) > + info(0, "Loading store file `%s'", octstr_get_cstr(context->bakfile)); > + else { > + info(0, "Cannot open any store file, starting a new one"); > + retval = open_file(context); > + 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(context, &msg, store_file, &pos) == -1) { > + error(0, "Garbage at store-file, skipped."); > + continue; > + } > + if (msg_type(msg) == sms) { > + store_to_dict(context, msg); > + msgs++; > + } else if (msg_type(msg) == ack) { > + store_to_dict(context, 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(context->sms_dict)); > + > + /* now create a new sms_store out of messages left */ > + > + keys = dict_keys(context->sms_dict); > + while ((key = gwlist_extract_first(keys)) != NULL) { > + msg = dict_remove(context->sms_dict, key); > + if (store_to_dict(context, 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(context); > + > +end: > + mutex_unlock(context->file_mutex); > + > + /* allow using of store */ > + gwlist_remove_producer(context->loaded); > + > + /* start dumper thread */ > + if ((context->cleanup_thread = gwthread_create(store_dumper, context)) == -1) > + panic(0, "Failed to create a cleanup thread!"); > + > + return retval; > +} > + > + > +static int store_file_dump(void *arg) > +{ > + msg_store_file *context = arg; > + int retval; > + > + debug("msg.store", 0, "Dumping %ld messages to store", > + dict_key_count(context->sms_dict)); > + mutex_lock(context->file_mutex); > + if (context->file != NULL) { > + fclose(context->file); > + context->file = NULL; > + } > + retval = do_dump(context); > + mutex_unlock(context->file_mutex); > + > + return retval; > +} > + > + > +static void store_file_shutdown(void *arg) > +{ > + msg_store_file *context = arg; > + > + if (context->filename == NULL) > + return; > + > + context->active = 0; > + gwthread_wakeup(context->cleanup_thread); > + /* wait for cleanup thread */ > + if (context->cleanup_thread != -1) > + gwthread_join(context->cleanup_thread); > + > + gwlist_destroy(context->loaded, NULL); > + > + gw_free(context); > +} > + > + > +static int store_file_init(MsgStore *ms, va_list ap) > +{ > + msg_store_file *msf; > + Octstr *fname = NULL; > + long dump_freq = -1; > + > + /* Get our required arguments. */ > + fname = va_arg(ap, Octstr*); > + dump_freq = va_arg(ap, long); > + va_end(ap); > + > + /* Constraint checks first. */ > + if (fname == NULL) > + panic(0, "Message store for store-file needs a file name."); > + > + if (octstr_len(fname) > (FILENAME_MAX-5)) > + panic(0, "Message store filename too long: `%s', failed to init.", > + octstr_get_cstr(fname)); > + > + /* Initialize the type specific data. */ > + msf = gw_malloc(sizeof(*msf)); > + msf->file = NULL; > + msf->filename = octstr_duplicate(fname); > + msf->newfile = octstr_format("%s.new", octstr_get_cstr(fname)); > + msf->bakfile = octstr_format("%s.bak", octstr_get_cstr(fname)); > + msf->sms_dict = dict_create(1024, msg_destroy_item); > + msf->cleanup_thread = -1; > + msf->dump_frequency = (dump_freq > 0 ? dump_freq : MSG_STORE_DEFAULT_DUMP_FREQ); > + msf->file_mutex = mutex_create(); > + msf->active = 1; > + msf->loaded = gwlist_create(); > + gwlist_add_producer(msf->loaded); > + > + msf->func = ms; > + ms->context = msf; > + > + return 1; > +} > + > + > +static struct store_ops file_ops = { > + .store_init = store_file_init, > + .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_status = store_file_status > +}; > + > +#endif /* DO_MSG_STORE */ > Index: gwlib/msg_store_p.h > =================================================================== > RCS file: gwlib/msg_store_p.h > diff -N gwlib/msg_store_p.h > --- /dev/null 1 Jan 1970 00:00:00 -0000 > +++ gwlib/msg_store_p.h 1 Jan 1970 00:00:00 -0000 > @@ -0,0 +1,107 @@ > +/* ==================================================================== > + * The Kannel Software License, Version 1.0 > + * > + * Copyright (c) 2001-2009 Kannel Group > + * Copyright (c) 1998-2001 WapIT Ltd. > + * All rights reserved. > + * > + * Redistribution and use in source and binary forms, with or without > + * modification, are permitted provided that the following conditions > + * are met: > + * > + * 1. Redistributions of source code must retain the above copyright > + * notice, this list of conditions and the following disclaimer. > + * > + * 2. Redistributions in binary form must reproduce the above copyright > + * notice, this list of conditions and the following disclaimer in > + * the documentation and/or other materials provided with the > + * distribution. > + * > + * 3. The end-user documentation included with the redistribution, > + * if any, must include the following acknowledgment: > + * "This product includes software developed by the > + * Kannel Group (http://www.kannel.org/)." > + * Alternately, this acknowledgment may appear in the software itself, > + * if and wherever such third-party acknowledgments normally appear. > + * > + * 4. The names "Kannel" and "Kannel Group" must not be used to > + * endorse or promote products derived from this software without > + * prior written permission. For written permission, please > + * contact [email protected]. > + * > + * 5. Products derived from this software may not be called "Kannel", > + * nor may "Kannel" appear in their name, without prior written > + * permission of the Kannel Group. > + * > + * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED > + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES > + * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE > + * DISCLAIMED. IN NO EVENT SHALL THE KANNEL GROUP OR ITS CONTRIBUTORS > + * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, > + * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT > + * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR > + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, > + * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE > + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, > + * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. > + * ==================================================================== > + * > + * This software consists of voluntary contributions made by many > + * individuals on behalf of the Kannel Group. For more information on > + * the Kannel Group, please see <http://www.kannel.org/>. > + * > + * Portions of this software are based upon software originally written at > + * WapIT Ltd., Helsinki, Finland for the Kannel project. > + */ > + > +/* > + * msg_store_p.h - abstracted Msg storage, based on the gw/bb_store.c code. > + * > + * Alexander Malysh <amalysh at kannel dot org>, 2006 > + * Stipe Tolj <stolj at kannel dot org>, 2010 > + */ > + > +#ifndef MSG_STORE_P_H > +#define MSG_STORE_P_H > + > +#define MSG_STORE_DEFAULT_DUMP_FREQ 10 > + > +/* type of output given by various status functions */ > +enum { > + MSG_STORE_STATUS_HTML = 0, > + MSG_STORE_STATUS_TEXT = 1, > + MSG_STORE_STATUS_WML = 2, > + MSG_STORE_STATUS_XML = 3 > +}; > + > + > +struct store_ops { > + int (*store_init)(MsgStore *ms, va_list ap); > + long (*store_messages)(void *arg); > + int (*store_save)(void *arg, Msg *msg); > + int (*store_save_ack)(void *arg, Msg *msg, ack_status_t status); > + int (*store_load)(void *arg, void (*receive_msg)(Msg*)); > + int (*store_dump)(void *arg); > + void (*store_shutdown)(void *arg); > + Octstr* (*store_status)(void *arg, int status_type); > +}; > + > + > +struct MsgStore { > + > + /* Type of storage. */ > + enum store_type store_type; > + > + /* Type specific data structure. */ > + void *context; > + > + /* Type specific operations. */ > + struct store_ops *store_ops; > + > + /* Function pointers. */ > + Octstr* (*store_msg_pack)(Msg *msg); > + Msg* (*store_msg_unpack)(Octstr *os); > +}; > + > + > +#endif /* MSG_STORE_P_H */ > Index: gwlib/msg_store.h > =================================================================== > RCS file: gwlib/msg_store.h > diff -N gwlib/msg_store.h > --- /dev/null 1 Jan 1970 00:00:00 -0000 > +++ gwlib/msg_store.h 1 Jan 1970 00:00:00 -0000 > @@ -0,0 +1,93 @@ > +/* ==================================================================== > + * The Kannel Software License, Version 1.0 > + * > + * Copyright (c) 2001-2009 Kannel Group > + * Copyright (c) 1998-2001 WapIT Ltd. > + * All rights reserved. > + * > + * Redistribution and use in source and binary forms, with or without > + * modification, are permitted provided that the following conditions > + * are met: > + * > + * 1. Redistributions of source code must retain the above copyright > + * notice, this list of conditions and the following disclaimer. > + * > + * 2. Redistributions in binary form must reproduce the above copyright > + * notice, this list of conditions and the following disclaimer in > + * the documentation and/or other materials provided with the > + * distribution. > + * > + * 3. The end-user documentation included with the redistribution, > + * if any, must include the following acknowledgment: > + * "This product includes software developed by the > + * Kannel Group (http://www.kannel.org/)." > + * Alternately, this acknowledgment may appear in the software itself, > + * if and wherever such third-party acknowledgments normally appear. > + * > + * 4. The names "Kannel" and "Kannel Group" must not be used to > + * endorse or promote products derived from this software without > + * prior written permission. For written permission, please > + * contact [email protected]. > + * > + * 5. Products derived from this software may not be called "Kannel", > + * nor may "Kannel" appear in their name, without prior written > + * permission of the Kannel Group. > + * > + * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED > + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES > + * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE > + * DISCLAIMED. IN NO EVENT SHALL THE KANNEL GROUP OR ITS CONTRIBUTORS > + * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, > + * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT > + * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR > + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, > + * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE > + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, > + * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. > + * ==================================================================== > + * > + * This software consists of voluntary contributions made by many > + * individuals on behalf of the Kannel Group. For more information on > + * the Kannel Group, please see <http://www.kannel.org/>. > + * > + * Portions of this software are based upon software originally written at > + * WapIT Ltd., Helsinki, Finland for the Kannel project. > + */ > + > +/* > + * msg_store.h - abstracted Msg storage, based on the gw/bb_store.c code. > + * > + * Alexander Malysh <amalysh at kannel dot org>, 2006 > + * Stipe Tolj <stolj at kannel dot org>, 2010 > + */ > + > +#ifndef MSG_STORE_H > +#define MSG_STORE_H > + > + > +/* Supported storage types. */ > +enum store_type { > + STORE_FILE, STORE_SPOOL > +}; > + > +typedef struct MsgStore MsgStore; > + > + > +/* > + * Initialize the message storage of the specified type. > + * Supported type values: 'spool', 'file'. > + * Returns the pointer to the message storage instance, or NULL if > + * any error occured. > + */ > +MsgStore *msg_store_init(enum store_type type, void *pack_func, void *unpack_func, ...); > + > +void msg_store_shutdown(MsgStore *ms); > +long msg_store_messages(MsgStore *ms); > +int msg_store_save(MsgStore *ms, Msg *msg); > +int msg_store_save_ack(MsgStore *ms, Msg *msg, ack_status_t status); > +int msg_store_load(MsgStore *ms, void (*receive_msg)(Msg*)); > +int msg_store_dump(MsgStore *ms); > +Octstr *msg_store_status(MsgStore *ms, int status_type); > + > + > +#endif /* MSG_STORE_H */ > Index: gwlib/msg_store.c > =================================================================== > RCS file: gwlib/msg_store.c > diff -N gwlib/msg_store.c > --- /dev/null 1 Jan 1970 00:00:00 -0000 > +++ gwlib/msg_store.c 1 Jan 1970 00:00:00 -0000 > @@ -0,0 +1,150 @@ > +/* ==================================================================== > + * The Kannel Software License, Version 1.0 > + * > + * Copyright (c) 2001-2009 Kannel Group > + * Copyright (c) 1998-2001 WapIT Ltd. > + * All rights reserved. > + * > + * Redistribution and use in source and binary forms, with or without > + * modification, are permitted provided that the following conditions > + * are met: > + * > + * 1. Redistributions of source code must retain the above copyright > + * notice, this list of conditions and the following disclaimer. > + * > + * 2. Redistributions in binary form must reproduce the above copyright > + * notice, this list of conditions and the following disclaimer in > + * the documentation and/or other materials provided with the > + * distribution. > + * > + * 3. The end-user documentation included with the redistribution, > + * if any, must include the following acknowledgment: > + * "This product includes software developed by the > + * Kannel Group (http://www.kannel.org/)." > + * Alternately, this acknowledgment may appear in the software itself, > + * if and wherever such third-party acknowledgments normally appear. > + * > + * 4. The names "Kannel" and "Kannel Group" must not be used to > + * endorse or promote products derived from this software without > + * prior written permission. For written permission, please > + * contact [email protected]. > + * > + * 5. Products derived from this software may not be called "Kannel", > + * nor may "Kannel" appear in their name, without prior written > + * permission of the Kannel Group. > + * > + * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED > + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES > + * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE > + * DISCLAIMED. IN NO EVENT SHALL THE KANNEL GROUP OR ITS CONTRIBUTORS > + * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, > + * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT > + * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR > + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, > + * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE > + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, > + * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. > + * ==================================================================== > + * > + * This software consists of voluntary contributions made by many > + * individuals on behalf of the Kannel Group. For more information on > + * the Kannel Group, please see <http://www.kannel.org/>. > + * > + * Portions of this software are based upon software originally written at > + * WapIT Ltd., Helsinki, Finland for the Kannel project. > + */ > + > +/* > + * msg_store.c - abstracted Msg storage, based on the gw/bb_store.c code. > + * > + * Alexander Malysh <amalysh at kannel dot org>, 2006 > + * Stipe Tolj <stolj at kannel dot org>, 2010 > + */ > + > +#include <errno.h> > + > +#include "gw-config.h" > + > +#include "gwlib/gwlib.h" > +#include "msg.h" > +#include "msg_store.h" > +#include "msg_store_p.h" > + > +#define DO_MSG_STORE 1 > + > +#include "msg_store_file.c" > +#include "msg_store_spool.c" > + > + > +MsgStore *msg_store_init(enum store_type type, void *pack_func, void *unpack_func, ...) > +{ > + MsgStore *ms; > + va_list ap; > + va_list aq; > + > + ms = gw_malloc(sizeof(*ms)); > + ms->store_msg_pack = pack_func; > + ms->store_msg_unpack = unpack_func; > + > + switch (type) { > + case STORE_FILE: > + ms->store_ops = &file_ops; > + break; > + case STORE_SPOOL: > + ms->store_ops = &spool_ops; > + break; > + default: > + panic(0, "Unknown 'store-type' defined."); > + break; > + } > + > + va_start(ap, unpack_func); > + va_copy(aq, ap); > + ms->store_ops->store_init(ms, aq); > + va_end(aq); > + va_end(ap); > + > + return ms; > +} > + > + > +inline void msg_store_shutdown(MsgStore *ms) > +{ > + gw_assert(ms != NULL); > + > + /* Destroy the context. */ > + ms->store_ops->store_shutdown(ms->context); > + > + /* Destroy our own structure. */ > + gw_free(ms); > +} > + > +inline long msg_store_messages(MsgStore *ms) > +{ > + return ms->store_ops->store_messages(ms->context); > +} > + > +inline int msg_store_save(MsgStore *ms, Msg *msg) > +{ > + return ms->store_ops->store_save(ms->context, msg); > +} > + > +inline int msg_store_save_ack(MsgStore *ms, Msg *msg, ack_status_t status) > +{ > + return ms->store_ops->store_save_ack(ms->context, msg, status); > +} > + > +inline int msg_store_load(MsgStore *ms, void (*receive_msg)(Msg*)) > +{ > + return ms->store_ops->store_load(ms->context, receive_msg); > +} > + > +inline int msg_store_dump(MsgStore *ms) > +{ > + return ms->store_ops->store_dump(ms->context); > +} > + > +inline Octstr *msg_store_status(MsgStore *ms, int status_type) > +{ > + return ms->store_ops->store_status(ms->context, status_type); > +} > Index: gwlib/msg_store_spool.c > =================================================================== > RCS file: gwlib/msg_store_spool.c > diff -N gwlib/msg_store_spool.c > --- /dev/null 1 Jan 1970 00:00:00 -0000 > +++ gwlib/msg_store_spool.c 1 Jan 1970 00:00:00 -0000 > @@ -0,0 +1,458 @@ > +/* ==================================================================== > + * The Kannel Software License, Version 1.0 > + * > + * Copyright (c) 2001-2009 Kannel Group > + * Copyright (c) 1998-2001 WapIT Ltd. > + * All rights reserved. > + * > + * Redistribution and use in source and binary forms, with or without > + * modification, are permitted provided that the following conditions > + * are met: > + * > + * 1. Redistributions of source code must retain the above copyright > + * notice, this list of conditions and the following disclaimer. > + * > + * 2. Redistributions in binary form must reproduce the above copyright > + * notice, this list of conditions and the following disclaimer in > + * the documentation and/or other materials provided with the > + * distribution. > + * > + * 3. The end-user documentation included with the redistribution, > + * if any, must include the following acknowledgment: > + * "This product includes software developed by the > + * Kannel Group (http://www.kannel.org/)." > + * Alternately, this acknowledgment may appear in the software itself, > + * if and wherever such third-party acknowledgments normally appear. > + * > + * 4. The names "Kannel" and "Kannel Group" must not be used to > + * endorse or promote products derived from this software without > + * prior written permission. For written permission, please > + * contact [email protected]. > + * > + * 5. Products derived from this software may not be called "Kannel", > + * nor may "Kannel" appear in their name, without prior written > + * permission of the Kannel Group. > + * > + * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED > + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES > + * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE > + * DISCLAIMED. IN NO EVENT SHALL THE KANNEL GROUP OR ITS CONTRIBUTORS > + * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, > + * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT > + * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR > + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, > + * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE > + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, > + * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. > + * ==================================================================== > + * > + * This software consists of voluntary contributions made by many > + * individuals on behalf of the Kannel Group. For more information on > + * the Kannel Group, please see <http://www.kannel.org/>. > + * > + * Portions of this software are based upon software originally written at > + * WapIT Ltd., Helsinki, Finland for the Kannel project. > + */ > + > +/** > + * msg_store_spool.c - MsgStore type storage/retrieval module using spool directory > + * > + * Author: Alexander Malysh, 2006 > + * Modified to MsgStore abstraction: Stipe Tolj, 2010 > + */ > + > +#ifdef DO_MSG_STORE > + > +#include <unistd.h> > +#include <dirent.h> > +#include <fcntl.h> > + > + > +/* how much subdirs allowed ? */ > +#define MAX_DIRS 100 > + > +struct msg_store_spool { > + Octstr *spool; > + Counter *counter; > + List *loaded; > + > + /* pointer back to the MsgStore */ > + MsgStore *func; > +}; > + > +typedef struct msg_store_spool msg_store_spool; > + > + > +static int store_spool_dump(void *arg) > +{ > + /* This type doesn't dump at all */ > + return 0; > +} > + > + > +static long store_spool_messages(void *arg) > +{ > + msg_store_spool *context = arg; > + > + return context->counter ? counter_value(context->counter) : -1; > +} > + > + > +static int for_each_file(msg_store_spool *context, const Octstr *dir_s, int ignore_err, > + void (*cb)(msg_store_spool*, const Octstr*, void*), void *data) > +{ > + DIR *dir; > + struct dirent *ent; > + struct stat stat; > + int ret = 0; > + > + if ((dir = opendir(octstr_get_cstr(dir_s))) == NULL) { > + error(errno, "Could not open directory `%s'", octstr_get_cstr(dir_s)); > + return -1; > + } > + while((ent = readdir(dir)) != NULL) { > + Octstr *filename; > + if (*(ent->d_name) == '.') /* skip hidden files */ > + continue; > + filename = octstr_format("%S/%s", dir_s, ent->d_name); > + if (lstat(octstr_get_cstr(filename), &stat) == -1) { > + if (!ignore_err) > + error(errno, "Could not get stat for `%s'", octstr_get_cstr(filename)); > + ret = -1; > + } else if (S_ISDIR(stat.st_mode) && > + for_each_file(context, filename, ignore_err, cb, data) == -1) { > + ret = -1; > + } else if (S_ISREG(stat.st_mode) && cb != NULL) > + cb(context, filename, data); > + octstr_destroy(filename); > + if (ret == -1 && ignore_err) > + ret = 0; > + else if (ret == -1) > + break; > + } > + closedir(dir); > + > + return ret; > +} > + > + > +struct status { > + const char *format; > + Octstr *status; > +}; > + > + > +static void status_cb(msg_store_spool *context, const Octstr *filename, void *d) > +{ > + struct status *data = d; > + struct tm tm; > + char id[UUID_STR_LEN + 1]; > + Octstr *msg_s; > + Msg *msg; > + > + msg_s = octstr_read_file(octstr_get_cstr(filename)); > + msg = context->func->store_msg_unpack(msg_s); > + octstr_destroy(msg_s); > + if (msg == NULL) > + return; > + > + /* transform the time value */ > +#if LOG_TIMESTAMP_LOCALTIME > + tm = gw_localtime(msg->sms.time); > +#else > + tm = gw_gmtime(msg->sms.time); > +#endif > + if (msg->sms.udhdata) > + octstr_binary_to_hex(msg->sms.udhdata, 1); > + if (msg->sms.msgdata && > + (msg->sms.coding == DC_8BIT || msg->sms.coding == DC_UCS2 || > + (msg->sms.coding == DC_UNDEF && msg->sms.udhdata))) > + octstr_binary_to_hex(msg->sms.msgdata, 1); > + > + uuid_unparse(msg->sms.id, id); > + > + octstr_format_append(data->status, data->format, > + id, > + (msg->sms.sms_type == mo ? "MO" : > + msg->sms.sms_type == mt_push ? "MT-PUSH" : > + msg->sms.sms_type == mt_reply ? "MT-REPLY" : > + msg->sms.sms_type == report_mo ? "DLR-MO" : > + msg->sms.sms_type == report_mt ? "DLR-MT" : ""), > + tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, > + tm.tm_hour, tm.tm_min, tm.tm_sec, > + (msg->sms.sender ? octstr_get_cstr(msg->sms.sender) : ""), > + (msg->sms.receiver ? octstr_get_cstr(msg->sms.receiver) : ""), > + (msg->sms.smsc_id ? octstr_get_cstr(msg->sms.smsc_id) : ""), > + (msg->sms.boxc_id ? octstr_get_cstr(msg->sms.boxc_id) : ""), > + (msg->sms.udhdata ? octstr_get_cstr(msg->sms.udhdata) : ""), > + (msg->sms.msgdata ? octstr_get_cstr(msg->sms.msgdata) : "")); > + > + msg_destroy(msg); > +} > + > + > +static Octstr *store_spool_status(void *arg, int status_type) > +{ > + msg_store_spool *context = arg; > + Octstr *ret = octstr_create(""); > + const char *format; > + struct status data; > + > + /* check if we are active */ > + if (context->spool == NULL) > + return ret; > + > + /* set the type based header */ > + if (status_type == MSG_STORE_STATUS_HTML) { > + octstr_append_cstr(ret, "<table border=1>\n" > + "<tr><td>SMS ID</td><td>Type</td><td>Time</td><td>Sender</td><td>Receiver</td>" > + "<td>SMSC ID</td><td>BOX ID</td><td>UDH</td><td>Message</td>" > + "</tr>\n"); > + > + format = "<tr><td>%s</td><td>%s</td>" > + "<td>%04d-%02d-%02d %02d:%02d:%02d</td>" > + "<td>%s</td><td>%s</td><td>%s</td>" > + "<td>%s</td><td>%s</td><td>%s</td></tr>\n"; > + } else if (status_type == MSG_STORE_STATUS_XML) { > + format = "<message>\n\t<id>%s</id>\n\t<type>%s</type>\n\t" > + "<time>%04d-%02d-%02d %02d:%02d:%02d</time>\n\t" > + "<sender>%s</sender>\n\t" > + "<receiver>%s</receiver>\n\t<smsc-id>%s</smsc-id>\n\t" > + "<box-id>%s</box-id>\n\t" > + "<udh-data>%s</udh-data>\n\t<msg-data>%s</msg-data>\n\t" > + "</message>\n"; > + } else { > + octstr_append_cstr(ret, "[SMS ID] [Type] [Time] [Sender] [Receiver] [SMSC ID] [BOX ID] [UDH] [Message]\n"); > + format = "[%s] [%s] [%04d-%02d-%02d %02d:%02d:%02d] [%s] [%s] [%s] [%s] [%s] [%s]\n"; > + } > + > + data.format = format; > + data.status = ret; > + /* ignore error because files may disappear */ > + for_each_file(context, context->spool, 1, status_cb, &data); > + > + /* set the type based footer */ > + if (status_type == MSG_STORE_STATUS_HTML) { > + octstr_append_cstr(ret,"</table>"); > + } > + > + return ret; > +} > + > + > +static void dispatch(msg_store_spool *context, const Octstr *filename, void *data) > +{ > + Octstr *msg_s; > + Msg *msg; > + void (*receive_msg)(Msg*) = data; > + > + /* debug("", 0, "dispatch(%s,...) called", octstr_get_cstr(filename)); */ > + > + msg_s = octstr_read_file(octstr_get_cstr(filename)); > + if (msg_s == NULL) > + return; > + msg = context->func->store_msg_unpack(msg_s); > + octstr_destroy(msg_s); > + if (msg != NULL) { > + receive_msg(msg); > + counter_increase(context->counter); > + } else { > + error(0, "Could not unpack message `%s'", octstr_get_cstr(filename)); > + } > +} > + > + > +static int store_spool_load(void *arg, void (*receive_msg)(Msg*)) > +{ > + msg_store_spool *context = arg; > + int rc; > + > + /* check if we are active */ > + if (context->spool == NULL) > + return 0; > + > + /* sanity check */ > + if (receive_msg == NULL) > + return -1; > + > + rc = for_each_file(context, context->spool, 0, dispatch, receive_msg); > + > + info(0, "Loaded %ld messages from store.", counter_value(context->counter)); > + > + /* allow using of storage */ > + gwlist_remove_producer(context->loaded); > + > + return rc; > +} > + > + > +static int store_spool_save(void *arg, Msg *msg) > +{ > + msg_store_spool *context = arg; > + char id[UUID_STR_LEN + 1]; > + Octstr *id_s; > + > + /* 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 (context->spool == NULL) > + return 0; > + > + /* Block here if the store isn't loaded still */ > + gwlist_consume(context->loaded); > + > + switch (msg_type(msg)) { > + case sms: > + { > + Octstr *os = context->func->store_msg_pack(msg); > + Octstr *filename, *dir; > + int fd; > + size_t wrc; > + > + if (os == NULL) { > + error(0, "Could not pack message."); > + return -1; > + } > + uuid_unparse(msg->sms.id, id); > + id_s = octstr_create(id); > + dir = octstr_format("%S/%ld", context->spool, octstr_hash_key(id_s) % MAX_DIRS); > + octstr_destroy(id_s); > + if (mkdir(octstr_get_cstr(dir), S_IRUSR|S_IWUSR|S_IXUSR) == -1 && errno != EEXIST) { > + error(errno, "Could not create directory `%s'.", octstr_get_cstr(dir)); > + octstr_destroy(dir); > + octstr_destroy(os); > + return -1; > + } > + filename = octstr_format("%S/%s", dir, id); > + octstr_destroy(dir); > + if ((fd = open(octstr_get_cstr(filename), O_CREAT|O_EXCL|O_WRONLY, S_IRUSR|S_IWUSR)) == -1) { > + error(errno, "Could not open file `%s'.", octstr_get_cstr(filename)); > + octstr_destroy(filename); > + octstr_destroy(os); > + return -1; > + } > + for (wrc = 0; wrc < octstr_len(os); ) { > + size_t rc = write(fd, octstr_get_cstr(os) + wrc, octstr_len(os) - wrc); > + if (rc == -1) { > + /* remove file */ > + error(errno, "Could not write message to `%s'.", octstr_get_cstr(filename)); > + close(fd); > + if (unlink(octstr_get_cstr(filename)) == -1) > + error(errno, "Oops, Could not remove failed file `%s'.", octstr_get_cstr(filename)); > + octstr_destroy(os); > + octstr_destroy(filename); > + return -1; > + } > + wrc += rc; > + } > + close(fd); > + counter_increase(context->counter); > + octstr_destroy(filename); > + octstr_destroy(os); > + break; > + } > + case ack: > + { > + Octstr *filename; > + uuid_unparse(msg->ack.id, id); > + id_s = octstr_create(id); > + filename = octstr_format("%S/%ld/%s", context->spool, octstr_hash_key(id_s) % MAX_DIRS, id); > + octstr_destroy(id_s); > + if (unlink(octstr_get_cstr(filename)) == -1) { > + error(errno, "Could not unlink file `%s'.", octstr_get_cstr(filename)); > + octstr_destroy(filename); > + return -1; > + } > + counter_decrease(context->counter); > + octstr_destroy(filename); > + break; > + } > + default: > + return -1; > + } > + > + return 0; > +} > + > + > +static int store_spool_save_ack(void *arg, Msg *msg, ack_status_t status) > +{ > + msg_store_spool *context = arg; > + int ret; > + Msg *nack = msg_create(ack); > + > + nack->ack.nack = status; > + uuid_copy(nack->ack.id, msg->sms.id); > + nack->ack.time = msg->sms.time; > + ret = store_spool_save(context, nack); > + msg_destroy(nack); > + > + return ret; > +} > + > + > +static void store_spool_shutdown(void *arg) > +{ > + msg_store_spool *context = arg; > + > + if (context->spool == NULL) > + return; > + > + counter_destroy(context->counter); > + octstr_destroy(context->spool); > + gwlist_destroy(context->loaded, NULL); > + > + gw_free(context); > +} > + > + > +static int store_spool_init(MsgStore *ms, va_list ap) > +{ > + msg_store_spool *mss; > + Octstr *store_dir; > + DIR *dir; > + > + /* Get our required arguments. */ > + store_dir = va_arg(ap, Octstr*); > + va_end(ap); > + > + /* Constraint checks first. */ > + if (store_dir == NULL) > + panic(0, "Message store for spool-dir needs a directory name."); > + > + /* Check if we can open directory */ > + if ((dir = opendir(octstr_get_cstr(store_dir))) == NULL) { > + panic(errno, "Could not open directory `%s'", octstr_get_cstr(store_dir)); > + } > + closedir(dir); > + > + /* Initialize the type specific data. */ > + mss = gw_malloc(sizeof(*mss)); > + mss->loaded = gwlist_create(); > + gwlist_add_producer(mss->loaded); > + mss->spool = octstr_duplicate(store_dir); > + mss->counter = counter_create(); > + > + mss->func = ms; > + ms->context = mss; > + > + return 1; > +} > + > + > +static struct store_ops spool_ops = { > + .store_init = store_spool_init, > + .store_messages = store_spool_messages, > + .store_save = store_spool_save, > + .store_save_ack = store_spool_save_ack, > + .store_load = store_spool_load, > + .store_dump = store_spool_dump, > + .store_shutdown = store_spool_shutdown, > + .store_status = store_spool_status > +}; > + > +#endif /* DO_MSG_STORE */ > Index: test/test_msg_store_dump.c > =================================================================== > RCS file: test/test_msg_store_dump.c > diff -N test/test_msg_store_dump.c > --- /dev/null 1 Jan 1970 00:00:00 -0000 > +++ test/test_msg_store_dump.c 1 Jan 1970 00:00:00 -0000 > @@ -0,0 +1,115 @@ > +/* ==================================================================== > + * The Kannel Software License, Version 1.0 > + * > + * Copyright (c) 2001-2009 Kannel Group > + * Copyright (c) 1998-2001 WapIT Ltd. > + * All rights reserved. > + * > + * Redistribution and use in source and binary forms, with or without > + * modification, are permitted provided that the following conditions > + * are met: > + * > + * 1. Redistributions of source code must retain the above copyright > + * notice, this list of conditions and the following disclaimer. > + * > + * 2. Redistributions in binary form must reproduce the above copyright > + * notice, this list of conditions and the following disclaimer in > + * the documentation and/or other materials provided with the > + * distribution. > + * > + * 3. The end-user documentation included with the redistribution, > + * if any, must include the following acknowledgment: > + * "This product includes software developed by the > + * Kannel Group (http://www.kannel.org/)." > + * Alternately, this acknowledgment may appear in the software itself, > + * if and wherever such third-party acknowledgments normally appear. > + * > + * 4. The names "Kannel" and "Kannel Group" must not be used to > + * endorse or promote products derived from this software without > + * prior written permission. For written permission, please > + * contact [email protected]. > + * > + * 5. Products derived from this software may not be called "Kannel", > + * nor may "Kannel" appear in their name, without prior written > + * permission of the Kannel Group. > + * > + * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED > + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES > + * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE > + * DISCLAIMED. IN NO EVENT SHALL THE KANNEL GROUP OR ITS CONTRIBUTORS > + * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, > + * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT > + * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR > + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, > + * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE > + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, > + * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. > + * ==================================================================== > + * > + * This software consists of voluntary contributions made by many > + * individuals on behalf of the Kannel Group. For more information on > + * the Kannel Group, please see <http://www.kannel.org/>. > + * > + * Portions of this software are based upon software originally written at > + * WapIT Ltd., Helsinki, Finland for the Kannel project. > + */ > + > +/* > + * Simple tool that prints out content of a kannel storefile. > + * > + * Tobias Weber <[email protected]> > + * Stipe Tolj <[email protected]> > + */ > + > +#include "gwlib/gwlib.h" > +#include "shared.h" > +#include "bearerbox.h" > + > +#include "gwlib/msg_store.h" > + > +static int counter = 0; > + > +static void print_msg(Msg *msg) > +{ > + counter++; > + msg_dump(msg, 0); > +} > + > +/* void function to make gwlib happy */ > +static int check_args(int i, int argc, char **argv) { > + return -1; > +} > + > +int main(int argc, char **argv) > +{ > + int cf_index; > + Octstr *type, *filename; > + MsgStore *ms; > + > + gwlib_init(); > + > + cf_index = get_and_set_debugs(argc, argv, check_args); > + > + if (argv[cf_index] == NULL) > + panic(0, "Usage: %s <store-file>", argv[0]); > + > + type = octstr_create("file"); > + filename = octstr_create(argv[cf_index]); > + > + /* init store subsystem */ > + ms = msg_store_init(STORE_FILE, msg_pack, msg_unpack_wrapper, filename, (long) -1); > + > + /* pass every entry in the store to callback print_msg() */ > + msg_store_load(ms, print_msg); > + > + info(0, "Store file contains %d msg entries", counter); > + info(0, "Store file funciton returns %ld", msg_store_messages(ms)); > + info(0, "Shutting down."); > + > + msg_store_shutdown(ms); > + > + gwlib_shutdown(); > + > + return 1; > +} > +