update APUE socket project

guowenxue

3 hours ago 805b28fc2e633bcd9c823a4b23614378ebfd50a1

update APUE socket project

 project/2.socketd/.gitignore

New file
@@ -0,0 +1,8 @@
# running files
socket_server
socket_client
*.log
*.db

# openlibs
sqlite-*

 project/2.socketd/booster/database.c

@@ -32,7 +32,7 @@
 * $db_file: sqlite database file name
 * return value: <0: failure   0:ok
 * */
int database_init(const char *db_file)
int db_open(const char *db_file)
{
    char               sql[SQL_COMMAND_LEN]={0};
    char              *errmsg = NULL;
@@ -43,48 +43,40 @@
        return -1;
    }

    /*+------------------------------------------+
     *|   database already exist, just open it   |
     *+------------------------------------------+*/
    if( 0==access(db_file, F_OK) )
    /* open database */
    if( SQLITE_OK != sqlite3_open_v2(db_file, &s_clidb, SQLITE_OPEN_READWRITE|SQLITE_OPEN_CREATE, NULL) )
    {
        if( SQLITE_OK != sqlite3_open(db_file, &s_clidb) )
        {
            log_error("open database file '%s' failure\n", db_file);
            return -2;
        }
        log_info("open database file '%s' ok\n", db_file);
        return 0;
    }

    /*+-----------------------------------------+
     *|  database not exist, create and init it |
     *+-----------------------------------------+*/

    if( SQLITE_OK != sqlite3_open(db_file, &s_clidb) )
    {
        log_error("create database file '%s' failure\n", db_file);
        log_error("open/create database '%s' failed: %s\n", db_file, sqlite3_errmsg(s_clidb));
        sqlite3_close(s_clidb);
        return -2;
    }

    /* SQLite continues without syncing as soon as it has handed data off to the operating system */
    /* ---------------------------------------------
     * SQLite performance & storage tuning
     * --------------------------------------------- */

    /* Disable synchronous for higher performance */
    sqlite3_exec(s_clidb, "pragma synchronous = OFF; ", NULL, NULL, NULL);

    /* enable full auto vacuum, Auto increase/decrease  */
    sqlite3_exec(s_clidb, "pragma auto_vacuum = 2 ; ", NULL, NULL, NULL);

    /* Create firehost table in the database */
    snprintf(sql, sizeof(sql), "CREATE TABLE %s(packet BLOB);", TABLE_NAME);

    /* ---------------------------------------------
     * Create table if not exists
     * --------------------------------------------- */

    snprintf(sql, sizeof(sql), "CREATE TABLE IF NOT EXISTS %s (packet BLOB);", TABLE_NAME);
    if( SQLITE_OK != sqlite3_exec(s_clidb, sql, NULL, NULL, &errmsg) )
    {
        log_error("create data_table in database file '%s' failure: %s\n", db_file, errmsg);
        sqlite3_free(errmsg); /* free errmsg  */
        sqlite3_close(s_clidb);   /* close databse */
        unlink(db_file);      /* remove database file */
        log_error("create table '%s' failed: %s\n", TABLE_NAME, errmsg);
        sqlite3_free(errmsg);   /* free errmsg  */
        sqlite3_close(s_clidb); /* close databse */
        unlink(db_file);        /* remove file */
        return -3;
    }

    log_info("create and init database file '%s' ok\n", db_file);
    log_info("database '%s' initialized ok (table=%s)\n", db_file, TABLE_NAME);
    return 0;
}

@@ -92,7 +84,7 @@
/* description: close sqlite database handler
 * return value: none
 */
void database_term(void)
void db_close(void)
{
    log_warn("close sqlite database now\n");
    sqlite3_close(s_clidb);
@@ -101,164 +93,206 @@
}


