smsc_emi2.c

gnu的专业网关smpp协议支持源代码。

    while (emi2_can_send (conn) &&
	   (msg = list_extract_first(PRIVDATA(conn)->outgoing_queue)) != NULL) {
	int nexttrn = emi2_next_trn (conn);

	if (PRIVDATA(conn)->throughput)
	    gwthread_sleep(delay);

	/* convert the generic Kannel message into an EMI type message */
	emimsg = msg_to_emimsg(msg, nexttrn, PRIVDATA(conn));

	/* remember the message for retransmission or DLR */
	PRIVDATA(conn)->slots[nexttrn].sendmsg = msg;
	PRIVDATA(conn)->slots[nexttrn].sendtype = 51;
	PRIVDATA(conn)->slots[nexttrn].sendtime = time(NULL);

	/* send the message */
	if (emi2_emimsg_send(conn, server, emimsg) == -1) {
	    emimsg_destroy(emimsg);
	    return -1;
	}

	/* report the submission to the DLR code */
	if (msg->sms.dlr_mask & 0x18) {
	    Octstr *ts;
	    ts = octstr_create("");
	    octstr_append(ts, (conn->id ? conn->id : PRIVDATA(conn)->name));
	    octstr_append_char(ts, '-');
	    octstr_append_decimal(ts, nexttrn);

	    dlr_add(octstr_get_cstr((conn->id ? conn->id : PRIVDATA(conn)->name)), 
		    octstr_get_cstr(ts),
		    octstr_get_cstr(msg->sms.sender),
		    octstr_get_cstr(emimsg->fields[E50_ADC]),
		    octstr_get_cstr(msg->sms.service),
		    octstr_get_cstr(msg->sms.dlr_url),
		    msg->sms.dlr_mask);
	    
	    octstr_destroy(ts);
	    PRIVDATA(conn)->slots[nexttrn].dlr = 1;
	} else {
	    PRIVDATA(conn)->slots[nexttrn].dlr = 0;
	}

	/* we just sent a message */
	PRIVDATA(conn)->unacked++;

	emimsg_destroy(emimsg);

	/*
	 * remember that there is an open request for stop-wait flow control
	 * FIXME: couldn't this be done with the unacked field as well? After
	 * all stop-wait is just a window of size 1.
	 */
	PRIVDATA(conn)->can_write = 0;
    }

    return 0;
}

