test case for malamute issue 81

malamute.cfg

#   Malamute configuration

#   Apply to the whole broker
server
    timeout = 10000     #   Client connection timeout, msec
    background = 0      #   Run as background process
    workdir = .         #   Working directory for daemon
    verbose = 0         #   Do verbose logging of activity?
    auth
        verbose = 1     #   Debug authentication steps?
        plain = passwords.cfg

#   Apply to the Malamute service
mlm_server
    echo = binding Malamute service to 'ipc://@/malamute'
    security
        mechanism = null
    bind
        endpoint = ipc://@/malamute

test-issue81.c

/*
 * Test case for https://github.com/zeromq/malamute/issues/81
 *
 * Michal Vyskocil <michalvyskocil@eaton.com>, licensed under Public Domain
 *
 */

#include <malamute.h>

#define ENDPOINT "ipc://@/malamute"
#define STREAM "stream"
#define TIMEOUT 500

static void
s_producer(zsock_t *pipe, void *args) {
    int* p_id = (int*) args;

    char* name;
    asprintf(&name, "producer-%d", *p_id);

    mlm_client_t *client = mlm_client_new();
    assert (client);
    
    int r = mlm_client_connect(client, ENDPOINT, 5000, name);
    assert (r == 0);

    mlm_client_set_producer(client, STREAM);

    int cnt = 0;
    zsock_signal(pipe, 0);
    while(!zsys_interrupted) {
        mlm_client_sendx(client, "SUBJECT", "hello", NULL);
        zsys_debug("[%s] sent %d messages", name, ++cnt);
        zclock_sleep(TIMEOUT);
    }
    mlm_client_destroy(&client);
    zsock_destroy(&pipe);
}

static void
s_consumer(zsock_t *pipe, void *args) {
    int* p_id = (int*) args;

    char* name;
    asprintf(&name, "consumer-%d", *p_id);

    mlm_client_t *client = mlm_client_new();
    assert (client);
    
    int r = mlm_client_connect(client, ENDPOINT, 5000, name);
    assert (r == 0);

    mlm_client_set_consumer(client, STREAM, ".*");

    int cnt = 0;
    zsock_signal(pipe, 0);
    while(!zsys_interrupted) {
        char *subject, *msg;
        mlm_client_recvx(client, &subject, &msg, NULL);
        zsys_debug("got subject=%s, msg=%s", subject, msg);
        free(subject);
        free(msg);
    }
    mlm_client_destroy(&client);
    zsock_destroy(&pipe);
}

int main() {

    //  Start a server to test against, and bind to endpoint
    zactor_t *server = zactor_new (mlm_server, ENDPOINT);
    zstr_sendx (server, "LOAD", "malamute.cfg", NULL);

    int id = 1;
    zactor_t *producer1 = zactor_new(s_producer, &id);
    assert(producer1);
    zactor_t *consumer1 = zactor_new(s_consumer, &id);
    assert(consumer1);

    zpoller_t *poller = zpoller_new(producer1, consumer1, NULL);
    assert(poller);

    while(!zpoller_terminated(poller)) {
        zsock_t *which = zpoller_wait(poller, -1);
        break;
    }

    zactor_destroy(&server);
    zactor_destroy(&producer1);
    zactor_destroy(&consumer1);
    zpoller_destroy(&poller);
}