融合libevent和protobuf

写了一个简单的例子，把libevent中的bufferevent网络收发服务和protobuf里面的序列反序列结合起来。

protobuf文件message.proto：

message PMessage {
        required int32 id = 1;
        optional int32 num = 2;
        optional string str = 3;
}

生成接口命令：

protoc -I=proto --cpp_out=src proto/message.proto

服务器端 lserver.cc：

#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

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

#include <event.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <event2/thread.h>

#include "message.pb.h"

using namespace std;

void listener_cb(evconnlistener *listener, evutil_socket_t fd,
                sockaddr *sock, int socklen, void *arg);

void socket_read_cb(bufferevent *bev, void *arg);

void socket_event_cb(bufferevent *bev, short events, void *arg);

int main(int argc, char **argv) {
        sockaddr_in sin;
        memset(&sin, 0, sizeof(sockaddr_in));

        sin.sin_family = AF_INET;
        sin.sin_port = htons(8899);

        event_base *base = event_base_new();
        evconnlistener *listener
                = evconnlistener_new_bind(base,
                        listener_cb, base,
                        LEV_OPT_REUSEABLE|LEV_OPT_CLOSE_ON_FREE,
                        10, (sockaddr*)&sin, sizeof(sockaddr_in));

        event_base_dispatch(base);

        evconnlistener_free(listener);
        event_base_free(base);

}

void listener_cb(evconnlistener *listener, evutil_socket_t fd,
        sockaddr *sock, int socklen, void *arg) {

        printf("accept a client %d
", fd);
        event_base *base = (event_base *)arg;

        bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);

        bufferevent_setcb(bev, socket_read_cb, NULL, socket_event_cb, NULL);
        bufferevent_enable(bev, EV_READ|EV_PERSIST);

}

