nanomsg / NNG 消息库完全教程 / 第 4 章：nanomsg C API 详解

#include <nanomsg/nn.h>          // 核心 API
#include <nanomsg/pair.h>        // PAIR 协议常量
#include <nanomsg/pubsub.h>      // PUB/SUB 协议常量
#include <nanomsg/reqrep.h>      // REQ/REP 协议常量
#include <nanomsg/pipeline.h>    // PUSH/PULL 协议常量
#include <nanomsg/bus.h>         // BUS 协议常量
#include <nanomsg/survey.h>      // SURVEY 协议常量

int nn_socket(int domain, int protocol);

// 创建 PAIR Socket
int pair_sock = nn_socket(AF_SP, NN_PAIR);

// 创建 PUB Socket
int pub_sock = nn_socket(AF_SP, NN_PUB);

// 创建 REQ Socket
int req_sock = nn_socket(AF_SP, NN_REQ);

// 创建 PUSH Socket
int push_sock = nn_socket(AF_SP, NN_PUSH);

int sock = nn_socket(AF_SP, NN_PAIR);
if (sock < 0) {
    fprintf(stderr, "nn_socket failed: %s (errno=%d)\n",
            nn_strerror(nn_errno()), nn_errno());
    exit(1);
}

int nn_bind(int sock, const char *url);

// TCP 绑定
int ep = nn_bind(sock, "tcp://*:5555");

// TCP 绑定到指定接口
int ep = nn_bind(sock, "tcp://eth0:5555");

// IPC 绑定
int ep = nn_bind(sock, "ipc:///tmp/myapp.ipc");

// inproc 绑定
int ep = nn_bind(sock, "inproc://mychannel");

// 多次绑定（同一 Socket 可绑定多个地址）
nn_bind(sock, "tcp://*:5555");
nn_bind(sock, "tcp://*:5556");

int nn_connect(int sock, const char *url);

// TCP 连接
int ep = nn_connect(sock, "tcp://192.168.1.100:5555");

// IPC 连接
int ep = nn_connect(sock, "ipc:///tmp/myapp.ipc");

// inproc 连接
int ep = nn_connect(sock, "inproc://mychannel");

// 连接多个服务端（REQ 可轮询多个 REP）
nn_connect(sock, "tcp://server1:5555");
nn_connect(sock, "tcp://server2:5555");
nn_connect(sock, "tcp://server3:5555");

// 创建端点
int ep = nn_connect(sock, "tcp://server:5555");

// 关闭特定端点（不影响其他端点）
nn_shutdown(sock, ep);

#include <nanomsg/nn.h>
#include <nanomsg/reqrep.h>
#include <stdio.h>
#include <string.h>

int main() {
    int sock = nn_socket(AF_SP, NN_REP);
    if (sock < 0) {
        fprintf(stderr, "Socket error: %s\n", nn_strerror(nn_errno()));
        return 1;
    }

    // 绑定多个地址
    int ep1 = nn_bind(sock, "tcp://*:5555");
    int ep2 = nn_bind(sock, "ipc:///tmp/rep.ipc");

    if (ep1 < 0 || ep2 < 0) {
        fprintf(stderr, "Bind error: %s\n", nn_strerror(nn_errno()));
        nn_close(sock);
        return 1;
    }

    printf("Listening on tcp://*:5555 and ipc:///tmp/rep.ipc\n");

    // ... 收发消息 ...

    // 关闭特定端点
    nn_shutdown(sock, ep1);

    // 关闭 Socket
    nn_close(sock);
    return 0;
}

int nn_send(int sock, const void *buf, size_t len, int flags);

const char *msg = "Hello, nanomsg!";
int bytes = nn_send(sock, msg, strlen(msg), 0);
if (bytes < 0) {
    fprintf(stderr, "nn_send: %s\n", nn_strerror(nn_errno()));
}

int bytes = nn_send(sock, msg, len, NN_DONTWAIT);
if (bytes < 0) {
    if (nn_errno() == EAGAIN) {
        printf("发送缓冲区满，稍后重试\n");
    }
}

struct nn_iov iov[3];
iov[0].iov_base = "Header: ";
iov[0].iov_len  = 8;
iov[1].iov_base = "Body content";
iov[1].iov_len  = 12;
iov[2].iov_base = " | Footer";
iov[2].iov_len  = 9;

int bytes = nn_sendmsg(sock, iov, 3, 0);

// 分配消息缓冲区
void *msg = nn_allocmsg(13, 0);
memcpy(msg, "Hello, NNG!", 13);

// 发送（零拷贝，发送后 msg 被 nanomsg 接管）
nn_send(sock, &msg, NN_MSG, 0);
// 注意：发送成功后不要再释放 msg

int nn_recv(int sock, void *buf, size_t len, int flags);

char buf[256];
int bytes = nn_recv(sock, buf, sizeof(buf), 0);
if (bytes > 0) {
    printf("Received %d bytes: %.*s\n", bytes, bytes, buf);
}

void *buf = NULL;
int bytes = nn_recv(sock, &buf, NN_MSG, 0);
if (bytes > 0) {
    printf("Received %d bytes: %.*s\n", bytes, bytes, (char *)buf);
    nn_freemsg(buf);  // 必须手动释放
}

struct nn_iov iov[3];
iov[0].iov_base = malloc(100);
iov[0].iov_len  = 100;
iov[1].iov_base = malloc(1000);
iov[1].iov_len  = 1000;
iov[2].iov_base = malloc(100);
iov[2].iov_len  = 100;

int bytes = nn_recvmsg(sock, iov, 3, 0);

