update

wangzhengquan

2020-06-15 73866a4c527cbdf726c5fd824526d5657d0e15ee

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
#include "bipc.h"
#include <nng/protocol/pipeline0/pull.h>
#include <nng/protocol/pipeline0/push.h>
#include <nng/protocol/survey0/survey.h>
#include <nng/protocol/survey0/respond.h>
#include <nng/protocol/pubsub0/pub.h>
#include <nng/protocol/pubsub0/sub.h>
#include <nng/protocol/pair0/pair.h>
#include <nng/protocol/bus0/bus.h>
#include <nng/protocol/reqrep0/rep.h>
#include <nng/protocol/reqrep0/req.h>
 
void
static fatal(const char *func, int rv)
{
        fprintf(stderr, "%s: %s\n", func, nng_strerror(rv));
        exit(1);
}
 
int  bipc_listen(bipc_socket_t *sock, const char *url, bipc_mod_t mod) {
     
    int rv;
    switch(mod) {
        case PULL_PUSH:
            rv = nng_pull0_open(sock);
             break;
        case PAIR:
            rv = nng_pair0_open(sock);
            break;
        case BUS:
            rv = nng_bus0_open(sock);
            break;
        case REQ_REP:
            rv = nng_rep0_open(sock);
            break;
        case SURVEY:
            rv = nng_surveyor0_open(sock);
            break;
        case PUB_SUB:
            rv = nng_pub0_open(sock);
            break;
        default:
            fprintf(stderr, "无法识别的模式");
            return -1;
    }
 
    if (rv != 0) {
       // fatal("open", rv);
        return rv;
    }
 
    if ((rv = nng_listen(*sock, url, NULL, 0)) != 0) {
       // fatal("nng_listen", rv);
        return rv;
    }
    return 0;
 
}
 
int bipc_connect(bipc_socket_t *sock, const char *url, bipc_mod_t mod) {
    int rv  = 0 ;
    switch(mod) {
        case PULL_PUSH:
            rv = nng_push0_open(sock);
            break;
        case PAIR:
            rv = nng_pair0_open(sock);
            break;
        case BUS:
            break;
        case REQ_REP:
            rv = nng_req0_open(sock);
            break;
        case SURVEY:
            rv = nng_respondent0_open(sock);
            break;
        case PUB_SUB:
            if ((rv = nng_sub0_open(sock)) != 0) {
                //fatal("sub0_open", rv);
                return rv;
            }
            if ((rv = nng_setopt(*sock, NNG_OPT_SUB_SUBSCRIBE, "", 0)) != 0) {
                //fatal("nng_setopt", rv);
                return rv;
            }
            break;
        default:
            fprintf(stderr, "无法识别的模式");
            return rv;
    }
 
    if (rv != 0) {
        fatal("bipc_connect open socket", rv);
        return rv;
    }
 
    if ((rv = nng_dial(*sock, url, NULL, 0)) != 0) {
       // fatal("dial", rv);
        return rv;
    }
    return 0;
}
 
int bipc_send(bipc_socket_t *sock,  const void *data, size_t size) {
    return nng_send(*sock, const_cast<void *>(data), size, 0);
}
 
 
int bipc_recv(nng_socket *sock,  void *data, size_t *sizep) {
    
    //int rv = nng_recv(*sock, data, sizep, 0);
    // char *buf = NULL;
    // int rv = nng_recv(*sock, &buf, sizep, NNG_FLAG_ALLOC);
    // memcpy(data, buf, *sizep);
    // nng_free(buf, *sizep);
    int rv = nng_recv(*sock, data, sizep, NNG_FLAG_ALLOC);
    if (rv == NNG_ETIMEDOUT)
        return BIPC_ETIMEDOUT ;
    
    return rv;
}
 
 
void bipc_free(void *ptr, size_t size) {
     nng_free(ptr, size);
}
 
 
 
const char * bipc_strerror(int error) {
    return nng_strerror(error);
}