libuv_tcp_redis_server_example.c

libuv_tcp_redis_server_example.c

#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include "hiredis/hiredis.h"
#include <string.h>

#include "uv.h"

typedef struct {
    uv_write_t req;
    uv_buf_t buf;
} write_req_t;

static uint64_t data_cntr = 0;

static void alloc_cb(uv_handle_t *handle,
                     size_t suggested_size,
                     uv_buf_t *buf) {
    buf->base = malloc(suggested_size);
    assert(buf->base != NULL);
    buf->len = suggested_size;
}

static void on_close(uv_handle_t *handle) {
    free(handle);
}

static void after_shutdown(uv_shutdown_t *req, int status) {
    /*assert(status == 0);*/
    if (status < 0)
        fprintf(stderr, "err: %s\n", uv_strerror(status));
    fprintf(stderr, "data received: %lu byte\n", data_cntr);
    data_cntr = 0;
    uv_close((uv_handle_t *) req->handle, on_close);
    free(req);
}

static void after_write(uv_write_t *req, int status) {
    write_req_t *wr = (write_req_t *) req;

    if (wr->buf.base != NULL)
        free(wr->buf.base);
    free(wr);

    if (status == 0)
        return;

    fprintf(stderr, "uv_write error: %s\n", uv_strerror(status));

    //operation canceled
    if (status == UV_ECANCELED)
        return;

    //broken pipe
    assert(status == UV_EPIPE);

    /*
     * Request handle to be closed.
     * close_cb will be called asynchronously after this call.
     * This MUST be called on each handle before memory is released.
     * Moreover, the memory can only be released in close_cb or after it has returned.
     * Handles that wrap file descriptors are closed immediately but close_cb will still be deferred to the next iteration of the event loop.
     * It gives you a chance to free up any resources associated with the handle.
     * In-progress requests, like uv_connect_t or uv_write_t,
     * are cancelled and have their callbacks called asynchronously with status=UV_ECANCELED.
     */
    uv_close((uv_handle_t *) req->handle, on_close);
}

static void after_read(uv_stream_t *handle, ssize_t nread, const uv_buf_t *buf) {
    int r;
    write_req_t *wr;
    uv_shutdown_t *req;

    //如果数据读取异常，则进行内存释放
    if (nread <= 0 && buf->base != NULL)
        free(buf->base);

    if (nread == 0)
        return;

    if (nread < 0) {
        /*assert(nread == UV_EOF);*/
        //数据读取结束 或者 出现异常 ： 关闭客户端连接
        if (nread != UV_EOF) {
            fprintf(stderr, "err: %s\n", uv_strerror(nread));
        }

        req = (uv_shutdown_t *) malloc(sizeof(*req));
        assert(req != NULL);

        //关闭tcp客户端，并执行回调函数
        /*
         * Shutdown the outgoing (write) side of a duplex stream.
         * It waits for pending write requests to complete.
         * The handle should refer to a initialized stream.
         * req should be an uninitialized shutdown request struct.
         * The cb is called after shutdown is complete.
         */
        r = uv_shutdown(req, handle, after_shutdown);
        assert(r == 0);

        return;
    }


    data_cntr += nread;

    redisReader *reader = redisReaderCreate();

    redisReaderFeed(reader,buf->base,nread);
    redisReply *reply = (redisReply*)malloc(sizeof(redisReply));

    redisReaderGetReply(reader,&reply);

    //打印redis指令
//    for (int i = 0; i < reply->elements; ++i) {
//        printf("%s ",reply->element[i]->str);
//    }
//    printf("\n");

    //reply
    wr = (write_req_t *) malloc(sizeof(*wr));
    assert(wr != NULL);

    /*
     * Constructor for uv_buf_t.
     * Due to platform differences the user cannot rely on the ordering of the base and len members of the uv_buf_t struct.
     * The user is responsible for freeing base after the uv_buf_t is done. Return struct passed by value.
     */


    const char* OK_REPLY = "+OK\r\n";
    int replyLen = strlen(OK_REPLY);

    char *replyOK = (char*)malloc(sizeof(char)*replyLen);

    memcpy(replyOK,OK_REPLY,replyLen);

    wr->buf = uv_buf_init(replyOK, strlen(replyOK));

    /*
     * Write data to stream. Buffers are written in order.
     */
    r = uv_write(&wr->req, handle, &wr->buf, 1, after_write);
    assert(r == 0);
}

//libuv 客户端新建连接 回调
static void on_connection(uv_stream_t *server, int status) {

    assert(status == 0);

    //libuv tcp 对象内存分配
    uv_tcp_t *client = malloc(sizeof(uv_tcp_t));
    assert(client != NULL);

    /*初始化socket句柄
     * Initialize the handle. No socket is created as of yet.
     */
    int r = uv_tcp_init(uv_default_loop(), client);
    assert(r == 0);

    client->data = server;

    /* 接受客户端tcp链接
     * This call is used in conjunction with uv_listen() to accept incoming connections.
     * Call this function after receiving a uv_connection_cb to accept the connection.
     * Before calling this function the client handle must be initialized. < 0 return value indicates an error.
     * When the uv_connection_cb callback is called it is guaranteed that this function will complete successfully the first time.
     * If you attempt to use it more than once, it may fail. It is suggested to only call this function once per uv_connection_cb call.
     * Note server and client must be handles running on the same loop.
     */
    r = uv_accept(server, (uv_stream_t *) client);
    assert(r == 0);

    /* 开始读取数据
     * Read data from an incoming stream.
     * The uv_read_cb callback will be made several times until there is no more data to read or uv_read_stop() is called.
     */
    r = uv_read_start((uv_stream_t *) client, alloc_cb, after_read);
    assert(r == 0);
}

char *hostIPV4 = "127.0.0.1";
int hostPort = 8080;
const int BACK_LOG_SIZE = 256;

int main() {

    uv_loop_t *dl = uv_default_loop();

    uv_tcp_t tcpServer;

    uv_tcp_init(dl, &tcpServer);

    struct sockaddr ipv4Addr;

    uv_ip4_addr(hostIPV4, hostPort, (struct sockaddr_in *) &ipv4Addr);

    uv_tcp_bind(&tcpServer, &ipv4Addr, 0);

    int r = uv_listen((uv_stream_t *) &tcpServer, BACK_LOG_SIZE, on_connection);

    if (r) {
        fprintf(stderr, "Listen error %s\n", uv_strerror(r));
        return 1;
    } else {
        printf("Listening on %s:%d\n", hostIPV4, hostPort);
    }

    return uv_run(dl, UV_RUN_DEFAULT);

}