場(chǎng)景說(shuō)明
選擇ENet代替TCP用于弱網(wǎng)環(huán)境(通常丟包率高)的數(shù)據(jù)傳輸，提高可靠性及傳輸效率。為了說(shuō)明怎樣正確有效地應(yīng)用ENet，本文按照TCP C/S同步通信的流程作了對(duì)應(yīng)的接口封裝實(shí)現(xiàn)，取庫(kù)名為
rudp。
接口對(duì)照
左邊為rudp庫(kù)的API，右邊為標(biāo)準(zhǔn)的Berkeley套接字API。rudp庫(kù)所有API前綴為rudp，rudp_listen和rudp_accept僅用于服務(wù)端，rudp_connect和rudp_disconnect僅用于客戶端；其它接口共用于兩端，其中rudp_send調(diào)用rudp_sendmsg實(shí)現(xiàn)，rudp_recv調(diào)用rudp_recvmsg實(shí)現(xiàn)。
具體實(shí)現(xiàn)
所有接口遵循Berkeley套接字接口的語(yǔ)義，為簡(jiǎn)單起見(jiàn)，錯(cuò)誤描述輸出到標(biāo)準(zhǔn)錯(cuò)誤流。
◆
監(jiān)聽(tīng)，成功返回0，失敗返回-1
1 int rudp_listen(const char *ip, int port, ENetHost **host)
2 {
3 ENetAddress address;
4
5 if(!strcmp(ip, "*"))
6 ip = "0.0.0.0";
7
8 if(enet_address_set_host_ip(&address, ip)){
9 fprintf(stderr, "enet_address_set_host_ip %s fail", ip);
10 return -1;
11 }
12
13 address.port = port;
14
15 assert(host);
16 *host = enet_host_create(&address, 1, 1, 0, 0);
17 if(NULL==*host){
18 fprintf(stderr, "enet_host_create %s:%d fail", address.host, address.port);
19 return -1;
20 }
21
22 int size = 1024*1024*1024;
23 if(enet_socket_set_option((*host)->socket, ENET_SOCKOPT_RCVBUF, size)){
24 fprintf(stderr, "enet set server socket rcvbuf %d bytes fail", size);
25 }
26
27 return 0;
28 }
◆
接受連接，成功返回0，失敗返回-1
1 int rudp_accept(ENetHost *host, unsigned int timeout, ENetPeer **peer)
2 {
3 int ret;
4 ENetEvent event;
5
6 ret = enet_host_service(host, &event, timeout);
7 if(ret > 0){
8 if(event.type != ENET_EVENT_TYPE_CONNECT){
9 if(event.type == ENET_EVENT_TYPE_RECEIVE)
10 enet_packet_destroy(event.packet);
11 fprintf(stderr, "enet_host_service event type %d is not connect", event.type);
12 return -1;
13 }
14
15 assert(peer);
16 *peer = event.peer;
17
18 }else if(0==ret){
19 fprintf(stderr, "enet_host_service timeout %d", timeout);
20 return -1;
21
22 }else{
23 fprintf(stderr, "enet_host_service fail");
24 return -1;
25 }
26
27 return 0;
28 }
◆
建立連接，成功返回0，失敗返回-1，conn_timeout是連接超時(shí)，rw_timeout是收發(fā)超時(shí)，單位為毫秒
1 int rudp_connect(const char *srv_ip, int srv_port, unsigned int conn_timeout, unsigned int rw_timeout, ENetHost **host, ENetPeer **peer)
2 {
3 assert(host);
4 *host = enet_host_create(NULL, 1, 1, 0, 0);
5 if(NULL==*host){
6 fprintf(stderr, "enet_host_create fail");
7 goto fail;
8 }
9 if(enet_socket_set_option((*host)->socket, ENET_SOCKOPT_RCVBUF, 1024*1024*1024)){
10 fprintf(stderr, "enet set server socket rcvbuf 1M bytes fail");
11 }
12
13 ENetAddress srv_addr;
14 if(enet_address_set_host_ip(&srv_addr, srv_ip)){
15 fprintf(stderr, "enet_address_set_host_ip %s fail", srv_ip);
16 goto fail;
17 }
18 srv_addr.port = srv_port;
19
20 assert(peer);
21 *peer = enet_host_connect(*host, &srv_addr, 1, 0);
22 if(*peer==NULL){
23 fprintf(stderr, "enet_host_connect %s:%d fail", srv_ip, srv_port);
24 goto fail;
25 }
26
27 enet_peer_timeout(*peer, 0, rw_timeout, rw_timeout);
28
29 int cnt = 0;
30 ENetEvent event;
31
32 while(1){
33 ret = enet_host_service(*host, &event, 1);
34 if(ret == 0){
35 if(++cnt >= conn_timeout){
36 fprintf(stderr, "enet_host_service timeout %d", conn_timeout);
37 goto fail;
38 }
39
40 }else if(ret > 0){
41 if(event.type != ENET_EVENT_TYPE_CONNECT){
42 fprintf(stderr, "enet_host_service event type %d is not connect", event.type);
43 goto fail;
44 }
45 break; //connect successfully
46
47 }else{
48 fprintf(stderr, "enet_host_service fail");
49 goto fail;
50 }
51 }
52
53 #ifdef _DEBUG
54 char local_ip[16], foreign_ip[16];
55 ENetAddress local_addr;
56
57 enet_socket_get_address((*host)->socket, &local_addr);
58 enet_address_get_host_ip(&local_addr, local_ip, sizeof(local_ip));
59 enet_address_get_host_ip(&(*peer)->address, foreign_ip, sizeof(foreign_ip));
60
61 printf("%s:%d connected to %s:%d", local_ip, loca_addr.port, foreign_ip, (*peer)->address.port);
62 #endif
63
64 return 0;
65
66 fail:
67 if(*host) enet_host_destroy(*host);
68 return -1;
69 }
◆
斷開(kāi)連接，若成功則返回0，超時(shí)返回1，出錯(cuò)返回-1。先進(jìn)行優(yōu)雅關(guān)閉，如失敗再?gòu)?qiáng)制關(guān)閉
1 int rudp_disconnect(ENetHost *host, ENetPeer *peer)
2 {
3 int ret;
4
5 #ifdef _DEBUG
6 char local_ip[16], foreign_ip[16];
7 ENetAddress local_addr;
8
9 enet_socket_get_address(host->socket, &local_addr);
10 enet_address_get_host_ip(&local_addr, local_ip, sizeof(local_ip));
11 enet_address_get_host_ip(&peer->address, foreign_ip, sizeof(foreign_ip));
12
13 printf("%s:%d is disconnected from %s:%d", local_ip, local_addr.port, foreign_ip, peer->address.port);
14 #endif
15
16 ENetEvent event;
17 enet_peer_disconnect(peer, 0);
18
19 while((ret = enet_host_service(host, &event, peer->roundTripTime)) > 0){
20 switch (event.type){
21 case ENET_EVENT_TYPE_RECEIVE:
22 enet_packet_destroy (event.packet);
23 break;
24
25 case ENET_EVENT_TYPE_DISCONNECT:
26 ret = 0;
27 goto disconn_ok;
28 }
29 }
30
31 ret = 0==ret ? 1 : -1;
32
33 fprintf(stderr, "enet_host_service with timeout %d %s", peer->roundTripTime, 1==ret?"timeout":"failure");
34
35 enet_peer_reset(conn->peer);
36
37 disconn_ok:
38 enet_host_destroy(host);
39 return ret;
40 }
◆
發(fā)送數(shù)據(jù)，若成功則返回已發(fā)送數(shù)據(jù)的長(zhǎng)度，否則返回-1
1 int rudp_sendmsg(ENetHost *host, ENetPeer *peer, ENetPacket *packet)
2 {
3 int ret;
4
5 if(enet_peer_send(peer, 0, packet)){
6 fprintf(stderr, "enet send packet %lu bytes to peer fail", packet->dataLength);
7 return -1;
8 }
9
10 ret = enet_host_service(host, NULL, peer->roundTripTime);
11 if(ret >= 0){
12 if(peer->state == ENET_PEER_STATE_ZOMBIE){
13 fprintf(stderr, "enet peer state is zombie");
14 return -1;
15 }
16 return packet->dataLength;
17
18 }else{
19 fprintf(stderr, "enet host service %u millsecond failure", peer->roundTripTime);
20 return -1;
21 }
22 }
23
24 int rudp_send(ENetHost *host, ENetPeer *peer, const void *buf, size_t len)
25 {
26 int ret;
27
28 ENetPacket *packet = enet_packet_create(buf, len, ENET_PACKET_FLAG_RELIABLE);
29 if(NULL==packet){
30 fprintf(stderr, "enet create packet %lu bytes fail", sizeof(int)+len);
31 return -1;
32 }
33
34 return rudp_sendmsg(host, peer, packet);
35 }
發(fā)送數(shù)據(jù)時(shí)需根據(jù)對(duì)端狀態(tài)判斷是否斷線，并且packet標(biāo)志設(shè)為可靠
◆
接收數(shù)據(jù)，若成功則返回已接收數(shù)據(jù)的長(zhǎng)度，否則返回-1
1 int rudp_recvmsg(ENetHost *host, ENetPeer *peer, ENetPacket **packet, unsigned int timeout)
2 {
3 int ret;
4 ENetEvent event;
5
6 ret = enet_host_service(host, &event, timeout);
7 if(ret > 0){
8 if(event.peer != peer){
9 fprintf(stderr, "enet receive peer is not matched");
10 goto fail;
11 }
12 if(event.type != ENET_EVENT_TYPE_RECEIVE){
13 fprintf(stderr, "enet receive event type %d is not ENET_EVENT_TYPE_RECEIVE", event.type);
14 goto fail;
15 }
16
17 *packet = event.packet;
18 return （*packet)->dataLength;
19
20 fail:
21 enet_packet_destroy(event.packet);
22 return -1;
23
24 }else {
25 fprintf(stderr, "enet receive %u millsecond %s", timeout, ret?"failure":"timeout");
26 return -1;
27 }
28 }
29
30 int rudp_recv(ENetHost *host, ENetPeer *peer, void *buf, size_t maxlen, unsigned int timeout)
31 {
32 ENetPacket *packet;
33
34 if(-1==rudp_recvmsg(host, peer, &packet, timeout))
35 return -1;
36
37 if(packet->dataLength > maxlen) {
38 fprintf(stderr, "enet packet data length %d is greater than maxlen %lu", packet->dataLength, maxlen);
39 return -1;
40 }
41
42 memcpy(buf, packet->data, packet->dataLength);
43 enet_packet_destroy(packet);
44
45 return packet->dataLength;
46 }
◆
等待所有確認(rèn)，若成功返回0，超時(shí)返回1，失敗返回-1
1 int rudp_wait_allack(ENetHost *host, ENetPeer *peer, unsigned int timeout)
2 {
3 int ret, cnt = 0;
4
5 while((ret = enet_host_service(host, NULL, 1)) >= 0){
6 if(enet_peer_is_empty_sent_reliable_commands(peer, 0,
7 ENET_PROTOCOL_COMMAND_SEND_RELIABLE|ENET_PROTOCOL_COMMAND_SEND_FRAGMENT))
8 return 0;
9
10 if(peer->state == ENET_PEER_STATE_ZOMBIE){
11 fprintf(stderr, "enet peer state is zombie");
12 return -1;
13 }
14
15 if(0==ret && ++cnt>=timeout){
16 return 1;
17 }
18 }
19
20 fprintf(stderr, "enet host service fail");
21 return -1;
22 }
等待已發(fā)送數(shù)據(jù)的所有確認(rèn)時(shí)，需根據(jù)對(duì)端狀態(tài)判斷是否斷線
示例流程
左邊為客戶端，壓縮并傳輸文件；右邊為服務(wù)端，接收并解壓存儲(chǔ)文件。
客戶端【讀文件塊并壓縮】這個(gè)環(huán)節(jié)，需要顯式創(chuàng)建可靠packet，并將壓縮后的塊拷貝到其中
posted on 2020-05-04 19:08
春秋十二月 閱讀(2367)
評(píng)論(0) 編輯 收藏 引用 所屬分類:
Network