static int emi2_handle_smscreq (SMSCConn *conn, Connection *server)
{
    Octstr	  *str;
    struct emimsg *emimsg;
    PrivData *privdata = conn->data;
    
    /* Read acks/nacks/ops from the server */
    while ((str = conn_read_packet(server, 2, 3))) {
	debug("smsc.emi2", 0, "EMI2[%s]: Got packet from the main socket",
	      octstr_get_cstr(privdata->name));

	/* parse the msg */
	emimsg = get_fields(str, privdata->name);
	octstr_destroy(str);
	
	if (emimsg == NULL) {
	    continue; /* The parse functions logged errors */
	}
	
	if (emimsg->or == 'O') {
	    /* If the SMSC wants to send operations through this
	     * socket, we'll have to read them because there
	     * might be ACKs too. We just drop them while stopped,
	     * hopefully the SMSC will resend them later. */
	    if (!conn->is_stopped) {
		if (handle_operation(conn, server, emimsg) < 0)
		    return -1; /* Connection broke */
	    } else {
		info(0, "EMI2[%s]: Ignoring operation from main socket "
		     "because the connection is stopped.",
		     octstr_get_cstr(privdata->name));
	    }
	} else {   /* Already checked to be 'O' or 'R' */
	    if (!SLOTBUSY(conn,emimsg->trn) ||
		emimsg->ot != PRIVDATA(conn)->slots[emimsg->trn].sendtype) {
		error(0, "EMI2[%s]: Got ack for TRN %d, don't remember sending O?", 
		      octstr_get_cstr(privdata->name), emimsg->trn);
	    } else {
		PRIVDATA(conn)->can_write = 1;
		PRIVDATA(conn)->slots[emimsg->trn].sendtime = 0;
		PRIVDATA(conn)->unacked--;
		
		if (emimsg->ot == 51) {
		    if (PRIVDATA(conn)->slots[emimsg->trn].dlr) {
			Msg *dlrmsg;
			Octstr *ts;
			Msg *origmsg;

			origmsg = PRIVDATA(conn)->slots[emimsg->trn].sendmsg;

			ts = octstr_create("");
			octstr_append(ts, (conn->id ? conn->id : privdata->name));
			octstr_append_char(ts, '-');
			octstr_append_decimal(ts, emimsg->trn);

			dlrmsg = dlr_find(octstr_get_cstr((conn->id ? conn->id : privdata->name)), 
					  octstr_get_cstr(ts), /* timestamp */
					  octstr_get_cstr(origmsg->sms.receiver), /* destination */
					  (octstr_get_char(emimsg->fields[0], 0) == 'A' ? 
					   DLR_SMSC_SUCCESS : DLR_SMSC_FAIL));

			octstr_destroy(ts);
			if (dlrmsg != NULL) {
			    Octstr *moretext;

                /*
                 * Recode the msg structure with the given msgdata.
                 * Note: the DLR URL is delivered in msg->sms.dlr_url already.
                 */
                dlrmsg->sms.msgdata = octstr_duplicate(emimsg->fields[E50_AMSG]);
                octstr_hex_to_binary(dlrmsg->sms.msgdata);
                dlrmsg->sms.sms_type = report;

			    moretext = octstr_create("");
			    if (octstr_get_char(emimsg->fields[0], 0) == 'N') {
                    octstr_append(moretext, emimsg->fields[1]);
                    octstr_append_char(moretext, '-');
                    octstr_append(moretext, emimsg->fields[2]);
			    }
			    octstr_append_char(moretext, '/');
			    octstr_insert(dlrmsg->sms.msgdata, moretext, 0);
			    octstr_destroy(moretext);

			    bb_smscconn_receive(conn, dlrmsg);
			}
		    }

		    if (octstr_get_char(emimsg->fields[0], 0) == 'A') {
			/* we got an ack back. We might have to store the */
			/* timestamp for delivery notifications now */
			Octstr *ts, *adc;
			int	i;
			Msg *m;
			  
			ts = octstr_duplicate(emimsg->fields[2]);
			if (octstr_len(ts)) {
			    i = octstr_search_char(ts,':',0);
			    if (i>0) {
				octstr_delete(ts,0,i+1);
				adc = octstr_duplicate(emimsg->fields[2]);
				octstr_truncate(adc,i);
			        
				m = PRIVDATA(conn)->slots[emimsg->trn].sendmsg;
				if(m == NULL) {
				    info(0,"EMI2[%s]: uhhh m is NULL, very bad",
					 octstr_get_cstr(privdata->name));
				} else if (m->sms.dlr_mask & 0x7) {
				    dlr_add(octstr_get_cstr((conn->id ? conn->id : privdata->name)), 
					    octstr_get_cstr(ts),
					    octstr_get_cstr(m->sms.sender),
					    octstr_get_cstr(adc),
					    octstr_get_cstr(m->sms.service),
					    octstr_get_cstr(m->sms.dlr_url),
					    m->sms.dlr_mask);
				}
				octstr_destroy(ts);
				octstr_destroy(adc);
			    } else {
				octstr_destroy(ts);
			    }

			    
			}
			/*
			 * report the successful transmission to the generic bb code.
			 */
			bb_smscconn_sent(conn,
					 PRIVDATA(conn)->slots[emimsg->trn].sendmsg);
		    } else {
			/* XXX Process error code here
			long errorcode;
			octstr_parse_long(&errorcode, emimsg->fields[1], 0, 10);
			... switch(errorcode) ...
			} 
			
			else { */
			    bb_smscconn_send_failed(conn,
						PRIVDATA(conn)->slots[emimsg->trn].sendmsg,
						SMSCCONN_FAILED_REJECTED);
			/* } */
		    }
		} else if (emimsg->ot == 31) {
		    /* XXX Process error codes here
		    if (octstr_get_char(emimsg->fields[0], 0) == 'N') {
			long errorcode;
			octstr_parse_long(&errorcode, emimsg->fields[1], 0, 10);
			... switch errorcode ...
		    } 
		    
		    else { */
			;
		    /* } */
		} else {
		    panic(0, "EMI2[%s]: Bug, ACK handler missing for sent packet",
			  octstr_get_cstr(privdata->name));
		}
	    }
	}
	emimsg_destroy(emimsg);
    }

    if (conn_read_error(server)) {
	error(0, "EMI2[%s]: Error trying to read ACKs from SMSC",
	      octstr_get_cstr(privdata->name));
	return -1;
    }
    
    if (conn_eof(server)) {
	info(0, "EMI2[%s]: Main connection closed by SMSC",
	     octstr_get_cstr(privdata->name));
	return -1;
    }

    return 0;
}