/* description: push a blob packet into database
/*
 * description: write a blob packet into database
 * input args:
 *      $pack:  blob packet data address
 *      $size:  blob packet data bytes
 * return value: <0: failure   0:ok
 *   pack: blob packet data address
 *   size: blob packet data length
 * return value:
 *   <0: failure
 *    0: ok
 */
int database_push_packet(void *pack, int size)
int db_write(const void *pack, int size)
{
    char               sql[SQL_COMMAND_LEN]={0};
    int                rv = 0;
    sqlite3_stmt      *stat = NULL;
    static sqlite3_stmt     *stmt = NULL;
    int                     rv = 0;

    if( !pack || size<=0 )
    if (!pack || size <= 0)
    {
        log_error("%s() Invalid input arguments\n", __func__);
        log_error("%s(): invalid input arguments\n", __func__);
        return -1;
    }

    if( ! s_clidb )
    if (!s_clidb)
    {
        log_error("sqlite database not opened\n");
        return -2;
    }

    snprintf(sql, sizeof(sql), "INSERT INTO %s(packet) VALUES(?)", TABLE_NAME);
    rv = sqlite3_prepare_v2(s_clidb, sql, -1, &stat, NULL);
    if(SQLITE_OK!=rv || !stat)
    /* Prepare statement once */
    if (!stmt)
    {
        log_error("blob add sqlite3_prepare_v2 failure\n");
        rv = -2;
        goto OUT;
        const char *sql = "INSERT INTO " TABLE_NAME " (packet) VALUES (?);";
        if( SQLITE_OK != sqlite3_prepare_v2(s_clidb, sql, -1, &stmt, NULL) )
        {
            log_error("sqlite3_prepare_v2 failed: %s\n", sqlite3_errmsg(s_clidb));
            return -3;
        }
    }

    if( SQLITE_OK != sqlite3_bind_blob(stat, 1, pack, size, NULL) )
    /* Bind blob */
    if( SQLITE_OK != sqlite3_bind_blob(stmt, 1, pack, size, SQLITE_STATIC) )
    {
        log_error("blob add sqlite3_bind_blob failure\n");
        rv = -3;
        goto OUT;
    }

    rv = sqlite3_step(stat);
    if( SQLITE_DONE!=rv && SQLITE_ROW!=rv )
    {
        log_error("blob add sqlite3_step failure\n");
        log_error("sqlite3_bind_blob failed: %s\n", sqlite3_errmsg(s_clidb));
        rv = -4;
        goto OUT;
        goto out;
    }

OUT:
    sqlite3_finalize(stat);
    /* Execute */
    if( SQLITE_DONE != sqlite3_step(stmt) )
    {
        log_error("sqlite3_step failed: %s\n", sqlite3_errmsg(s_clidb));
        rv = -5;
        goto out;
    }

    if( rv < 0 )
        log_error("add new blob packet into database failure, rv=%d\n", rv);
    rv = 0;
out:
    sqlite3_clear_bindings(stmt);
    sqlite3_reset(stmt);

    if (rv == 0)
        log_debug("add new blob packet(size=%d) into database ok\n", size);
    else
        log_info("add new blob packet into database ok\n");
        log_error("add new blob packet(size=%d) into database failed, rv=%d\n", rv);

    return rv;
}


/* description: pop the first blob packet from database
/*
 * description: read the first blob packet and its rowid from database
 * input args:
 *      $pack:  blob packet output buffer address
 *      $size:  blob packet output buffer size
 *      $byte:  blob packet bytes
 * return value: <0: failure   0:ok
 *   pack     : blob output buffer
 *   size     : buffer size
 *   bytes    : actual blob size (output)
 *   rowid    : rowid of the record (output)
 *
 * return value:
 *   <0 : failure
 *    0 : ok
 */
