test_smsc.c

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

/* ==================================================================== 
 * The Kannel Software License, Version 1.0 
 * 
 * Copyright (c) 2001-2004 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 [e-mail]. 
 * 
 * 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.  
 */ 

#include <unistd.h>
#include "gwlib/gwlib.h"
#include "gw/smsc/smpp_pdu.h"
#include <string.h>

/***********************************************************************
 * Configurable stuff.
 */


/*
 * The port at which our HTTP server emulator listens.
 */
static long http_port = 8080;


/*
 * The HTTP admin port and password for Kannel, needed to do shutdown.
 */
static long admin_port = 13000;
static char *admin_password = "bar";


/*
 * The port at which the SMPP SMS center emulator listens.
 */
static long smpp_port = 2345;


/*
 * Number of messages to use in the "Send N messages as fast as possible"
 * benchmark.
 */
static long num_messages = 1;


/***********************************************************************
 * Events and event queues.
 */

typedef List EventQueue;


typedef struct Event {
    enum event_type {
	got_smsc,
	deliver,
	deliver_ack,
	http_request,
	http_response,
	submit,
	got_enquire_link
    } type;
    long id;
    long time;
    
    Connection *conn;	    /* SMPP: Connection for response PDU */
    long sequence_number;   /* SMPP: Sequence number of resp PDU */

    /* HTTP related stuff */
    HTTPClient *client;
    Octstr *body;
} Event;


static Counter *event_id_counter = NULL;


static const char *eq_type(Event *e)
{
#define TYPE(name) case name: return #name;
    switch (e->type) {
	TYPE(got_smsc)
	TYPE(deliver)
	TYPE(deliver_ack)
	TYPE(http_request)
	TYPE(http_response)
	TYPE(submit)
	TYPE(got_enquire_link)
    }
#undef TYPE
    return "unknown";
}


static Event *eq_create_event(enum event_type type)
{
    Event *e;
    
    e = gw_malloc(sizeof(*e));
    e->type = type;
    e->time = date_universal_now();
    e->id = counter_increase(event_id_counter);
    e->conn = NULL;
    e->sequence_number = -1;
    e->client = NULL;
    e->body = NULL;
    return e;
}


static Event *eq_create_submit(Connection *conn, long sequence_number, 
    	    	    	       Octstr *body)
{
    Event *e;
    
    gw_assert(conn != NULL);
    gw_assert(sequence_number >= 0);

    e = eq_create_event(submit);
    e->conn = conn;
    e->sequence_number = sequence_number;
    e->body = octstr_duplicate(body);
    return e;
}


static Event *eq_create_http_request(HTTPClient *client, Octstr *body)
{
    Event *e;
    
    gw_assert(client != NULL);
    gw_assert(body != NULL);

    e = eq_create_event(http_request);
    e->client = client;
    e->body = octstr_duplicate(body);
    return e;
}


static void eq_destroy_event(Event *e)
{
    octstr_destroy(e->body);
    gw_free(e);
}


static EventQueue *eq_create(void)
{
    return list_create();
}


static void eq_add_producer(EventQueue *eq)
{
    list_add_producer(eq);
}


static void eq_remove_producer(EventQueue *eq)
{
    list_remove_producer(eq);
}


static void eq_destroy(EventQueue *eq)
{
    list_destroy(eq, NULL);
}


static void eq_append(EventQueue *eq, Event *e)
{
    list_produce(eq, e);
}


static Event *eq_extract(EventQueue *eq)
{
    return list_consume(eq);
}


static void eq_log(Event *e)
{
    info(0, "Event %ld, type %s, time %ld", e->id, eq_type(e), e->time);
}


static void eq_init(void)
{
    event_id_counter = counter_create();
}


static void eq_shutdown(void)
{
    counter_destroy(event_id_counter);
}


static long eq_round_trip_time(Event *e)
{
    long now, then;
    
    now = date_universal_now();
    if (octstr_parse_long(&then, e->body, 0, 10) == -1)
    	return 0;
    return now - then;
}


/***********************************************************************
 * SMS center emulator, declarations.
 */


struct smsc_emu_arg {
    Semaphore *sema;
    EventQueue *eq;
};


static EventQueue *undelivered_messages = NULL;


/***********************************************************************
 * SMS center emulator, SMPP internals.
 */


enum { MAX_THREADS = 2 };
enum { SMPP_MAX_QUEUE = 10 };


struct smpp_emu_arg {
    EventQueue *eq;
    Connection *conn;
    long id;
    Semaphore *ok_to_send;
    long writer_id;
    int quit;
};


static Counter *smpp_emu_counter = NULL;


static void smpp_emu_writer(void *arg)
{
    Event *e;
    SMPP_PDU *pdu;
    Octstr *os;
    struct smpp_emu_arg *p;

    p = arg;
    for (;;) {
	semaphore_down(p->ok_to_send);
	e = eq_extract(undelivered_messages);
	if (e == NULL)
	    break;
    	e->time = date_universal_now();
    	eq_log(e);
	pdu = smpp_pdu_create(deliver_sm,
			      counter_increase(smpp_emu_counter));
    	pdu->u.deliver_sm.source_addr = octstr_create("123");
    	pdu->u.deliver_sm.destination_addr = octstr_create("456");
	pdu->u.deliver_sm.short_message = octstr_format("%ld", e->time);
	os = smpp_pdu_pack(pdu);
	conn_write(p->conn, os);
	octstr_destroy(os);
	smpp_pdu_destroy(pdu);
	eq_destroy_event(e);
    }
}


