tux_proxy.c

使用python调用tuxedo服务的代码, 中间以 c 调用tuxedo客户端的atmi接口来作为连接代理

/*手机邮箱查询用户主资费接口服务程序

*/

#include <stdio.h>
#include <string.h>
#include <errno.h>

#include <time.h>
#include <sys/time.h>

#include <fml.h>
#include <fml32.h>
#include <atmi.h>
#include <pthread.h>
#include <semaphore.h>

#define L "tux_proxy"


#define GPARM32_0               0x0a000c80    /* number: 3200  type: string */
#define SEND_PARMS_NUM32        0x0a000ce4    /* number: 3300  type: string */
#define RECP_PARMS_NUM32        0x0a000ce6    /* number: 3302  type: string */
#define SVC_ERR_NO32            0x0a000cf8    /* number: 3320  type: string */
#define SVC_ERR_MSG32           0x0a000cf9    /* number: 3321  type: string */
#define F_ServiceName32         0x0a000d03    /* number: 3331  type: string */
#define F_Wsnaddr32             0x0a000d04    /* number: 3331  type: string */

struct seq_cd
{
    int seq;
    int cd;
};

#define MAX_CALL 200
struct seq_cd sc_list[MAX_CALL];


char *cal_time(char *s, struct timeval t1, struct timeval t2)
{
    if(t2.tv_usec<t1.tv_usec)
            sprintf(s, "%ld.%06ld",
         t2.tv_sec-t1.tv_sec-1, t2.tv_usec-t1.tv_usec+1000000);
    else
            sprintf(s, "%ld.%06ld", t2.tv_sec-t1.tv_sec, t2.tv_usec-t1.tv_usec);
    return s;
}

int getParm(FBFR32 *fbuf, int col, int row, char *r, int size)
{
    unsigned long l;
    l = size;
    memset(r, 0, l);
    if(col == -1)
        return Fget32(fbuf, SVC_ERR_NO32, 0, r, &l);
    else if(col == -2)
        return Fget32(fbuf, SVC_ERR_MSG32, 0, r, &l);
    if(col > 50)
        return Fget32(fbuf, col, row, r, &l);
    else
        return Fget32(fbuf, GPARM32_0+col, row, r, &l);
}

pthread_mutex_t mutex;
sem_t sem;

void *thread_getreply(void *parm)
{
    FILE *fw;
    int ret, i, cd, seq;
    FBFR32  *fb;
    long len;
    char tmp[50];

    fw = (FILE *)parm;
    
    file_log(L, "enter thread getreply");
    while(1){
        sem_wait(&sem);
        fb = (FBFR32 *)tpalloc("FML32", NULL, 1024);
        if(fb == NULL){
            file_log(L, "tpalloc failed\n");
            break;
        }
        len = Fsizeof32(fb);
        ret = tpgetrply(&cd, (char **)&fb, &len, TPGETANY);
        if(ret == -1){
            file_log(L, "tpgetreply failed %d", tperrno);
            break;
        }
        seq = -1;
        for(i=0; i<MAX_CALL; i++){
            if(sc_list[i].cd == cd){
                seq = sc_list[i].seq;
                sc_list[i].seq = 0;
                sc_list[i].cd  = 0;
                break;
            }
        }
        if(seq == -1){
            file_log(L, "can not found replied cd %d", cd);
            continue;
        }
        if(Funused32(fb) < 32){
            fb = (FBFR32  *)tprealloc((char *)fb, Fsizeof32(fb) + 32);
        }
        sprintf(tmp, "%d", seq);
        Fadds32(fb, F_Wsnaddr32, tmp);
        pthread_mutex_lock(&mutex);
        ret = Fwrite32(fb, fw);
        fflush(fw);
        pthread_mutex_unlock(&mutex);
        if(ret == -1){
            printf("Fwrite32 failed [%d], seq:%d\n", Ferror, seq);
            break;
        }
        tpfree((char *)fb);
    }
    exit(0);
}

int send_error(FILE *fw, int seq, int errno, char *errmsg)
{
    FBFR32 *fbs;
    char tmp[50];
    int ret;

    fbs = (FBFR32 *)tpalloc("FML32", NULL, 1024);
    sprintf(tmp, "%d", errno); Fadds32(fbs, SVC_ERR_NO32, tmp);
    sprintf(tmp, "%d", seq); Fadds32(fbs, F_Wsnaddr32, tmp);
    Fadds32(fbs, SVC_ERR_MSG32, errmsg);
    pthread_mutex_lock(&mutex);
    ret = Fwrite32(fbs, fw);
    fflush(fw);
    pthread_mutex_unlock(&mutex);
    if(ret == -1){
        printf("Fwrite32 failed [%d]\n", Ferror);
        return -1;
    }
    tpfree((char *)fbs);
    return 0;
}