int database_pop_packet(void *pack, int size, int *bytes)
int db_read(void *pack, int size, int *bytes, int *rowid)
{
    char               sql[SQL_COMMAND_LEN]={0};
    int                rv = 0;
    sqlite3_stmt      *stat = NULL;
    const void        *blob_ptr;
    static sqlite3_stmt     *stmt = NULL;
    const void              *blob_ptr;
    int                     rv = 0;

    if( !pack || size<=0 )
    if (!pack || size <= 0 || !bytes || !rowid)
    {
        log_error("%s() Invalid input arguments\n", __func__);
        log_error("%s(): invalid input arguments\n", __func__);
        return -1;
    }

    if( ! s_clidb )
    if (!s_clidb)
    {
        log_error("sqlite database not opened\n");
        return -2;
    }

    /* Only query the first packet record */
    snprintf(sql, sizeof(sql), "SELECT packet FROM %s WHERE rowid = (SELECT rowid FROM %s LIMIT 1);", TABLE_NAME, TABLE_NAME);
    rv = sqlite3_prepare_v2(s_clidb, sql, -1, &stat, NULL);
    if(SQLITE_OK!=rv || !stat)
    /* Prepare statement once */
    if (!stmt)
    {
        log_error("firehost sqlite3_prepare_v2 failure\n");
        rv = -3;
        const char *sql = "SELECT rowid, packet FROM " TABLE_NAME " ORDER BY rowid ASC LIMIT 1;";
        if( SQLITE_OK != sqlite3_prepare_v2(s_clidb, sql, -1, &stmt, NULL) )
        {
            log_error("sqlite3_prepare_v2 failed: %s\n", sqlite3_errmsg(s_clidb));
            return -3;
        }
    }

    /* Step to first row */
    if ( SQLITE_ROW != sqlite3_step(stmt) )
    {
        rv = -4;
        goto out;
    }

    rv = sqlite3_step(stat);
    if( SQLITE_DONE!=rv && SQLITE_ROW!=rv )
    /* get rowid */
    *rowid = (int)sqlite3_column_int64(stmt, 0);

    /* get blob */
    blob_ptr = sqlite3_column_blob(stmt, 1);
    *bytes = sqlite3_column_bytes(stmt, 1);

    if (!blob_ptr || *bytes <= 0)
    {
        log_error("firehost sqlite3_step failure\n");
        log_error("invalid blob data (bytes=%d)\n", *bytes);
        rv = -5;
        goto out;
    }

    /* 1rd argument<0> means first segement is packet  */
    blob_ptr = sqlite3_column_blob(stat, 0);
    if( !blob_ptr )
    if (*bytes > size)
    {
        log_error("blob too large (%d > %d)\n", *bytes, size);
        rv = -6;
        goto out;
    }

    *bytes = sqlite3_column_bytes(stat, 0);

    if( *bytes > size )
    {
        log_error("blob packet bytes[%d] larger than bufsize[%d]\n", *bytes, size);
        *bytes = 0;
        rv = -1;
    }

    memcpy(pack, blob_ptr, *bytes);
    rv = 0;

out:
    sqlite3_finalize(stat);
    sqlite3_reset(stmt);
    return rv;
}


/* description: remove the first blob packet from database
 * input args: none
/*
 * description: remove a blob packet by rowid
 * input args:
 *   rowid : rowid of the record to delete
 * return value: <0: failure   0:ok
 */
