smsc_wrapper.c File Reference

#include "gwlib/gwlib.h"
#include "smscconn.h"
#include "smscconn_p.h"
#include "bb_smscconn_cb.h"
#include "smsc.h"
#include "smsc_p.h"

Go to the source code of this file.

Data Structures
struct	smsc_wrapper

Typedefs
typedef struct smsc_wrapper	SmscWrapper

Functions
static void	smscwrapper_destroy (SmscWrapper *wrap)
static int	reconnect (SMSCConn *conn)
static Msg *	sms_receive (SMSCConn *conn)
static void	wrapper_receiver (void *arg)
static int	sms_send (SMSCConn *conn, Msg *msg)
static void	wrapper_sender (void *arg)
static int	wrapper_add_msg (SMSCConn *conn, Msg *sms)
static int	wrapper_shutdown (SMSCConn *conn, int finish_sending)
static void	wrapper_stop (SMSCConn *conn)
static void	wrapper_start (SMSCConn *conn)
static long	wrapper_queued (SMSCConn *conn)
int	smsc_wrapper_create (SMSCConn *conn, CfgGroup *cfg)

Typedef Documentation

SmscWrapper

typedef struct smsc_wrapper SmscWrapper

Function Documentation

reconnect()

static int reconnect ( SMSCConn *  conn )

static

Definition at line 97 of file smsc_wrapper.c.

References bb_smscconn_connected(), bb_smscconn_send_failed(), smscconn::connect_time, smscconn::data, debug(), error(), smscconn::flow_mutex, gwlist_extract_first(), gwthread_sleep(), info(), msg, mutex_lock, mutex_unlock, smscconn::name, octstr_get_cstr, smsc_wrapper::outgoing_queue, smsc_wrapper::reconnect_mutex, smsc_wrapper::smsc, smsc_reopen(), SMSCCONN_ACTIVE, SMSCCONN_ALIVE, SMSCCONN_DISCONNECTED, SMSCCONN_FAILED_TEMPORARILY, SMSCCONN_RECONNECTING, smscconn::status, and smscconn::why_killed.

Referenced by at2_device_thread(), sms_receive(), and sms_send().

{
    SmscWrapper *wrap = conn->data;
    Msg *msg;
    int ret;
    int wait = 1;

    /* disable double-reconnect
     * NOTE: it is still possible that we do double-connect if
     *   first thread gets through this if-statement and then
     *   execution switches to another thread.. this can be avoided
     *   via double-mutex system, but I do not feel it is worth it,
     *   maybe later --rpr
     */
    if (conn->status == SMSCCONN_RECONNECTING) {
    mutex_lock(wrap->reconnect_mutex);  /* wait here */
    mutex_unlock(wrap->reconnect_mutex);
    return 0;
    }
    mutex_lock(wrap->reconnect_mutex);

    debug("bb.sms", 0, "smsc_wrapper <%s>: reconnect started",
      octstr_get_cstr(conn->name));

    while((msg = gwlist_extract_first(wrap->outgoing_queue))!=NULL) {
    bb_smscconn_send_failed(conn, msg, SMSCCONN_FAILED_TEMPORARILY, NULL);
    }
    conn->status = SMSCCONN_RECONNECTING;
    

    while(conn->why_killed == SMSCCONN_ALIVE) {
    ret = smsc_reopen(wrap->smsc);
    if (ret == 0) {
        info(0, "Re-open of %s succeeded.", octstr_get_cstr(conn->name));
        mutex_lock(conn->flow_mutex);
        conn->status = SMSCCONN_ACTIVE;
        conn->connect_time = time(NULL);
        mutex_unlock(conn->flow_mutex);
        bb_smscconn_connected(conn);
        break;
    }
    else if (ret == -2) {
        error(0, "Re-open of %s failed permanently",
          octstr_get_cstr(conn->name));
        mutex_lock(conn->flow_mutex);
        conn->status = SMSCCONN_DISCONNECTED;
        mutex_unlock(wrap->reconnect_mutex);
        mutex_unlock(conn->flow_mutex);
        return -1;  /* permanent failure */
    }
    else {
        error(0, "Re-open to <%s> failed, retrying after %d minutes...",
          octstr_get_cstr(conn->name), wait);
        gwthread_sleep(wait*60.0);

        wait = wait > 10 ? 10 : wait * 2 + 1;
    }
    }
    mutex_unlock(wrap->reconnect_mutex);
    return 0;
}