static void emi2_idleprocessing(SMSCConn *conn)
{
    time_t current_time;
    int i;
    PrivData *privdata = conn->data;
    
    /*
     * Check whether there are messages the server hasn't acked in a
     * reasonable time
     */
    current_time = time(NULL);
    
    if (PRIVDATA(conn)->unacked && (current_time > (PRIVDATA(conn)->check_time + 30))) {
	PRIVDATA(conn)->check_time = current_time;
	for (i = 0; i < EMI2_MAX_TRN; i++) {
	    if (SLOTBUSY(conn,i)
		&& PRIVDATA(conn)->slots[i].sendtime < (current_time - PRIVDATA(conn)->waitack)) {
		PRIVDATA(conn)->slots[i].sendtime = 0;
		PRIVDATA(conn)->unacked--;
		if (PRIVDATA(conn)->slots[i].sendtype == 51) {
		    warning(0, "EMI2[%s]: received neither ACK nor NACK for message %d " 
			    "in %d seconds, resending message", octstr_get_cstr(privdata->name),
			    i, PRIVDATA(conn)->waitack);
		    list_produce(PRIVDATA(conn)->outgoing_queue,
				 PRIVDATA(conn)->slots[i].sendmsg);
		    if (PRIVDATA(conn)->flowcontrol) PRIVDATA(conn)->can_write=1;
		    /* Wake up this same thread to send again
		     * (simpler than avoiding sleep) */
		    gwthread_wakeup(PRIVDATA(conn)->sender_thread);
		} else if (PRIVDATA(conn)->slots[i].sendtype == 31) {
		    warning(0, "EMI2[%s]: Alert (operation 31) was not "
			    "ACKed within %d seconds", octstr_get_cstr(privdata->name),
			    PRIVDATA(conn)->waitack);
		    if (PRIVDATA(conn)->flowcontrol) PRIVDATA(conn)->can_write=1;
		} else {
		    panic(0, "EMI2[%s]: Bug, no timeout handler for sent packet",
			  octstr_get_cstr(privdata->name));
		}
	    }
	}
    }
}

static void emi2_idletimeout_handling (SMSCConn *conn, Connection **server)
{
    PrivData *privdata = conn->data;

    /*
     * close the connection if there was no activity.
     */
    if ((*server != NULL) && CONNECTIONIDLE(conn)) {
	info(0, "EMI2[%s]: closing idle connection.",
	     octstr_get_cstr(privdata->name));
	conn_destroy(*server);
	*server = NULL;
    }
}

/*
 * this function calculates the new timeouttime.
 */