int database_del_packet(void)
int db_remove(int rowid)
{
    char               sql[SQL_COMMAND_LEN]={0};
    char              *errmsg = NULL;
    char        sql[SQL_COMMAND_LEN] = {0};
    char        *errmsg = NULL;
    int         changes = 0;
    static int  count = 0;

    if( ! s_clidb )
    if (!s_clidb)
    {
        log_error("sqlite database not opened\n");
        return -2;
        return -1;
    }

    /*  remove packet from db */
    memset(sql, 0, sizeof(sql));
    snprintf(sql, sizeof(sql), "DELETE FROM %s WHERE rowid = (SELECT rowid FROM %s LIMIT 1);", TABLE_NAME, TABLE_NAME);
    if( SQLITE_OK != sqlite3_exec(s_clidb, sql, NULL, 0, &errmsg) )
    if (rowid <= 0)
    {
        log_error("delete first blob packet from database failure: %s\n", errmsg);
        sqlite3_free(errmsg);
        log_error("invalid rowid (%d)\n", rowid);
        return -2;
    }
    log_warn("delete first blob packet from database ok\n");

    /*  Vacuum the database */
    sqlite3_exec(s_clidb, "VACUUM;", NULL, 0, NULL);
    /* Execute SQL */
    snprintf(sql, sizeof(sql), "DELETE FROM %s WHERE rowid = %d;", TABLE_NAME, rowid);
    if (SQLITE_OK != sqlite3_exec(s_clidb, sql, NULL, NULL, &errmsg))
    {
        log_error("remove packet failed (rowid=%d): %s\n",
                  rowid, errmsg ? errmsg : "unknown error");
        sqlite3_free(errmsg);
        return -3;
    }

    /* Check how many rows were actually deleted */
    changes = sqlite3_changes(s_clidb);
    if (changes == 0)
    {
        log_warn("no record deleted (rowid=%d)\n", rowid);
        return -4;
    }

    log_debug("remove packet ok (rowid=%d)\n", rowid);

    /*
     * Optional: vacuum after N successful deletions
     * WARNING: VACUUM is expensive and blocks DB access
     */
    if (++count >= 500)
    {
        log_info("trigger database VACUUM\n");
        sqlite3_exec(s_clidb, "VACUUM;", NULL, 0, NULL);
        count = 0;
    }

    return 0;
}

 project/2.socketd/booster/database.h

@@ -22,13 +22,13 @@
 *              $db_file: sqlite database file name
 * return value: <0: failure   0:ok
 * */
extern int database_init(const char *db_file);
extern int db_open(const char *db_file);


/*  description: close sqlite database handler
 * return value: none
 */
extern void database_term(void);
extern void db_close(void);


/*  description: push a blob packet into database
@@ -37,24 +37,26 @@
 *               $size:  blob packet data bytes
 * return value: <0: failure   0:ok
 */
extern int database_push_packet(void *pack, int size);
extern int db_write(const void *pack, int size);


/*  description: pop the first blob packet from database
 *   input args:
 *               $pack:  blob packet output buffer address
 *               $size:  blob packet output buffer size
 *               $byte:  blob packet bytes
 *               $pack :  blob packet output buffer address
 *               $size :  blob packet output buffer size
 *               $byte :  blob packet bytes
 *               $rowid:  rowid of the record
 * return value: <0: failure   0:ok
 */
extern int database_pop_packet(void *pack, int size, int *bytes);
extern int db_read(void *pack, int size, int *bytes, int *rowid);


/*  description: remove the first blob packet from database
 *   input args: none
 *   input args:
 *               $rowid:  rowid of the record
 * return value: <0: failure   0:ok
 */
extern int database_del_packet(void);
extern int db_remove(int rowid);


#endif   /* ----- #ifndef _DATABASE_H_  ----- */

 project/2.socketd/booster/packet.c

@@ -34,13 +34,15 @@