sms_receive()

static Msg* sms_receive ( SMSCConn *  conn )

static

Definition at line 160 of file smsc_wrapper.c.

References smscconn::data, smscconn::id, msg_destroy(), smscconn::name, octstr_duplicate, octstr_get_cstr, reconnect(), smsc_wrapper::smsc, smscconn_shutdown(), smscenter_pending_smsmessage(), smscenter_receive_msg(), and warning().

Referenced by wrapper_receiver().

{
    SmscWrapper *wrap = conn->data;
    int ret;
    Msg *newmsg = NULL;

    if (smscenter_pending_smsmessage(wrap->smsc) == 1) {

        ret = smscenter_receive_msg(wrap->smsc, &newmsg);
        if (ret == 1) {

            /* if any smsc_id available, use it */
            newmsg->sms.smsc_id = octstr_duplicate(conn->id);

        return newmsg;
        } else if (ret == 0) { /* "NEVER" happens */
            warning(0, "SMSC %s: Pending message returned '1', "
                    "but nothing to receive!", octstr_get_cstr(conn->name));
            msg_destroy(newmsg);
            return NULL;
        } else {
            msg_destroy(newmsg);
        if (reconnect(conn) == -1)
        smscconn_shutdown(conn, 0);
        return NULL;
        }
    }
    return NULL;
}

sms_send()

static int sms_send ( SMSCConn *  conn, Msg *  msg  )

static

Definition at line 235 of file smsc_wrapper.c.

References bb_smscconn_send_failed(), bb_smscconn_sent(), smscconn::data, debug(), msg, smscconn::name, octstr_create, octstr_get_cstr, reconnect(), smsc_wrapper::smsc, SMSCCONN_FAILED_REJECTED, smscconn_shutdown(), and smscenter_submit_msg().

Referenced by wrapper_sender().

{
    SmscWrapper *wrap = conn->data;
    int ret;

    debug("bb.sms", 0, "smscconn_sender (%s): sending message",
      octstr_get_cstr(conn->name));

    ret = smscenter_submit_msg(wrap->smsc, msg);
    if (ret == -1) {
    bb_smscconn_send_failed(conn, msg,
                SMSCCONN_FAILED_REJECTED, octstr_create("REJECTED"));

    if (reconnect(conn) == -1)
        smscconn_shutdown(conn, 0);
        return -1;
    } else {
    bb_smscconn_sent(conn, msg, NULL);
        return 0;
    }
}

smsc_wrapper_create()

int smsc_wrapper_create	(	SMSCConn *	conn,
		CfgGroup *	cfg
	)

Definition at line 397 of file smsc_wrapper.c.

References cfg, smscconn::connect_time, smscconn::data, error(), gwlist_add_producer(), gwlist_create, gwthread_create, smscconn::is_stopped, mutex_create, smscconn::name, octstr_create, smsc_wrapper::outgoing_queue, smscconn::queued, smsc_wrapper::receiver_thread, smsc_wrapper::reconnect_mutex, smscconn::send_msg, smsc_wrapper::sender_thread, smscconn::shutdown, smsc_wrapper::smsc, smsc_name(), smsc_open(), SMSCCONN_ACTIVE, SMSCCONN_DEAD, SMSCCONN_KILLED_CANNOT_CONNECT, smscwrapper_destroy(), smscconn::start_conn, smscconn::status, smscconn::stop_conn, smsc_wrapper::stopped, smscconn::why_killed, wrapper_add_msg(), wrapper_queued(), wrapper_receiver(), wrapper_sender(), wrapper_shutdown(), wrapper_start(), and wrapper_stop().

