smsc_oisd.c

The Kannel Open Source WAP and SMS gateway works as both an SMS gateway, for implementing keyword b

{
    PrivData *pdata = conn->data;
    long ret;
    struct packet *packet;

    gw_assert(pdata != NULL);

    if (list_len(pdata->received) > 0) {
        *msg = list_consume(pdata->received);
        return 1;
    }

    if (pdata->socket < 0) {
        /* XXX We have to assume that smsc_send_message is
         * currently trying to reopen, so we have to make
         * this thread wait.  It should be done in a nicer
         * way. */
        return 0;
    }

    ret = read_available(pdata->socket, 0);
    if (ret == 0) {
        if (pdata->keepalive > 0 && pdata->next_ping < time(NULL)) {
            if (oisd_send_delivery_request(conn) < 0)
                return -1;
        }
        return 0;
    }

    if (ret < 0) {
        warning(errno, "OISD[%s]: oisd_receive_msg: read_available failed",
                octstr_get_cstr(conn->id));
        return -1;
    }

    /* We have some data waiting... see if it is an sms delivery. */
    ret = octstr_append_from_socket(pdata->inbuffer, pdata->socket);

    if (ret == 0) {
        warning(0, "OISD[%s]: oisd_receive_msg: service center closed connection.",
                octstr_get_cstr(conn->id));
        return -1;
    }
    if (ret < 0) {
        warning(0, "OISD[%s]: oisd_receive_msg: read failed",
                octstr_get_cstr(conn->id));
        return -1;
    }


    for (;;) {
        packet = packet_extract(pdata->inbuffer, conn);
        if (!packet)
            break;

        packet_check_can_receive(packet, conn);
        debug("bb.sms.oisd", 0, "OISD[%s]: received",
              octstr_get_cstr(pdata->conn->id));
        octstr_dump(packet->data, 0);

        if (packet->operation < RESPONSE)
            oisd_handle_request(packet, conn);
        else {
            error(0, "OISD[%s]: oisd_receive_msg: unexpected response packet",
                  octstr_get_cstr(conn->id));
            octstr_dump(packet->data, 0);
        }

        packet_destroy(packet);
    }

    if (list_len(pdata->received) > 0) {
        *msg = list_consume(pdata->received);
        return 1;
    }
    return 0;
}

static Msg *oisd_accept_delivery_report_message(struct packet *request,
                                                SMSCConn *conn)
{
    Msg *msg = NULL;
    Octstr *destination = NULL;
    Octstr *timestamp = NULL;
    int st_code;
    int code;
    int dest_len;

    /* MSISDN length */
    dest_len = octstr_get_char(request->data, 10);
    /* MSISDN */
    destination = octstr_copy(request->data, 10+1, dest_len);
    /* Accept time */
    timestamp = octstr_copy(request->data, 10+1+dest_len+1+4+4, 14);
    /* SM status */
    st_code = octstr_get_char(request->data, 10+1+dest_len+1+4+4+14);

    switch (st_code) {
    case 1:
    case 2:
        code = DLR_FAIL;
        break;
    case 3:   /* success */
        code = DLR_SUCCESS;
        break;
    case 4:
    case 5:
    case 6:
    default:
        code = 0;
    }

    if (code)
        msg = dlr_find(conn->name, timestamp, destination, code);

    octstr_destroy(destination);
    octstr_destroy(timestamp);

    return msg;
}

static Msg *sms_receive(SMSCConn *conn)
{
    PrivData *pdata = conn->data;
    int ret;
    Msg *newmsg = NULL;

    ret = oisd_receive_msg(conn, &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) { /* no message, just retry... */
        return NULL;
    }
    /* error. reconnect. */
    msg_destroy(newmsg);
    mutex_lock(conn->flow_mutex);
    oisd_close_socket(pdata);
    conn->status = SMSCCONN_DISCONNECTED;
    mutex_unlock(conn->flow_mutex);
    return NULL;
}