int get_time(struct tm *ptm)
{
    time_t          now;

    if( !ptm )
    {
        log_error("Invalid input arugments\n");
        return -1;
    }

    time_t now = time(NULL);
    now = time(NULL);
    localtime_r(&now, ptm);

    return 0;

 project/2.socketd/booster/socket.c

@@ -210,7 +210,7 @@
 *               $sock:  socket context pointer
 * return value: 1: connected   0:disconnected
 */
int socket_connected(socket_t *sock)
int socket_check(socket_t *sock)
{
    struct tcp_info   info;
    int               len=sizeof(info);
@@ -270,6 +270,7 @@
    if( !sock )
        return -1;

    /* close and clear previous socket */
    socket_term(sock);

    /*+--------------------------------------------------+
@@ -284,7 +285,7 @@
    /* If $host is a valid IP address, then don't use name resolution */
    if( inet_aton(sock->host, &inaddr) )
    {
        //log_info("%s is a valid IP address, don't use domain name resolution.\n", sock->host);
        //log_info("Hostname %s is an IP address, so domain resolution can be skipped.\n", sock->host);
        hints.ai_flags |= AI_NUMERICHOST;
    }

@@ -326,6 +327,8 @@
        if( 0 == rv )
        {
            sock->fd = sockfd;
            sock->connected = 1;
            socket_set_keepalive(sockfd, 10, 3);
            log_info("Connect to server[%s:%d] on fd[%d] successfully!\n", sock->host, sock->port, sockfd);
            break;
        }
@@ -590,6 +593,7 @@
        return -2;
    }

    /* If keepintvl and keepcnt are both 0, the system default configuration will be used. */
    if(keepintvl || keepcnt)
    {
        /*

 project/2.socketd/booster/socket.h

@@ -94,7 +94,7 @@
 *               $sock:  socket context pointer
 * return value: 1: connected   0:disconnected
 */
extern int socket_connected(socket_t *sock);
extern int socket_check(socket_t *sock);

/*  description: socket client connect to server
 *   input args:

 project/2.socketd/openlibs/sqlite/build.sh

@@ -6,7 +6,7 @@
PACK_SUFIX=tar.gz

# LingYun source code FTP server
LY_FTP=http://weike-iot.com:2211/src/
LY_FTP=http://master.weike-iot.com:2211/src/

# library download URL address
LIB_URL=$LY_FTP

 project/2.socketd/socket_client.c

@@ -27,7 +27,7 @@
#include "database.h"

#define PROG_VERSION               "v1.0.0"
#define DAEMON_PIDFILE             "/tmp/.socketc.pid"
#define DAEMON_PIDFILE             "/tmp/socketc.pid"

static void print_usage(char *progname)
{
@@ -46,15 +46,15 @@
    return;
}

int check_sample_time(time_t *last_time, int interval);

int main (int argc, char **argv)
{
    char               *progname=NULL;
    int                 daemon = 1;
    int                 rv;
    int                 rowid = 0;
    time_t              now;

    char               *logfile="sock_client.log";
    char               *logfile="/tmp/socketc.log";
    int                 loglevel=LOG_LEVEL_INFO;
    int                 logsize=10; /* logfile size max to 10K */

@@ -119,7 +119,6 @@
            default:
                break;
        }

    }

    if( !serverip || !port )