char buf[256];
int bytes = nn_recv(sock, buf, sizeof(buf), NN_DONTWAIT);
if (bytes < 0) {
    if (nn_errno() == EAGAIN) {
        printf("没有消息可接收\n");
    }
}

int nn_setsockopt(int sock, int level, int option,
                  const void *val, size_t len);

// 设置接收超时 5 秒
int timeout = 5000;
nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVTIMEO, &timeout, sizeof(timeout));

// 设置发送超时 3 秒
timeout = 3000;
nn_setsockopt(sock, NN_SOL_SOCKET, NN_SNDTIMEO, &timeout, sizeof(timeout));

// 现在 recv/send 会在超时后返回 -1，errno 为 EAGAIN
char buf[256];
int bytes = nn_recv(sock, buf, sizeof(buf), 0);
if (bytes < 0 && nn_errno() == EAGAIN) {
    printf("接收超时\n");
}

// 订阅所有消息
const char *all = "";
nn_setsockopt(sock, NN_SUB, NN_SUB_SUBSCRIBE, all, 0);

// 订阅 "weather" 前缀
const char *weather = "weather";
nn_setsockopt(sock, NN_SUB, NN_SUB_SUBSCRIBE, weather, strlen(weather));

// 取消订阅 "weather"
nn_setsockopt(sock, NN_SUB, NN_SUB_UNSUBSCRIBE, weather, strlen(weather));

int val;
size_t len = sizeof(val);
nn_getsockopt(sock, NN_SOL_SOCKET, NN_RCVBUF, &val, &len);
printf("Receive buffer: %d bytes\n", val);

#include <nanomsg/nn.h>
#include <nanomsg/pair.h>
#include <nanomsg/pubsub.h>
#include <stdio.h>
#include <string.h>

int main() {
    int pair = nn_socket(AF_SP, NN_PAIR);
    int sub  = nn_socket(AF_SP, NN_SUB);

    nn_bind(pair, "tcp://*:5560");
    nn_connect(sub, "tcp://localhost:5561");

    const char *filter = "";
    nn_setsockopt(sub, NN_SUB, NN_SUB_SUBSCRIBE, filter, 0);

    // 设置 poll
    struct nn_pollfd fds[2];
    fds[0].fd = pair;
    fds[0].events = NN_POLLIN;  // 监听可读
    fds[1].fd = sub;
    fds[1].events = NN_POLLIN;

    // 超时 5000ms
    int rc = nn_poll(fds, 2, 5000);
    if (rc == 0) {
        printf("Timeout, no events.\n");
    } else if (rc > 0) {
        if (fds[0].revents & NN_POLLIN) {
            char *buf = NULL;
            nn_recv(pair, &buf, NN_MSG, 0);
            printf("PAIR: %s\n", (char *)buf);
            nn_freemsg(buf);
        }
        if (fds[1].revents & NN_POLLIN) {
            char *buf = NULL;
            nn_recv(sub, &buf, NN_MSG, 0);
            printf("SUB: %s\n", (char *)buf);
            nn_freemsg(buf);
        }
    } else {
        fprintf(stderr, "nn_poll error: %s\n", nn_strerror(nn_errno()));
    }

    nn_close(pair);
    nn_close(sub);
    return 0;
}

#include <nanomsg/nn.h>
#include <nanomsg/pubsub.h>
#include <stdio.h>

int main() {
    int front = nn_socket(AF_SP, NN_XPUB);
    int back  = nn_socket(AF_SP, NN_XSUB);

    // XPUB/XSUB 是 PUB/SUB 的中间代理版本
    nn_bind(front, "tcp://*:5560");  // 订阅者连接这里
    nn_bind(back,  "tcp://*:5561");  // 发布者连接这里

    printf("Proxy running: XSUB:5561 -> XPUB:5560\n");

    // 阻塞，转发消息
    nn_device(front, back);

    // nn_device 永不返回（除非出错）
    nn_close(front);
    nn_close(back);
    return 0;
}

#include <nanomsg/nn.h>
#include <nanomsg/reqrep.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

int main() {
    // 1. 创建 Socket
    int sock = nn_socket(AF_SP, NN_REP);
    if (sock < 0) {
        fprintf(stderr, "nn_socket: %s\n", nn_strerror(nn_errno()));
        return 1;
    }

    // 2. 设置选项
    int timeout = 10000;  // 10 秒接收超时
    nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVTIMEO, &timeout, sizeof(timeout));

    int bufsize = 1048576;  // 1MB 接收缓冲区
    nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVBUF, &bufsize, sizeof(bufsize));

    // 3. 绑定地址
    int ep = nn_bind(sock, "tcp://*:5555");
    if (ep < 0) {
        fprintf(stderr, "nn_bind: %s\n", nn_strerror(nn_errno()));
        nn_close(sock);
        return 1;
    }
    printf("Server started on tcp://*:5555\n");

    // 4. 消息循环
    while (1) {
        // 接收消息（零拷贝）
        void *buf = NULL;
        int bytes = nn_recv(sock, &buf, NN_MSG, 0);
        if (bytes < 0) {
            if (nn_errno() == EAGAIN) {
                printf("Timeout, continuing...\n");
                continue;
            }
            fprintf(stderr, "nn_recv: %s\n", nn_strerror(nn_errno()));
            break;
        }

        printf("Received %d bytes: %.*s\n", bytes, bytes, (char *)buf);
        nn_freemsg(buf);

        // 发送响应
        const char *reply = "OK";
        nn_send(sock, reply, strlen(reply), 0);
    }

    // 5. 关闭
    nn_shutdown(sock, ep);
    nn_close(sock);

    printf("Server stopped.\n");
    return 0;
}