void t_call_back(int sockid, char *clientip, int clientport, void *parm, int servsockid)
{
    FILE *fr, *fw, *fw1;
    FBFR32  *fb, *fbs;
    long rcvlen;
    int ret, my_tperrno;
    char buf[1024], eno[7], emsg[2000], tmp[50];
    char service_name[30];
    char wsnaddr[50];
    struct timeval tp1, tp2;
    int tpret, i;
    struct seq_cd *sc;
    int cd, seq;
    pthread_t th1;

    printf("new client [%s:%d]\n", clientip, clientport);
    file_log(L, "new client [%s:%d]", clientip, clientport);

    memset(buf, 0, sizeof(buf));

    fr = fdopen(sockid, "rb");
    if(fr == NULL){
        file_log(L, "fdopen failed errno:%d", errno);
        tcp_close(sockid);
        return;
    }
    fw = fdopen(sockid, "wb");
    if(fw == NULL){
        file_log(L, "fdopen wb failed errno:%d", errno);
        tcp_close(sockid);
        return;
    }

    /*建立连接后, 读取需要连接的服务地址*/
    fb = (FBFR32 *)tpalloc("FML32", NULL, 1024);
    ret = Fread32(fb, fr);
    if(ret == -1){
        F_error32(buf);
        tpfree((char *)fb);
        file_log(L, "Fread32 failed [%s]", buf);
        tcp_close(sockid);
        return;
    }
    memset(wsnaddr, 0, sizeof(wsnaddr));
    getParm(fb, F_Wsnaddr32, 0, wsnaddr, sizeof(wsnaddr));
    if(strlen(wsnaddr) > 10){
        tuxputenv(wsnaddr);
        file_log(L, "new client wsnaddr:%s", wsnaddr);
    }else{
        file_log(L, "invalid wsnaddr value[%s], connect close", wsnaddr);
        tcp_close(sockid);
        return;
    }

    ret = tpinit(NULL);
    if(ret == -1){
        file_log(L, "tpinit failed, tperrno:%d", tperrno);
        tcp_close(sockid);
        return;
    }

    memset(&sc_list, 0, sizeof(sc_list));
    pthread_mutex_init(&mutex, NULL);
    sem_init( &sem, PTHREAD_PROCESS_PRIVATE, 0 );
    /*启动接收应答线程*/
    ret = pthread_create(&th1, NULL, thread_getreply, (void *)fw);
    

    while(1){
        gettimeofday(&tp1, NULL);
        fb = (FBFR32 *)tpalloc("FML32", NULL, 1000);
        ret = Fread32(fb, fr);
        if(ret == -1){
            F_error32(buf);
            tpfree((char *)fb);
            printf("Fread32 failed [%s]\n", buf);
            break;
        }
        getParm(fb, F_Wsnaddr32, 0, tmp, sizeof(tmp));
        seq = atoi(tmp); /*这个id存放的序列号, 请求端需要确保唯一*/
        getParm(fb, F_ServiceName32, 0, service_name, sizeof(service_name));
        file_log(L, "new tuxcall %s, seq:%d", service_name, seq);

        sc = NULL;
        for(i=0; i<MAX_CALL; i++){
            if(sc_list[i].seq == 0)
                sc = &sc_list[i];
        }
        if(sc == NULL){
            file_log(L, "too many call in q, seq:%d, service_name:%s", seq, service_name);
            sprintf(tmp, "too many call in q(%d)", MAX_CALL);
            if(send_error(fw, seq, 777001,  tmp) != 0)
                break;
            continue;
        }
        rcvlen = Fsizeof32(fb);
        cd = tpacall(service_name, (char *)fb, rcvlen, 0);
        my_tperrno = tperrno;
        tpfree((char *)fb);
        if(cd == -1)
        {
            file_log(L, "%s %s:Tperrno = %d\n",wsnaddr, service_name, my_tperrno);
            sprintf(tmp, "调用服务失败,tperrno:%d", my_tperrno);
            if(send_error(fw, seq, 777002, tmp) != 0){
                printf("Fwrite32 failed [%d]\n", Ferror);
                break;
            }
        }
        sem_post(&sem);

        /*将cd和seq的对应关系保存到列表中*/
        sc->cd = cd;
        sc->seq = seq;
    }
}


void test_fml()
{
    FBFR32 *fbs;
    int ret;
    FILE *fw;

    fbs = (FBFR32 *)tpalloc("FML32", NULL, 1024);
    Fadds32(fbs, SVC_ERR_NO32, "123456");
    Fadds32(fbs, F_Wsnaddr32, "F_Wsnaddr32");
    Fadds32(fbs, SVC_ERR_MSG32, "errmsg");

    fw = fopen("data1.dat", "wb");
    ret = Fwrite32(fbs, fw);
    fflush(fw);
    fclose(fw);

}

int main(int argc, char *argv[])
{
	int sd, psd;
	int port, ret, pid;
	char ip[50], buf[1024];
	char str[150];

    /*test_fml();
    if(1) return 0; */

    GenDaemon();

    printf("workdir[%s]\n", getenv("WORKDIR"));
    file_log(L, "welcome tux_proxy");
	port = 4777;
	if(argc > 1)
		port = atoi(argv[1]);
	sd = tcp_listen(NULL, port);
	if(sd <= 0){
		printf("tcp listen on %d failed\n", port);
		return -1;
	}

    tcp_server_loop(sd, t_call_back, NULL);
    tcp_close(sd);
	return 0;
}