@@ -128,28 +127,36 @@
        return 0;
    }

    /* initial socket context */
    socket_init(&sock, serverip, port);

    /* open logger system */
    if( log_open(logfile, loglevel, logsize, THREAD_LOCK_NONE) < 0 )
    {
        fprintf(stderr, "Initial log system failed\n");
        fprintf(stderr, "Initial logger failed\n");
        return 1;
    }

    /* install signal proc handler */
    install_default_signal();

    /* check program already running or not, if already running then exit, or set running as daemon */
    if( check_set_program_running(daemon, DAEMON_PIDFILE) < 0 )
        goto cleanup;

    log_info("socket client start running.\n");

    if( database_init("sock_client.db") < 0 )
    /* open database for cache data */
    if( db_open("sock_client.db") < 0 )
    {
        return 2;
        fprintf(stderr, "Initial database failed\n");
        rv = 2;
        goto cleanup;
    }

    socket_init(&sock, serverip, port);
    /* check whether the program is running; if not, start it in the background. */
    if( check_set_program_running(daemon, DAEMON_PIDFILE) < 0 )
    {
        fprintf(stderr, "Program is already running. Exiting.\n");
        rv = 3;
        goto cleanup;
    }

    /* install signal handler */
    install_default_signal();

    log_info("socket client start running.\n");

    while( ! g_signal.stop )
    {
@@ -158,11 +165,10 @@
         * +----------------------------------+*/

        sample_flag = 0; /* clear sample flag */
        time(&now);      /* get current time */

        if( check_sample_time(&last_time, interval) )
        if( difftime(now, last_time) >= interval )
        {
            log_debug("start DS18B20 sample termperature\n");

            if( (rv=ds18b20_get_temperature(&pack_info.temper)) < 0 )
            {
                log_error("DS18B20 sample temperature failure, rv=%d\n", rv);
@@ -176,6 +182,7 @@
            pack_bytes = pack_proc(&pack_info, (uint8_t *)pack_buf, sizeof(pack_buf));
            log_dump(LOG_LEVEL_DEBUG, NULL, pack_buf, pack_bytes);

            last_time = now; /* update last sample time */
            sample_flag = 1; /* set sample flag */
        }

@@ -184,7 +191,7 @@
         * +---------------------------------+*/

        /* start connect to server if not connected */
        if( !socket_connected(&sock) )
        if( !socket_check(&sock) )
        {
            socket_connect(&sock);
        }
@@ -192,13 +199,14 @@
        /* +-------------------------------+
         * |      socket disconnect        |
         * +-------------------------------+*/
        if( !socket_connected(&sock) )
        if( !sock.connected )
        {
            if( sample_flag )
            {
                database_push_packet(pack_buf, pack_bytes);
                db_write(pack_buf, pack_bytes);
            }

            /* Bypass the data send procedure for socket disconnect */
            continue;
        }

@@ -213,13 +221,13 @@
            if( socket_send(&sock, pack_buf, pack_bytes) < 0 )
            {
                log_warn("socket send sample packet failure, save it in database now.\n");
                database_push_packet(pack_buf, pack_bytes);
                db_write(pack_buf, pack_bytes);
                continue;
            }
        }

        /*  socket send packet in database  */
        if( !database_pop_packet(pack_buf, sizeof(pack_buf), &pack_bytes) )
        /*  socket send cache packet */
        if( !db_read(pack_buf, sizeof(pack_buf), &pack_bytes, &rowid) )
        {
            log_debug("socket send database packet bytes[%d]: %s\n", pack_bytes, pack_buf);
            if( socket_send(&sock, pack_buf, pack_bytes) < 0 )
@@ -230,34 +238,21 @@
            else
            {
                log_warn("socket send database packet okay, remove it from database now.\n");
                database_del_packet();
                db_remove(rowid);
            }
        }

        msleep(5);
    }

    rv = 0;

cleanup:
    socket_term(&sock);
    database_term();
    unlink(DAEMON_PIDFILE);
    db_close();
    log_close();
    unlink(DAEMON_PIDFILE);

    return 0;
    return rv;
}

int check_sample_time(time_t *last_time, int interval)
{
    int                  need = 0; /* no need sample now */
    time_t               now;

    time(&now);

    if( difftime(now, *last_time)>interval )
    {
        need = 1; /* need sample now  */
        *last_time = now;
    }

    return need;
}

 project/2.socketd/socket_server.c

@@ -25,7 +25,7 @@
#include "packet.h"

#define PROG_VERSION               "1.0.0"
#define DAEMON_PIDFILE             "/tmp/.sockets.pid"
#define DAEMON_PIDFILE             "/tmp/sockets.pid"
#define MAX_EVENTS                 512

/* Every client need get a stand alone buffer to save TLV packet data */
@@ -63,7 +63,7 @@
    int                   i = 0;
    int                   rv = 0;

    char                 *logfile="sock_server.log";
    char                 *logfile="/tmp/sockets.log";
    int                   loglevel=LOG_LEVEL_INFO;
    int                   logsize=10; /* logfile size max to 10K */

@@ -252,6 +252,7 @@

cleanup:
    socket_term(&sock);
    unlink(DAEMON_PIDFILE);
    log_close();
    return 0;
}