Referenced by smscconn_create().

{
    /* 1. Call smsc_open()
     * 2. create sender/receiver threads
     * 3. fill up the conn
     *
     * XXX open() SHOULD be done in distinct thread, not here!
     */

    SmscWrapper *wrap;

    wrap = gw_malloc(sizeof(SmscWrapper));
    wrap->smsc = NULL;
    conn->data = wrap;
    conn->send_msg = wrapper_add_msg;
    
    
    wrap->outgoing_queue = gwlist_create();
    wrap->stopped = gwlist_create();
    wrap->reconnect_mutex = mutex_create();
    gwlist_add_producer(wrap->outgoing_queue);
    
    if ((wrap->smsc = smsc_open(cfg)) == NULL)
    goto error;

    conn->name = octstr_create(smsc_name(wrap->smsc));
    conn->status = SMSCCONN_ACTIVE;
    conn->connect_time = time(NULL);

    if (conn->is_stopped)
    gwlist_add_producer(wrap->stopped);
    

    /* XXX here we could fail things... specially if the second one
     *     fails.. so fix this ASAP
     *
     * moreover, open should be in sender/receiver, so that we can continue
     * while trying to open... maybe move this, or just wait for new
     * implementations of various SMSC protocols
     */
    
    if ((wrap->receiver_thread = gwthread_create(wrapper_receiver, conn))==-1)
    goto error;

    if ((wrap->sender_thread = gwthread_create(wrapper_sender, conn))==-1)
    goto error;

    conn->shutdown = wrapper_shutdown;
    conn->queued = wrapper_queued;
    conn->stop_conn = wrapper_stop;
    conn->start_conn = wrapper_start;
    
    return 0;

error:
    error(0, "Failed to create Smsc wrapper");
    conn->data = NULL;
    smscwrapper_destroy(wrap);
    conn->why_killed = SMSCCONN_KILLED_CANNOT_CONNECT;
    conn->status = SMSCCONN_DEAD;
    return -1;
}

smscwrapper_destroy()

static void smscwrapper_destroy ( SmscWrapper *  wrap )

static

Definition at line 84 of file smsc_wrapper.c.

References gwlist_destroy(), mutex_destroy(), smsc_wrapper::outgoing_queue, smsc_wrapper::reconnect_mutex, smsc_wrapper::smsc, smsc_close(), and smsc_wrapper::stopped.

Referenced by smsc_wrapper_create(), and wrapper_sender().

{
    if (wrap == NULL)
    return;
    gwlist_destroy(wrap->outgoing_queue, NULL);
    gwlist_destroy(wrap->stopped, NULL);
    mutex_destroy(wrap->reconnect_mutex);
    if (wrap->smsc != NULL)
    smsc_close(wrap->smsc);
    gw_free(wrap);
}

wrapper_add_msg()

static int wrapper_add_msg ( SMSCConn *  conn, Msg *  sms  )

static

Definition at line 337 of file smsc_wrapper.c.

References smscconn::data, gwlist_produce(), msg_duplicate(), and smsc_wrapper::outgoing_queue.

Referenced by smsc_wrapper_create().

{
    SmscWrapper *wrap = conn->data;
    Msg *copy;

    copy = msg_duplicate(sms);
    gwlist_produce(wrap->outgoing_queue, copy);

    return 0;
}

wrapper_queued()

static long wrapper_queued ( SMSCConn *  conn )

static

Definition at line 386 of file smsc_wrapper.c.

References smscconn::data, gwlist_len(), smscconn::load, and smsc_wrapper::outgoing_queue.

Referenced by smsc_wrapper_create().