void socket_read_cb(bufferevent *bev, void *arg) {
        char msg[4096];
        size_t len = bufferevent_read(bev, msg, sizeof(msg)-1);
        msg[len] = '';

        PMessage pmsg;
        pmsg.ParseFromArray((const void*)msg, len);

        printf("Server read the data:%i, %i, %s
", pmsg.id(), pmsg.num(), pmsg.str().c_str());

        pmsg.set_str("I have read your data.");
        string sendbuf;
        pmsg.SerializeToString(&sendbuf);

        bufferevent_write(bev, sendbuf.c_str(), sendbuf.length());

}

void socket_event_cb(bufferevent *bev, short events, void *arg) {
        if (events & BEV_EVENT_EOF) {
                printf("connection close
");
        }
        else if (events & BEV_EVENT_ERROR) {
                printf("some other error
");
        }

        bufferevent_free(bev);
}

客户端lclient.cc

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>

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

#include <event.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/util.h>

#include "message.pb.h"

using namespace std;

void cmd_msg_cb(int fd, short events, void *arg);

void server_msg_cb(bufferevent *bev, void *arg);

void event_cb(bufferevent *bev, short event, void *arg);

static int gid = 1;

int main(int argc, char **argv) {
        if (argc < 3) {
                printf("please input IP and port
");
                return 1;
        }

        event_base *base = event_base_new();
        bufferevent *bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);

        event *ev_cmd = event_new(base, STDIN_FILENO,
                        EV_READ|EV_PERSIST,
                        cmd_msg_cb, (void *)bev);
        event_add(ev_cmd, NULL);

        sockaddr_in server_addr;
        memset(&server_addr, 0, sizeof(server_addr));

        server_addr.sin_family = AF_INET;
        server_addr.sin_port = htons(atoi(argv[2]));
        inet_aton(argv[1], &server_addr.sin_addr);

        bufferevent_socket_connect(bev, (sockaddr*)&server_addr, sizeof(server_addr));

        bufferevent_setcb(bev, server_msg_cb, NULL, event_cb, (void*)ev_cmd);
        bufferevent_enable(bev, EV_READ|EV_PERSIST);

        event_base_dispatch(base);

        printf("Finish
");
        return 0;

}

void cmd_msg_cb(int fd, short events, void *arg) {
        char msg[1024];
        int ret = read(fd, msg, sizeof(msg));
        if (ret < 0) {
                perror("read error.
");
                exit(1);
        }

        // protobuf
        PMessage pmsg;
        pmsg.set_id(gid++);
        pmsg.set_num(rand());
        pmsg.set_str(msg);

        string sendbuf;
        pmsg.SerializeToString(&sendbuf);

        // processing network transfer
        bufferevent *bev = (bufferevent *)arg;
        bufferevent_write(bev, sendbuf.c_str(), sendbuf.length());
}

void server_msg_cb(bufferevent *bev, void *arg) {
        char msg[1024];

        size_t len = bufferevent_read(bev, msg, sizeof(msg)-1);
        msg[len] = '';

        PMessage pmsg;
        pmsg.ParseFromArray((const void*)msg, len);

        printf("Recv %d, %d, %s from server.
", pmsg.id(), pmsg.num(), pmsg.str().c_str());
}

void event_cb(bufferevent *bev, short eventid, void *arg) {
        if (eventid & BEV_EVENT_EOF) {
                printf("Connection closed.
");
        }
        else if (eventid & BEV_EVENT_ERROR) {
                printf("Some other error.
");
        }
        else if (eventid & BEV_EVENT_CONNECTED) {
                printf("Client has successfully connected.
");
                return;
        }

        bufferevent_free(bev);
        event *ev = (event *)arg;
        event_free(ev);
}

服务器端和客户端共用的Makefile：

CXX=/opt/compiler/gcc-4.8.2/bin/g++

INCPATH= 
        /home/work/.jumbo/include/

DEP_LDFLAGS= 
        -L/home/work/.jumbo/lib/

DEP_LDLIBS= 
        -levent 
        -lprotobuf 
        -lpthread

TARGET= lserver lclient

all : $(TARGET)

lserver : lserver.cc message.pb.cc
        $(CXX) -o $@ $^ -I$(INCPATH) $(DEP_LDFLAGS) $(DEP_LDLIBS)

lclient : lclient.cc message.pb.cc
        $(CXX) -o $@ $^ -I$(INCPATH) $(DEP_LDFLAGS) $(DEP_LDLIBS)

.PHONY : all clean

clean :
        rm -rf $(TARGET)

服务器命令及输出：

src]$ ./lserver 
accept a client 7
Server read the data:1, 1804289383, aaaaaaaaaaaaaaaaaaaaaaaaa

Server read the data:2, 846930886, aa

Server read the data:3, 1681692777, bb

Server read the data:4, 1714636915, abcdefg

Server read the data:5, 1957747793, 

connection close
accept a client 7
Server read the data:1, 1804289383, 2aa

Server read the data:2, 846930886, 2bb

Server read the data:3, 1681692777, 111111111111111111111222222222222222222222223333333333333333333333333

Server read the data:4, 1714636915, 

^C

客户端命令及输出：

src]$ ./lclient localhost 8899
Client has successfully connected.
aaaaaaaaaaaaaaaaaaaaaaaaa
Recv 1, 1804289383, I have read your data. from server.
aa
Recv 2, 846930886, I have read your data. from server.
bb
Recv 3, 1681692777, I have read your data. from server.
abcdefg
Recv 4, 1714636915, I have read your data. from server.

Recv 5, 1957747793, I have read your data. from server.
^C
[src]$ ./lclient localhost 8899
Client has successfully connected.
2aa
Recv 1, 1804289383, I have read your data. from server.
2bb
Recv 2, 846930886, I have read your data. from server.
111111111111111111111222222222222222222222223333333333333333333333333
Recv 3, 1681692777, I have read your data. from server.

Recv 4, 1714636915, I have read your data. from server.
Connection closed.
Finish

注意：

1. 先后开了两个客户端。客户端退出，不影响服务器端。但是服务器端退出会让客户端一起退出，因为客户端在收到网络error信号处理的最后，会free掉从命令行读数据的监听event，这样eventbase就不会再有event需要监听了，所以会退出。

2. 开始在命令行输入的时候，在char数组中没有添加''，传输时会造成如下错误。

[libprotobuf ERROR google/protobuf/wire_format.cc:1053] String field contains invalid UTF-8 data when serializing a protocol buffer. Use the 'bytes' type if you intend to send raw bytes.

根据读入函数返回的长度，设置''即可避免这个错误。