test2_main.c

与secs相关的hsms协议

static void* thread_func3(void*);

static sem_t cnx_ready_sem;

int main(int argc, char **argv)
{
    int print_offset;
    int r, sent[4], recv[4], err[4];
    hsmsd_handle_t h;

    //fp = fopen("./secondaries.txt", "w+");
    memset(msg_flags, 0, sizeof(msg_flags));

    r = hsmsd_alloc_handle(&h, "hsms1");
    fprintf(stdout, "hsmsd_alloc_handle returned %d\n", r);
    if(r) exit(-1);

    r = hsmsd_subscribe_for_msgs(h, (hsmsd_msg_handler_t)hsms_msg_handler );
    fprintf(stdout, "hsmsd_subscribe_for_msgs returned %d\n", r);
    if(r) exit(-2);

    r = hsmsd_subscribe_for_cnx_state(h, (hsmsd_cnx_state_handler_t)hsms_cnx_state_handler);
    fprintf(stdout, "hsmsd_subscribe_for_cnx_state returned %d\n", r);
    if(r) exit(-3);

    r = hsmsd_subscribe_for_cnx_error(h, (hsmsd_cnx_error_handler_t)hsmsd_cnx_error_handler);
    fprintf(stdout, "hsmsd_subscribe_for_cnx_error returned %d\n", r);
    if(r) exit(-4);

    sem_init(&cnx_ready_sem, 0, 0);

    r = hsmsd_cnx_start(h);
    fprintf(stdout, "hsmsd_cnx_start returned %d\n", r);
    if(r) exit(-5);

    sem_wait(&cnx_ready_sem);

    prepare_dict();
    fprintf(stdout, "dict prepared\n");

    start_threads((void*)&h);
    fprintf(stdout, "threads started\n");

    while(1)
    {
        sleep(2);

        sem_getvalue(&msg_dict[0].sent_sem, &sent[0]);
        sem_getvalue(&msg_dict[1].sent_sem, &sent[1]);
        sem_getvalue(&msg_dict[2].sent_sem, &sent[2]);
        sem_getvalue(&msg_dict[3].sent_sem, &sent[3]);

        sem_getvalue(&msg_dict[0].recv_sem, &recv[0]);
        sem_getvalue(&msg_dict[1].recv_sem, &recv[1]);
        sem_getvalue(&msg_dict[2].recv_sem, &recv[2]);
        sem_getvalue(&msg_dict[3].recv_sem, &recv[3]);

        sem_getvalue(&msg_dict[0].err_sem, &err[0]);
        sem_getvalue(&msg_dict[1].err_sem, &err[1]);
        sem_getvalue(&msg_dict[2].err_sem, &err[2]);
        sem_getvalue(&msg_dict[3].err_sem, &err[3]);
        
        print_offset = 0;
        print_offset += fprintf(stdout, 
                                "sent:\t%d\t|\t%d\t|\t%d\t|\t%d\n", 
                                 sent[0], sent[1], sent[2], sent[3]);
        print_offset += fprintf(stdout, 
                                "recv:\t%d\t|\t%d\t|\t%d\t|\t%d\n", 
                                 recv[0], recv[1], recv[2], recv[3]);
        print_offset += fprintf(stdout, 
                                "err:\t%d\t|\t%d\t|\t%d\t|\t%d\n", 
                                 err[0], err[1], err[2], err[3]);
        //fseek(stdout, print_offset, SEEK_SET);
        fflush(stdout);

/*        if(sent[0] > recv[0]+MAX_DELTA
        || sent[1] > recv[1]+MAX_DELTA
        || sent[2] > recv[2]+MAX_DELTA
        || sent[3] > recv[3]+MAX_DELTA)
        {
            fprintf(stdout, "sent-recv delta too big!");
            exit(0);
        }
*/
    }


    exit(0);
}