static void io_thread (void *arg)
{
    Msg       *msg;
    SMSCConn  *conn = arg;
    PrivData *pdata = conn->data;
    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 (!pdata->quitting) {

        list_consume(pdata->stopped); /* block here if suspended/isolated */

        /* check that connection is active */
        if (conn->status != SMSCCONN_ACTIVE) {
            if (oisd_login(conn) != 0) {
                error(0, "OISD[%s]: Couldn't connect to SMSC (retrying in %ld seconds).",
                      octstr_get_cstr(conn->id),
                      conn->reconnect_delay);
                gwthread_sleep(conn->reconnect_delay);
                mutex_lock(conn->flow_mutex);
                conn->status = SMSCCONN_RECONNECTING;
                mutex_unlock(conn->flow_mutex);
                continue;
            }
            mutex_lock(conn->flow_mutex);
            conn->status = SMSCCONN_ACTIVE;
            conn->connect_time = time(NULL);
            bb_smscconn_connected(conn);
            mutex_unlock(conn->flow_mutex);
        }

        /* receive messages */
        do {
            msg = sms_receive(conn);
            if (msg) {
                sleep = 0;
                debug("bb.sms.oisd", 0, "OISD[%s]: new message received",
                      octstr_get_cstr(conn->id));
                bb_smscconn_receive(conn, msg);
            }
        } while (msg);

        /* send messages */
        do {
            msg = list_extract_first(pdata->outgoing_queue);
            if (msg) {
                sleep = 0;
                if (oisd_submit_msg(conn, msg) != 0) break;
            }
        } while (msg);

        if (sleep > 0) {

            /* 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;
        }
        else {
            sleep = 0.0001;
        }
    }
}

static int oisd_add_msg_cb (SMSCConn *conn, Msg *sms)
{
    PrivData *pdata = conn->data;
    Msg *copy;

    copy = msg_duplicate(sms);
    list_produce(pdata->outgoing_queue, copy);
    gwthread_wakeup(pdata->io_thread);

    return 0;
}

static int oisd_shutdown_cb (SMSCConn *conn, int finish_sending)
{
    PrivData *pdata = conn->data;

    debug("bb.sms", 0, "Shutting down SMSCConn OISD %s (%s)",
          octstr_get_cstr(conn->id),
          finish_sending ? "slow" : "instant");

    /* Documentation claims this would have been done by smscconn.c,
       but isn't when this code is being written. */
    conn->why_killed = SMSCCONN_KILLED_SHUTDOWN;
    pdata->quitting = 1;     /* Separate from why_killed to avoid locking, as
                              * why_killed may be changed from outside? */

    if (finish_sending == 0) {
        Msg *msg;
        while ((msg = list_extract_first(pdata->outgoing_queue)) != NULL) {
            bb_smscconn_send_failed(conn, msg, SMSCCONN_FAILED_SHUTDOWN, NULL);
        }
    }

    if (conn->is_stopped) {
        list_remove_producer(pdata->stopped);
        conn->is_stopped = 0;
    }

    if (pdata->io_thread != -1) {
        gwthread_wakeup(pdata->io_thread);
        gwthread_join(pdata->io_thread);
    }

    oisd_close_socket(pdata);
    oisd_destroy(pdata);

    debug("bb.sms", 0, "SMSCConn OISD %s shut down.",
          octstr_get_cstr(conn->id));
    conn->status = SMSCCONN_DEAD;
    bb_smscconn_killed();
    return 0;
}

static void oisd_start_cb (SMSCConn *conn)
{
    PrivData *pdata = conn->data;

    list_remove_producer(pdata->stopped);
    /* in case there are messages in the buffer already */
    gwthread_wakeup(pdata->io_thread);
    debug("bb.sms", 0, "SMSCConn OISD %s, start called",
          octstr_get_cstr(conn->id));
}

static void oisd_stop_cb (SMSCConn *conn)
{
    PrivData *pdata = conn->data;
    list_add_producer(pdata->stopped);
    debug("bb.sms", 0, "SMSCConn OISD %s, stop called",
          octstr_get_cstr(conn->id));
}

static long oisd_queued_cb (SMSCConn *conn)
{
    PrivData *pdata = conn->data;
    conn->load = (pdata ? (conn->status != SMSCCONN_DEAD ?
                  list_len(pdata->outgoing_queue) : 0) : 0);
    return conn->load;
}

int smsc_oisd_create(SMSCConn *conn, CfgGroup *grp)
{
    PrivData *pdata;
    int ok;

    pdata = gw_malloc(sizeof(PrivData));
    conn->data = pdata;
    pdata->conn = conn;

    pdata->no_dlr = 0;
    pdata->quitting = 0;
    pdata->socket = -1;
    pdata->received = list_create();
    pdata->inbuffer = octstr_create("");
    pdata->send_seq = 1;
    pdata->outgoing_queue = list_create();
    pdata->stopped = list_create();
    list_add_producer(pdata->outgoing_queue);

    if (conn->is_stopped)
        list_add_producer(pdata->stopped);

    pdata->host = cfg_get(grp, octstr_imm("host"));
    if (cfg_get_integer(&(pdata->port), grp, octstr_imm("port")) == -1)
        pdata->port = 0;
    pdata->my_number = cfg_get(grp, octstr_imm("my-number"));
    if (cfg_get_integer(&(pdata->keepalive), grp, octstr_imm("keepalive")) == -1)
        pdata->keepalive = 0;
    if (cfg_get_integer(&(pdata->validityperiod), grp, octstr_imm("validityperiod")) == -1)
        pdata->validityperiod = 0;

    cfg_get_bool(&pdata->no_dlr, grp, octstr_imm("no-dlr"));

    /* Check that config is OK */
    ok = 1;
    if (pdata->host == NULL) {
        error(0, "OISD[%s]: Configuration file doesn't specify host",
              octstr_get_cstr(conn->id));
        ok = 0;
    }
    if (pdata->port == 0) {
        error(0, "OISD[%s]: Configuration file doesn't specify port",
              octstr_get_cstr(conn->id));
        ok = 0;
    }
    if (pdata->my_number == NULL && pdata->keepalive > 0) {
        error(0, "OISD[%s]: Configuration file doesn't specify my-number.",
              octstr_get_cstr(conn->id));
        ok = 0;
    }

    if (!ok) {
        oisd_destroy(pdata);
        return -1;
    }

    conn->name = octstr_format("OISD:%s:%d",
                     octstr_get_cstr(pdata->host),
                     pdata->port);


    if (pdata->keepalive > 0) {
        debug("bb.sms.oisd", 0, "OISD[%s]: Keepalive set to %ld seconds",
              octstr_get_cstr(conn->id),
              pdata->keepalive);
        pdata->next_ping = time(NULL) + pdata->keepalive;
    }

    if (pdata->validityperiod > 0) {
        debug("bb.sms.oisd", 0, "OISD[%s]: Validity-Period set to %ld",
              octstr_get_cstr(conn->id),
              pdata->validityperiod);
    }

    pdata->io_thread = gwthread_create(io_thread, conn);

    if (pdata->io_thread == -1) {

        error(0, "OISD[%s]: Couldn't start I/O thread.",
              octstr_get_cstr(conn->id));
        pdata->quitting = 1;
        gwthread_wakeup(pdata->io_thread);
        gwthread_join(pdata->io_thread);
        oisd_destroy(pdata);
        return -1;
    }

    conn->send_msg = oisd_add_msg_cb;
    conn->shutdown = oisd_shutdown_cb;
    conn->queued = oisd_queued_cb;
    conn->start_conn = oisd_start_cb;
    conn->stop_conn = oisd_stop_cb;

    return 0;
}