static void smpp_emu_handle_pdu(struct smpp_emu_arg *p, SMPP_PDU *pdu)
{
    SMPP_PDU *resp;
    Octstr *os;
    
    resp = NULL;
    switch (pdu->type) {
    	case bind_transmitter:
	    resp = smpp_pdu_create(bind_transmitter_resp,
				   pdu->u.bind_transmitter.sequence_number);
	    break;

    	case bind_receiver:
	    resp = smpp_pdu_create(bind_receiver_resp,
				   pdu->u.bind_receiver.sequence_number);
    	    eq_append(p->eq, eq_create_event(got_smsc));
	    gw_assert(p->writer_id == -1);
	    p->writer_id = gwthread_create(smpp_emu_writer, p);
	    if (p->writer_id == -1)
	    	panic(0, "Couldn't create SMPP helper thread.");
    	    break;

    	case submit_sm:
	    eq_append(p->eq, 
	    	eq_create_submit(p->conn, pdu->u.submit_sm.sequence_number,
		    	    	 pdu->u.submit_sm.short_message));
    	    break;

    	case deliver_sm_resp:
	    eq_append(p->eq, eq_create_event(deliver_ack));
	    semaphore_up(p->ok_to_send);
	    break;

    	case enquire_link:
	    eq_append(p->eq, eq_create_event(got_enquire_link));
	    resp = smpp_pdu_create(enquire_link_resp,
	    	    	    	   pdu->u.enquire_link.sequence_number);
	    break;

    	case unbind:
	    resp = smpp_pdu_create(unbind_resp, 
	    	    	    	   pdu->u.unbind.sequence_number);
	    break;

    	default:
	    error(0, "SMPP: Unhandled PDU type %s", pdu->type_name);
	    break;
    }
		
    if (resp != NULL) {
	os = smpp_pdu_pack(resp);
	conn_write(p->conn, os);
	octstr_destroy(os);
	smpp_pdu_destroy(resp);
    }
}


static void smpp_emu_reader(void *arg)
{
    Octstr *os;
    long len;
    SMPP_PDU *pdu;
    struct smpp_emu_arg *p;

    p = arg;
    
    len = 0;
    while (!p->quit && conn_wait(p->conn, -1.0) != -1) {
    	for (;;) {
	    if (len == 0) {
		len = smpp_pdu_read_len(p->conn);
		if (len == -1) {
		    error(0, "Client sent garbage, closing connection.");
		    goto error;
		} else if (len == 0) {
		    if (conn_eof(p->conn) || conn_error(p->conn))
		    	goto error;
		    break;
		}
	    }
    
    	    gw_assert(len > 0);
	    os = smpp_pdu_read_data(p->conn, len);
	    if (os != NULL) {
    	    	len = 0;
		pdu = smpp_pdu_unpack(os);
		if (pdu == NULL) {
		    error(0, "PDU unpacking failed!");
		    octstr_dump(os, 0);
		} else {
		    smpp_emu_handle_pdu(p, pdu);
		    smpp_pdu_destroy(pdu);
		}
		octstr_destroy(os);
	    } else if (conn_eof(p->conn) || conn_error(p->conn))
	    	goto error;
	    else
		break;
	}
    }

error:
    if (p->writer_id != -1)
	gwthread_join(p->writer_id);
}


static void smpp_emu(void *arg)
{
    EventQueue *eq;
    struct smsc_emu_arg *p;
    int fd;
    int new_fd;
    Octstr *client_addr;
    long i;
    long num_threads;
    struct smpp_emu_arg *thread[MAX_THREADS];
    
    p = arg;
    eq = p->eq;
    eq_add_producer(eq);
    semaphore_up(p->sema);
    
    /*
     * Wait for SMPP clients.
     */
    fd = make_server_socket(smpp_port, NULL);
    if (fd == -1)
    	panic(0, "Couldn't create SMPP listen port.");
    
    num_threads = 0;
    for (;;) {
    	new_fd = gw_accept(fd, &client_addr);
	if (new_fd == -1)
	    break;
    	octstr_destroy(client_addr);
    	if (num_threads == MAX_THREADS) {
	    warning(0, "Too many SMPP client connections.");
	    (void) close(new_fd);
	} else {
	    thread[num_threads] = gw_malloc(sizeof(*thread[0]));
    	    thread[num_threads]->conn = conn_wrap_fd(new_fd, 0);
	    thread[num_threads]->eq = eq;
	    thread[num_threads]->quit = 0;
	    thread[num_threads]->writer_id = -1;
	    thread[num_threads]->ok_to_send = 
	    	semaphore_create(SMPP_MAX_QUEUE);
	    thread[num_threads]->id =