void prepare_dict(void)
{
    sem_init(&msg_dict[0].sent_sem, 0, 0);
    sem_init(&msg_dict[1].sent_sem, 0, 0);
    sem_init(&msg_dict[2].sent_sem, 0, 0);
    sem_init(&msg_dict[3].sent_sem, 0, 0);

    sem_init(&msg_dict[0].recv_sem, 0, 0);
    sem_init(&msg_dict[1].recv_sem, 0, 0);
    sem_init(&msg_dict[2].recv_sem, 0, 0);
    sem_init(&msg_dict[3].recv_sem, 0, 0);

    sem_init(&(msg_dict[0].err_sem), 0, 0);
    sem_init(&(msg_dict[1].err_sem), 0, 0);
    sem_init(&(msg_dict[2].err_sem), 0, 0);
    sem_init(&(msg_dict[3].err_sem), 0, 0);
}

void start_threads(void *pdata)
{
    pthread_t id0, id1, id2, id3;
    pthread_create(&id0, NULL, thread_func0, pdata);
    pthread_create(&id1, NULL, thread_func1, pdata);
    pthread_create(&id2, NULL, thread_func2, pdata);
    pthread_create(&id3, NULL, thread_func3, pdata);
}

void hsmsd_cnx_error_handler(int err)
{
    fprintf(stdout, "hsmsd_cnx_error_handler: %d\n", err);
}


void hsms_msg_handler(hsmsd_msg_t* msg)
{
    unsigned char *begin = NULL, *end = NULL, *iter = NULL;
    unsigned int ctr = 0;
    //fprintf(stdout, "hsms_msg_handler: s%df%d, sysbytes=%ld\n", 
    //                        msg->stream, msg->function, msg->sysbytes);
    
    if(fp && 0 == msg_flags[msg->stream][msg->function])
    {
        msg_flags[msg->stream][msg->function] = 1;
        
        fprintf(fp, "s%df%d, len=%d\n", msg->stream, msg->function, msg->data_len);
    
        begin   = &msg->data[0];
        end     = &msg->data[0] + msg->data_len;
        for(iter = begin; iter < end; ++iter)
        {
            fprintf(fp, "0x%x, ", *iter);
            if(ctr++ > 10)
            {
                fprintf(fp, "\n");
                ctr = 0;
            }
        }
        fprintf(fp, "\n\n\n");
        fflush(fp);
    }

    int pool_num = msg->sysbytes >> 24;
    if(4 > pool_num)
    {
        sem_post(&msg_dict[pool_num].recv_sem);
    }
    else
    {
        fprintf(stdout, ">>>!!!!!unknown id: %d!!!!!<<<\n",pool_num);
    }
    free(msg);
}

void hsms_cnx_state_handler(hsmsd_cnx_state_t state)
{
    static char str_state[][32] = {"NOT CONNECTED", "NOT SELECTED", "SELECTED"};

    if(CNX_SELECTED == state)
    {
        sem_post(&cnx_ready_sem);
    }
    fprintf(stdout, "hsms_cnx_state_handler: %s\n", str_state[state]);
}

static void* thread_funcx(int num, void* data)
{
    fprintf(stdout, "thread function %d\n", num);
    hsmsd_handle_t h = *((hsmsd_handle_t*)data);
    while(1)
    {
        usleep(msg_dict[num].us_timeout);
        msg_dict[num].pmsg->sysbytes += 1;
        if(0 == hsmsd_cnx_send_msg(h, msg_dict[num].pmsg))
        {
            sem_post(&msg_dict[num].sent_sem);
        }
        else
        {
            fprintf(stdout, "thread function %d: exit...\n", num);
            sem_post(&msg_dict[num].err_sem);
            pthread_exit(NULL);
        }
    }
    fprintf(stdout, "thread function %d: exit...\n", num);
    return NULL;
}

void* thread_func0(void* data)
{
    return thread_funcx(0, data);
}

void* thread_func1(void* data)
{
    return thread_funcx(1, data);
}

void* thread_func2(void* data)
{
    return thread_funcx(2, data);
}

void* thread_func3(void* data)
{
    return thread_funcx(3, data);
}