pyzmq_1244.c

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

void freen(void *data, void *hint){
    free(data);
}

int main (int argc, char **argv)
{
    int server = 0;
    if (argc > 1)
        server = 1;
    
    void *ctx = zmq_ctx_new();
    assert(ctx);
    void *sock = zmq_socket(ctx, server ? ZMQ_PUSH : ZMQ_PULL);
    assert(sock);
    int opt = 1;
    assert(0 == zmq_setsockopt(sock, server ? ZMQ_SNDHWM : ZMQ_RCVHWM, &opt, sizeof(int)));
    if (server)
        assert(0 == zmq_bind(sock, "tcp://127.0.0.1:12345"));
    else
        assert(0 == zmq_connect(sock, "tcp://127.0.0.1:12345"));
    
    if (server) {
        if (argc > 2) {
            zmq_msg_t msg_data, msg_tmp;
            size_t len = 512 * 1024 * 1024;
            char *data = malloc(len);
            assert (data);
            memset(data, 'a', len);
            assert(0 == zmq_msg_init_data(&msg_data, data, len, freen, NULL));
            for (int i = 0; ; ++i) {
                assert(0 < zmq_send(sock, &i, sizeof(int), ZMQ_SNDMORE));
                zmq_msg_init(&msg_tmp);
                assert(0 == zmq_msg_copy(&msg_tmp, &msg_data));
                assert(0 < zmq_msg_send(&msg_tmp, sock, 0));
            }
            assert(0 == zmq_msg_close(&msg_data));
        } else {
            zmq_msg_t msg_data;
            size_t len = 512 * 1024 * 1024;
            for (int i = 0; ; ++i) {
                assert(0 < zmq_send(sock, &i, sizeof(int), ZMQ_SNDMORE));
                assert(0 == zmq_msg_init_size(&msg_data, len));
                memset(zmq_msg_data(&msg_data), 'a', len);
                assert(0 < zmq_msg_send(&msg_data, sock, 0));
            }
        }
    } else {
        zmq_msg_t msg;
        do {
            zmq_msg_close(&msg);
            zmq_msg_init(&msg);
            assert(0 < zmq_msg_recv(&msg, sock, 0));
            printf("%d\n", *(int *)zmq_msg_data(&msg));
        } while (zmq_msg_get (&msg, ZMQ_MORE));
        assert(0 == zmq_msg_close(&msg));
    }
    assert(0 == zmq_close(sock));
    assert(0 == zmq_ctx_term(ctx));
    return 0;
}