{
    SmscWrapper *wrap = conn->data;
    long ret = gwlist_len(wrap->outgoing_queue);

    /* use internal queue as load, maybe something else later */
    
    conn->load = ret;
    return ret;
}

wrapper_receiver()

static void wrapper_receiver ( void *  arg )

static

Definition at line 191 of file smsc_wrapper.c.

References bb_smscconn_receive(), smscconn::data, debug(), gwlist_consume(), gwthread_sleep(), smscconn::log_idx, log_thread_to(), msg, smscconn::name, octstr_get_cstr, sms_receive(), SMSCCONN_ALIVE, SMSCCONN_KILLED_SHUTDOWN, smsc_wrapper::stopped, and smscconn::why_killed.

Referenced by smsc_wrapper_create().

{
    Msg     *msg;
    SMSCConn    *conn = arg;
    SmscWrapper *wrap = conn->data;
    /* SmscWrapper *wrap = conn->data; ** non-used */
    double  sleep = 0.0001;
    
    /* Make sure we log into our own log-file if defined */
    log_thread_to(conn->log_idx);
    
    /* remove messages from SMSC until we are killed */
    while(conn->why_killed == SMSCCONN_ALIVE) {

        gwlist_consume(wrap->stopped); /* block here if suspended/isolated */

    msg = sms_receive(conn);
    if (msg) {
            debug("bb.sms", 0, "smscconn (%s): new message received",
          octstr_get_cstr(conn->name));
            sleep = 0.0001;
        bb_smscconn_receive(conn, msg);
        }
        else {
        /* note that this implementations means that we sleep even
         * when we fail connection.. but time is very short, anyway
         */
            gwthread_sleep(sleep);
            /* gradually sleep longer and longer times until something starts to
             * happen - this of course reduces response time, but that's better than
             * extensive CPU usage when it is not used
             */
            sleep *= 2;
            if (sleep >= 2.0)
                sleep = 1.999999;
        }
    }
    conn->why_killed = SMSCCONN_KILLED_SHUTDOWN;

    /* this thread is joined at sender */
}

wrapper_sender()

static void wrapper_sender ( void *  arg )

static

Definition at line 258 of file smsc_wrapper.c.

References bb_smscconn_killed(), bb_smscconn_send_failed(), smscconn::data, debug(), smscconn::flow_mutex, gwlist_consume(), gwlist_destroy(), gwlist_extract_first(), gwlist_get(), gwlist_len(), gwlist_remove_producer(), gwthread_join(), gwthread_wakeup(), smscconn::is_stopped, smscconn::log_idx, log_thread_to(), msg, msg_destroy(), msg_duplicate(), mutex_lock, mutex_unlock, smscconn::name, octstr_destroy(), octstr_get_cstr, octstr_search_char(), octstr_split_words(), smsc_wrapper::outgoing_queue, smsc_wrapper::receiver_thread, sms_send(), SMSCCONN_DEAD, SMSCCONN_FAILED_SHUTDOWN, SMSCCONN_KILLED_SHUTDOWN, smscwrapper_destroy(), smscconn::status, smsc_wrapper::stopped, and smscconn::why_killed.

Referenced by smsc_wrapper_create().