static double emi2_get_timeouttime (SMSCConn *conn, Connection *server)
{
    double ka_timeouttime = PRIVDATA(conn)->keepalive ? PRIVDATA(conn)->keepalive + 1 : DBL_MAX;
    double idle_timeouttime = (PRIVDATA(conn)->idle_timeout && server) ? PRIVDATA(conn)->idle_timeout : DBL_MAX;
    double result = ka_timeouttime < idle_timeouttime ? ka_timeouttime : idle_timeouttime;

    if (result == DBL_MAX)
	result = 30;

    return result;
}

/*
 * the main event processing loop.
 */
static void emi2_send_loop(SMSCConn *conn, Connection **server)
{
    PrivData *privdata = conn->data;

    for (;;) {
	double timeouttime;
	EMI2Event event;
	
	if (emi2_needs_keepalive (conn)) {
	    if (*server == NULL) {
		return; /* reopen the connection */
	    }
	    
	    emi2_keepalive_handling (conn, *server);
	}
	
	timeouttime = emi2_get_timeouttime (conn, *server);
	
	event = emi2_wait (conn, *server, timeouttime);
	
	switch (event) {
	case EMI2_CONNERR:
	    return;
	    
	case EMI2_SENDREQ:
	    if (*server == NULL) {
		return; /* reopen the connection */
	    }
	    
	    if (emi2_do_send (conn, *server) < 0) {
		return; /* reopen the connection */
	    }
	    break;
	    
	case EMI2_SMSCREQ:
	    if (emi2_handle_smscreq (conn, *server) < 0) {
		return; /* reopen the connection */
	    }
	    break;
	    
	case EMI2_TIMEOUT:
	    break;
	}

        if ((*server !=NULL) && (emi2_handle_smscreq (conn, *server) < 0)) {
            return; /* reopen the connection */
        }
	
	emi2_idleprocessing (conn);
	emi2_idletimeout_handling (conn, server);

	if (PRIVDATA(conn)->shutdown && (PRIVDATA(conn)->unacked == 0)) {
	    /* shutdown and no open messages */
	    break;
	}

	if (*server != NULL) {
	    if (conn_read_error(*server)) {
		warning(0, "EMI2[%s]: Error reading from the main connection",
			octstr_get_cstr(privdata->name));
		break;
	    }
	
	    if (conn_eof(*server)) {
		info(0, "EMI2[%s]: Main connection closed by SMSC",
		     octstr_get_cstr(privdata->name));
		break;
	    }
	}
    }
}

static void emi2_sender(void *arg)
{
    SMSCConn *conn = arg;
    PrivData *privdata = conn->data;
    Msg *msg;
    Connection *server;

    while (!privdata->shutdown) {
	if ((server = open_send_connection(conn)) == NULL) {
	    privdata->shutdown = 1;
	    if (privdata->rport > 0)
		gwthread_wakeup(privdata->receiver_thread);
	    break;
	}
	emi2_send_loop(conn, &server);
	clear_sent(privdata);

	if (server != NULL) {
	    conn_destroy(server);
	}
    }

    while((msg = list_extract_first(privdata->outgoing_queue)) != NULL)
	bb_smscconn_send_failed(conn, msg, SMSCCONN_FAILED_SHUTDOWN);
    if (privdata->rport > 0)
	gwthread_join(privdata->receiver_thread);
    mutex_lock(conn->flow_mutex);

    conn->status = SMSCCONN_DEAD;

    debug("bb.sms", 0, "EMI2[%s]: connection has completed shutdown.",
	  octstr_get_cstr(privdata->name));

    list_destroy(privdata->outgoing_queue, NULL);
    octstr_destroy(privdata->name);
    octstr_destroy(privdata->allow_ip);
    octstr_destroy(privdata->deny_ip);
    octstr_destroy(privdata->host);
    octstr_destroy(privdata->alt_host);
    octstr_destroy(privdata->my_number);
    octstr_destroy(privdata->username);
    octstr_destroy(privdata->password);