{
    Msg     *msg;
    SMSCConn    *conn = arg;
    SmscWrapper *wrap = conn->data;

    /* Make sure we log into our own log-file if defined */
    log_thread_to(conn->log_idx);

    /* send messages to SMSC until our outgoing_list is empty and
     * no producer anymore (we are set to shutdown) */
    while(conn->status != SMSCCONN_DEAD) {

    if ((msg = gwlist_consume(wrap->outgoing_queue)) == NULL)
            break;

        if (octstr_search_char(msg->sms.receiver, ' ', 0) != -1) {
            /*
             * multi-send: this should be implemented in corresponding
             *  SMSC protocol, but while we are waiting for that...
             */
            int i;
        Msg *newmsg;
            /* split from spaces: in future, split with something more sensible,
             * this is dangerous... (as space is url-encoded as '+')
             */
            List *nlist = octstr_split_words(msg->sms.receiver);

        debug("bb.sms", 0, "Handling multi-receiver message");

            for(i=0; i < gwlist_len(nlist); i++) {

        newmsg = msg_duplicate(msg);
                octstr_destroy(newmsg->sms.receiver);

                newmsg->sms.receiver = gwlist_get(nlist, i);
                sms_send(conn, newmsg);
            }
            gwlist_destroy(nlist, NULL);
            msg_destroy(msg);
        }
        else
        sms_send(conn,msg);

    }
    /* cleanup, we are now dying */

    debug("bb.sms", 0, "SMSCConn %s sender died, waiting for receiver",
      octstr_get_cstr(conn->name));
    
    conn->why_killed = SMSCCONN_KILLED_SHUTDOWN;

    if (conn->is_stopped) {
    gwlist_remove_producer(wrap->stopped);
    conn->is_stopped = 0;
    }

    gwthread_wakeup(wrap->receiver_thread);
    gwthread_join(wrap->receiver_thread);

    /* call 'failed' to all messages still in queue */
    
    mutex_lock(conn->flow_mutex);

    conn->status = SMSCCONN_DEAD;

    while((msg = gwlist_extract_first(wrap->outgoing_queue))!=NULL) {
    bb_smscconn_send_failed(conn, msg, SMSCCONN_FAILED_SHUTDOWN, NULL);
    }
    smscwrapper_destroy(wrap);
    conn->data = NULL;
    
    mutex_unlock(conn->flow_mutex);

    bb_smscconn_killed();
}

wrapper_shutdown()

static int wrapper_shutdown ( SMSCConn *  conn, int  finish_sending  )

static

Definition at line 349 of file smsc_wrapper.c.

References bb_smscconn_send_failed(), smscconn::data, debug(), gwlist_extract_first(), gwlist_remove_producer(), gwthread_wakeup(), msg, smscconn::name, octstr_get_cstr, smsc_wrapper::outgoing_queue, smsc_wrapper::receiver_thread, smsc_wrapper::sender_thread, and SMSCCONN_FAILED_SHUTDOWN.

Referenced by smsc_wrapper_create().

{
    SmscWrapper *wrap = conn->data;

    debug("bb.sms", 0, "Shutting down SMSCConn %s, %s",
      octstr_get_cstr(conn->name), finish_sending ? "slow" : "instant");
    
    if (finish_sending == 0) {
    Msg *msg; 
    while((msg = gwlist_extract_first(wrap->outgoing_queue))!=NULL) {
        bb_smscconn_send_failed(conn, msg, SMSCCONN_FAILED_SHUTDOWN, NULL);
    }
    }
    gwlist_remove_producer(wrap->outgoing_queue);
    gwthread_wakeup(wrap->sender_thread);
    gwthread_wakeup(wrap->receiver_thread);
    return 0;
}

wrapper_start()

static void wrapper_start ( SMSCConn *  conn )

static

Definition at line 377 of file smsc_wrapper.c.

References smscconn::data, debug(), gwlist_remove_producer(), and smsc_wrapper::stopped.

Referenced by smsc_wrapper_create().

{
    SmscWrapper *wrap = conn->data;

    debug("smscconn", 0, "Starting wrapper");
    gwlist_remove_producer(wrap->stopped);
}

wrapper_stop()

static void wrapper_stop ( SMSCConn *  conn )

static

Definition at line 368 of file smsc_wrapper.c.

References smscconn::data, debug(), gwlist_add_producer(), and smsc_wrapper::stopped.

Referenced by smsc_wrapper_create().

{
    SmscWrapper *wrap = conn->data;

    debug("smscconn", 0, "Stopping wrapper");
    gwlist_add_producer(wrap->stopped);

}