<NME> zproxy.c
<BEF> /*  =========================================================================
    zproxy - run a steerable proxy in the background

    Copyright (c) the Contributors as noted in the AUTHORS file.
    This file is part of CZMQ, the high-level C binding for 0MQ:
    http://czmq.zeromq.org.

    This Source Code Form is subject to the terms of the Mozilla Public
    License, v. 2.0. If a copy of the MPL was not distributed with this
    file, You can obtain one at http://mozilla.org/MPL/2.0/.
    =========================================================================
*/

/*
@header
    A zproxy actor switches messages between a frontend and a backend socket.
    It acts much like the zmq_proxy_steerable method, though it makes benefit
    of CZMQ's facilities, to be somewhat simpler to set-up.
@discuss
    This class replaces zproxy_v2, and is meant for applications that use the
    CZMQ v3 API (meaning, zsock).
@end
*/

#include "czmq_classes.h"

typedef enum proxy_socket {
    NONE = -1,
    FRONTEND,
    BACKEND,
    SOCKETS
}
proxy_socket;

typedef struct {
    int ivl;
    int timeout;
    int ttl;
} zproxy_heartbeat_t;

#define AUTH_NONE  0
#define AUTH_PLAIN 1
#define AUTH_CURVE 2

//  --------------------------------------------------------------------------
//  The self_t structure holds the state for one actor instance

typedef struct {
    zsock_t *pipe;              //  Actor command pipe
    zpoller_t *poller;          //  Socket poller
    zsock_t *frontend;          //  Frontend socket
    zsock_t *backend;           //  Backend socket
    zsock_t *capture;           //  Capture socket
    int auth_type [SOCKETS];    //  Auth type for sockets
    char *domain [SOCKETS];     //  Auth domains for sockets
    char *public_key [SOCKETS]; //  Public keys for sockets
    char *secret_key [SOCKETS]; //  Secret keys for sockets
    bool terminated;            //  Did caller ask us to quit?
    bool verbose;               //  Verbose logging enabled?
} self_t;


static self_t *
s_self_new (zsock_t *pipe)
{
    self_t *self = (self_t *) zmalloc (sizeof (self_t));
    assert (self);
    self->pipe = pipe;
    self->poller = zpoller_new (self->pipe, NULL);
    assert (self->poller);
    return self;
}

static void
s_self_destroy (self_t **self_p)
{
    assert (self_p);
    if (*self_p) {
        self_t *self = *self_p;
        zpoller_destroy (&self->poller);
        zsock_destroy (&self->frontend);
        zsock_destroy (&self->backend);
        zsock_destroy (&self->capture);
        int index;
        for (index = 0; index < SOCKETS; index++) {
            zstr_free (&self->domain [index]);
            zstr_free (&self->public_key [index]);
            zstr_free (&self->secret_key [index]);
        }
        freen (self);
        *self_p = NULL;
    }
}

static zsock_t *
s_self_create_socket (self_t *self, char *type_name, char *endpoints, zproxy_heartbeat_t *heartbeat, proxy_socket selected_socket)
{
    //  This array matches ZMQ_XXX type definitions
    assert (ZMQ_PAIR == 0);
    char *type_names [] = {
        "PAIR", "PUB", "SUB", "REQ", "REP",
        "DEALER", "ROUTER", "PULL", "PUSH",
        "XPUB", "XSUB", type_name
    };
    //  We always match type at least at end of table
    int index;
    for (index = 0; strneq (type_name, type_names [index]); index++) ;
    if (index > ZMQ_XSUB) {
        zsys_error ("zproxy: invalid socket type '%s'", type_name);
        return NULL;
    }
    zsock_t *sock = zsock_new (index);
    if (sock) {
#if (ZMQ_VERSION_MAJOR == 4)
        if (self->domain [selected_socket]) {
            // Apply authentication domain
            zsock_set_zap_domain (sock, self->domain [selected_socket]);
        }
        if (self->auth_type [selected_socket] == AUTH_PLAIN) {
            // Enable plain authentication
            zsock_set_plain_server (sock, 1);
        }
        else
        if (self->auth_type [selected_socket] == AUTH_CURVE) {
            // Apply certificate keys
            char *public_key = self->public_key [selected_socket];
            assert (public_key);
            char *secret_key = self->secret_key [selected_socket];
            assert (secret_key);
            zsock_set_curve_publickey (sock, public_key);
            zsock_set_curve_secretkey (sock, secret_key);

            // Enable curve authentication
            zsock_set_curve_server (sock, 1);
        }
        if (heartbeat->ivl != 0) {
            if (self->verbose) {
                zsys_info ("zproxy - setting heartbeat ivl=%d, timeout=%d, ttl=%d",
                    heartbeat->ivl, heartbeat->timeout, heartbeat->ttl);
            }
            zsock_set_heartbeat_ivl (sock, heartbeat->ivl);
            zsock_set_heartbeat_timeout (sock, heartbeat->timeout);
            zsock_set_heartbeat_ttl (sock, heartbeat->ttl);
        }
#endif
        if (zsock_attach (sock, endpoints, true)) {
            zsys_error ("zproxy: invalid endpoints '%s'", endpoints);
            zsock_destroy (&sock);
        }
    }
    return sock;
}

static const char *
s_self_selected_socket_name (proxy_socket selected_socket)
{
    switch (selected_socket) {
        case FRONTEND:
            return "FRONTEND";
        case BACKEND:
            return "BACKEND";
        default:
            return "UNDEFINED";
    }
}

static const char *
s_self_selected_socket_auth (int auth_type)
{
    switch (auth_type) {
        case AUTH_PLAIN:
            return "PLAIN";
        case AUTH_CURVE:
            return "CURVE";
        default:
            return "NONE";
    }
}

static proxy_socket
s_self_selected_socket (zmsg_t *request)
{
    char *socket_name = zmsg_popstr (request);
    assert (socket_name);

    proxy_socket socket = NONE;

    if (streq (socket_name, "FRONTEND"))
        socket = FRONTEND;
    else
    if (streq (socket_name, "BACKEND"))
        socket = BACKEND;
    else {
        zsys_error ("zproxy: invalid proxy socket selection: %s", socket_name);
        assert (false);
    }
    zstr_free (&socket_name);
    return socket;
}

static int
s_get_int_value (zmsg_t *request)
{
    char *value = zmsg_popstr (request);
    assert (value);
    char *endp = NULL;
    int int_value = strtol (value, &endp, 0);
    assert (endp != value); // no digits found
    zstr_free (&value);
    return int_value;
}

static void
s_parse_heartbeat_command (zmsg_t *request, zproxy_heartbeat_t *heartbeat) {
    char *setting = zmsg_popstr (request);
    while (setting != NULL) {
        if (streq (setting, "HEARTBEAT_IVL"))
            heartbeat->ivl =  s_get_int_value (request);
        else
        if (streq (setting, "HEARTBEAT_TIMEOUT"))
            heartbeat->timeout = s_get_int_value (request);
        else
        if (streq (setting, "HEARTBEAT_TTL"))
            heartbeat->ttl = s_get_int_value (request);
        zstr_free (&setting);
        setting = zmsg_popstr (request);
    }
}

static void
s_self_configure (self_t *self, zsock_t **sock_p, zmsg_t *request, proxy_socket selected_socket)
{
    char *type_name = zmsg_popstr (request);
    assert (type_name);
    char *endpoints = zmsg_popstr (request);
    assert (endpoints);

    zproxy_heartbeat_t heartbeat = { 0, }; // default value 0 for all members
    s_parse_heartbeat_command (request, &heartbeat);

    if (self->verbose)
        zsys_info ("zproxy: - %s type=%s attach=%s authentication=%s",
            s_self_selected_socket_name (selected_socket), type_name, endpoints,
            s_self_selected_socket_auth (self->auth_type [selected_socket]));
    assert (*sock_p == NULL);
    *sock_p = s_self_create_socket (self, type_name, endpoints, &heartbeat, selected_socket);
    assert (*sock_p);

#ifdef CZMQ_BUILD_DRAFT_API
    if (strcmp (type_name, "SUB") || strcmp (type_name, "XSUB")) {
        char *topic;
        while ((topic = zmsg_popstr (request)) != NULL) {
            zsock_set_subscribe (*sock_p, topic);
            zstr_free (&topic);
        }
    }
#endif // CZMQ_BUILD_DRAFT_API

    zstr_free (&type_name);
    zstr_free (&endpoints);
}

static void
s_self_add_to_poller_when_configured (self_t *self)
{
    if (self->frontend && self->backend) {
        zpoller_add(self->poller, self->frontend);
        zpoller_add(self->poller, self->backend);
    }
}

//  --------------------------------------------------------------------------
//  Handle a command from calling application

static int
s_self_handle_pipe (self_t *self)
{
    //  Get the whole message off the pipe in one go
    zmsg_t *request = zmsg_recv (self->pipe);
    if (!request)
        return -1;                  //  Interrupted

    char *command = zmsg_popstr (request);
    assert (command);
    if (self->verbose)
        zsys_info ("zproxy: API command=%s", command);

    if (streq (command, "FRONTEND")) {
        s_self_configure (self, &self->frontend, request, FRONTEND);
        s_self_add_to_poller_when_configured (self);
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "BACKEND")) {
        s_self_configure (self, &self->backend, request, BACKEND);
        s_self_add_to_poller_when_configured (self);
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "CAPTURE")) {
        self->capture = zsock_new (ZMQ_PUSH);
        assert (self->capture);
        char *endpoint = zmsg_popstr (request);
        assert (endpoint);
        int rc = zsock_connect (self->capture, "%s", endpoint);
        assert (rc == 0);
        zstr_free (&endpoint);
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "PAUSE")) {
        zpoller_destroy (&self->poller);
        self->poller = zpoller_new (self->pipe, NULL);
        assert (self->poller);
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "RESUME")) {
        zpoller_destroy (&self->poller);
        self->poller = zpoller_new (self->pipe, self->frontend, self->backend, NULL);
        assert (self->poller);
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "VERBOSE")) {
        self->verbose = true;
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "DOMAIN")) {
        proxy_socket selected_socket = s_self_selected_socket (request);
        char *domain = zmsg_popstr (request);
        assert (domain);
        zstr_free (&self->domain [selected_socket]);
        self->domain [selected_socket] = domain;
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "PLAIN")) {
        proxy_socket selected_socket = s_self_selected_socket (request);
        self->auth_type [selected_socket] = AUTH_PLAIN;
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "CURVE")) {
        proxy_socket selected_socket = s_self_selected_socket (request);
        self->auth_type [selected_socket] = AUTH_CURVE;
        char *public_key = zmsg_popstr (request);
        assert (public_key);
        char *secret_key = zmsg_popstr (request);
        assert (secret_key);
        zstr_free (&self->public_key [selected_socket]);
        zstr_free (&self->secret_key [selected_socket]);
        self->public_key [selected_socket] = public_key;
        self->secret_key [selected_socket] = secret_key;
        zsock_signal (self->pipe, 0);
    }
    else
    if (streq (command, "$TERM"))
        self->terminated = true;
    else {
        zsys_error ("zproxy: - invalid command: %s", command);
        assert (false);
    }
    zstr_free (&command);
    zmsg_destroy (&request);
    return 0;
}


//  --------------------------------------------------------------------------
//  Switch messages from an input socket to an output socket until there are
//  no messages left waiting. We use this loop rather than zmq_poll, to reduce
//  the cost of polling, which is non-trivial on some boxes (OS/X, mainly).

static void
s_self_switch (self_t *self, zsock_t *input, zsock_t *output)
{
    //  We use the low-level libzmq API for best performance
    void *zmq_input = zsock_resolve (input);
    void *zmq_output = zsock_resolve (output);
    void *zmq_capture = self->capture? zsock_resolve (self->capture): NULL;

    zmq_msg_t msg;
    zmq_msg_init (&msg);
    while (true) {
        if (zmq_recvmsg (zmq_input, &msg, ZMQ_DONTWAIT) == -1)
            break;      //  Presumably EAGAIN
        int send_flags = zsock_rcvmore (input)? ZMQ_SNDMORE: 0;
        if (zmq_capture) {
            zmq_msg_t dup;
            zmq_msg_init (&dup);
            zmq_msg_copy (&dup, &msg);
            if (zmq_sendmsg (zmq_capture, &dup, send_flags) == -1)
                zmq_msg_close (&dup);
        }
        if (zmq_sendmsg (zmq_output, &msg, send_flags) == -1) {
            zmq_msg_close (&msg);
            break;
        }
    }
}


//  --------------------------------------------------------------------------
//  zproxy() implements the zproxy actor interface

void
zproxy (zsock_t *pipe, void *unused)
{
    self_t *self = s_self_new (pipe);
    assert (self);
    //  Signal successful initialization
    zsock_signal (pipe, 0);

    while (!self->terminated) {
        zsock_t *which = (zsock_t *) zpoller_wait (self->poller, -1);
        if (zpoller_terminated (self->poller))
            break;          //  Interrupted
        else
        if (which == self->pipe)
            s_self_handle_pipe (self);
        else
        if (which == self->frontend)
            s_self_switch (self, self->frontend, self->backend);
        else
        if (which == self->backend)
            s_self_switch (self, self->backend, self->frontend);
    }
    s_self_destroy (&self);
}


//  --------------------------------------------------------------------------
//  Selftest

#define LOCALENDPOINT "tcp://127.0.0.1:%d"

static int
s_get_available_port (void)
{
    int port_nbr = -1;
    int attempts = 0;

    //  Choosing a random port for better results in case multiple tests are running
    //  in parallel on the same machine, such as during CI runs
    while (port_nbr == -1 && attempts++ < 10) {
        port_nbr = 49152 + randof (16383);
        zsock_t *server = zsock_new (ZMQ_PUSH);
        assert (server);
        port_nbr = zsock_bind (server, LOCALENDPOINT, port_nbr);
        zsock_destroy (&server);
    }
    if (port_nbr < 0) {
        zsys_error ("zproxy: failed to find an available port number");
        assert (false);
    }
    return port_nbr;
}

#if (ZMQ_VERSION_MAJOR == 4)

static void
s_create_test_sockets (zactor_t **proxy, zsock_t **faucet, zsock_t **sink, bool verbose)
{
    zsock_destroy (faucet);
    zsock_destroy (sink);
    zactor_destroy (proxy);
    *faucet = zsock_new (ZMQ_PUSH);
    assert (*faucet);
    *sink = zsock_new (ZMQ_PULL);
    assert (*sink);
    *proxy = zactor_new (zproxy, NULL);
    assert (*proxy);
    if (verbose) {
        zstr_sendx (*proxy, "VERBOSE", NULL);
        zsock_wait (*proxy);
    }
}

//  Checks whether client can connect to server
//  expect_success is used to synchronise the connect as sleeps are unreliable
static bool
s_can_connect (zactor_t **proxy, zsock_t **faucet, zsock_t **sink, const char *frontend, const char *backend, bool verbose, bool expect_success)
{
    assert (*faucet);
    assert (*sink);
    assert (frontend);
    assert (backend);

    int rc = zsock_connect (*faucet, "%s", frontend);
    assert (rc == 0);
    rc = zsock_connect (*sink, "%s", backend);
    assert (rc == 0);
    if (expect_success) {
        zstr_send (*faucet, "Hello, World");
        char *hello = zstr_recv (*sink);
        assert (hello);
        assert (streq (hello, "Hello, World"));
        zstr_free (&hello);
    }
    zframe_t *frame = zframe_from ("Hello, World");
    rc = zframe_send (&frame, *faucet, expect_success ? 0 : ZFRAME_DONTWAIT);
    assert (rc == 0 || !expect_success);
    if (!expect_success && rc == -1)
        zframe_destroy (&frame);

    zpoller_t *poller = zpoller_new (*sink, NULL);
    assert (poller);
    bool success = (zpoller_wait (poller, 400) == *sink);
    zpoller_destroy (&poller);
    s_create_test_sockets (proxy, faucet, sink, verbose);

    return success;
}

static void
s_create_endpoints (zactor_t *proxy, char **frontend, char **backend)
{
    zstr_free (frontend);
    zstr_free (backend);
    assert (proxy);

    srandom ((uint) (time (NULL) ^ *(int *) proxy));
    *frontend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());
    *backend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());
}

static void
s_bind_test_sockets (zactor_t *proxy, char **frontend, char **backend)
{
    s_create_endpoints (proxy, frontend, backend);

    //  Wait a moment to make sure ports have time to unbind...
    zclock_sleep(200);
    zstr_sendx (proxy, "FRONTEND", "PULL", *frontend, NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "BACKEND", "PUSH", *backend, NULL);
    zsock_wait (proxy);
}
#endif

void
zproxy_test (bool verbose)
{
    printf (" * zproxy: ");
    proxy = zactor_new (zproxy, NULL);
    assert (proxy);

    sink = zsock_new_sub (">ipc://backend", "whatever");
    assert (sink);

    zstr_sendx (proxy, "BACKEND", "XPUB", "ipc://backend", NULL);
    zsock_wait (proxy);

    zsock_destroy(&sink);
    assert (faucet);
    zsock_t *sink = zsock_new_pull (">inproc://backend");
    assert (sink);

    //  Send some messages and check they arrived
    char *hello, *world;
    zstr_sendx (faucet, "Hello", "World", NULL);
    zstr_recvx (sink, &hello, &world, NULL);
    assert (streq (hello, "Hello"));
    assert (streq (world, "World"));
    zstr_free (&hello);
    zstr_free (&world);

    //  Test pause/resume functionality
    zstr_sendx (proxy, "PAUSE", NULL);
    zsock_wait (proxy);
    zstr_sendx (faucet, "Hello", "World", NULL);
    zsock_set_rcvtimeo (sink, 100);
    zstr_recvx (sink, &hello, &world, NULL);
    assert (!hello && !world);

    zstr_sendx (proxy, "RESUME", NULL);
    zsock_wait (proxy);
    zstr_recvx (sink, &hello, &world, NULL);
    assert (streq (hello, "Hello"));
    assert (streq (world, "World"));
    zstr_free (&hello);
    zstr_free (&world);

    //  Test capture functionality
    zsock_t *capture = zsock_new_pull ("inproc://capture");
    assert (capture);

    //  Switch on capturing, check that it works
    zstr_sendx (proxy, "CAPTURE", "inproc://capture", NULL);
    zsock_wait (proxy);
    zstr_sendx (faucet, "Hello", "World", NULL);
    zstr_recvx (sink, &hello, &world, NULL);
    assert (streq (hello, "Hello"));
    assert (streq (world, "World"));
    zstr_free (&hello);
    zstr_free (&world);

    zstr_recvx (capture, &hello, &world, NULL);
    assert (streq (hello, "Hello"));
    assert (streq (world, "World"));
    zstr_free (&hello);
    zstr_free (&world);

    zsock_destroy (&faucet);
    zsock_destroy (&sink);
    zsock_destroy (&capture);
    zactor_destroy (&proxy);

    //  Test socket creation dependency
    proxy = zactor_new (zproxy, NULL);
    assert (proxy);

    char *frontend = NULL;
    char *backend = NULL;
    backend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());
    zclock_sleep (200);
    sink = zsock_new_sub (backend, "whatever");
    assert (sink);

    zstr_sendx (proxy, "BACKEND", "XPUB", backend, NULL);
    zsock_wait (proxy);

    zsock_destroy(&sink);
    zactor_destroy(&proxy);
    zstr_free (&backend);

#ifdef CZMQ_BUILD_DRAFT_API
    //  Create and configure our proxy with PUB/SUB to test subscriptions
    proxy = zactor_new (zproxy, NULL);
    assert (proxy);
    if (verbose) {
        zstr_sendx (proxy, "VERBOSE", NULL);
        zsock_wait (proxy);
    }
    zstr_sendx (proxy, "FRONTEND", "SUB", "inproc://frontend", "He", "b", NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "BACKEND", "PUB", "inproc://backend", NULL);
    zsock_wait (proxy);

    //  Connect application sockets to proxy
    faucet = zsock_new_pub (">inproc://frontend");
    assert (faucet);
    sink = zsock_new_sub (">inproc://backend", "");
    assert (sink);

    //  Send some messages and check they arrived
    zstr_sendx (faucet, "Hello", "World", NULL);
    // since SUB is binding, subscription might be lost see:
    // https://github.com/zeromq/libzmq/issues/2267
    zsock_set_rcvtimeo (sink, 100);
    hello = zstr_recv (sink);
    if (hello) {
        assert (streq (hello, "Hello"));
        world = zstr_recv (sink);
        assert (streq (world, "World"));
        zstr_free (&hello);
        zstr_free (&world);
    }

    zsock_destroy (&faucet);
    zsock_destroy (&sink);
    zactor_destroy(&proxy);
#endif // CZMQ_BUILD_DRAFT_API

#if (ZMQ_VERSION_MAJOR == 4)
    // Test authentication functionality
    const char *basedirpath = "src/selftest-rw/.test_zproxy";
    const char *passfilepath = "src/selftest-rw/.test_zproxy/password-file";
    const char *certfilepath = "src/selftest-rw/.test_zproxy/mycert.txt";

    // Make sure old aborted tests do not hinder us
    zdir_t *dir = zdir_new (basedirpath, NULL);
    if (dir) {
        zdir_remove (dir, true);
        zdir_destroy (&dir);
    }
    zsys_file_delete (passfilepath);
    zsys_file_delete (certfilepath);
    zsys_dir_delete (basedirpath);

    //  Create temporary directory for test files
    zsys_dir_create (basedirpath);

    //  Check there's no authentication
    s_create_test_sockets (&proxy, &faucet, &sink, verbose);
    s_bind_test_sockets (proxy, &frontend, &backend);
    bool success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,
        verbose, true);
    assert (success);

    //  Install the authenticator
    zactor_t *auth = zactor_new (zauth, NULL);
    assert (auth);
    if (verbose) {
        zstr_sendx (auth, "VERBOSE", NULL);
        zsock_wait (auth);
    }

    //  Check there's no authentication on a default NULL server
    s_bind_test_sockets (proxy, &frontend, &backend);
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        true);
    assert (success);

    //  When we set a domain on the server, we switch on authentication
    //  for NULL sockets, but with no policies, the client connection
    //  will be allowed.
    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);
    zsock_wait (proxy);
    s_bind_test_sockets (proxy, &frontend, &backend);
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        true);
    assert (success);

    //  Block 127.0.0.1, connection should fail
    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);
    zsock_wait (proxy);
    s_bind_test_sockets (proxy, &frontend, &backend);
    zstr_sendx (auth, "DENY", "127.0.0.1", NULL);
    zsock_wait (auth);
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        false);
    assert (!success);

    //  Allow our address, which overrides the block list
    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "DOMAIN", "BACKEND", "global", NULL);
    zsock_wait (proxy);
    s_bind_test_sockets (proxy, &frontend, &backend);
    zstr_sendx (auth, "ALLOW", "127.0.0.1", NULL);
    zsock_wait (auth);
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        true);
    assert (success);

    //  Try PLAIN authentication

    //  Test negative case (no server-side passwords defined)
    zstr_sendx (proxy, "PLAIN", "FRONTEND", NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);
    zsock_wait (proxy);
    s_bind_test_sockets (proxy, &frontend, &backend);
    zsock_set_plain_username (faucet, "admin");
    zsock_set_plain_password (faucet, "Password");
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        false);
    assert (!success);

    //  Test positive case (server-side passwords defined)
    FILE *password = fopen (passfilepath, "w");
    assert (password);
    fprintf (password, "admin=Password\n");
    fclose (password);
    zstr_sendx (proxy, "PLAIN", "FRONTEND", NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "PLAIN", "BACKEND", NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "DOMAIN", "BACKEND", "global", NULL);
    zsock_wait (proxy);
    s_bind_test_sockets (proxy, &frontend, &backend);
    zsock_set_plain_username (faucet, "admin");
    zsock_set_plain_password (faucet, "Password");
    zsock_set_plain_username (sink, "admin");
    zsock_set_plain_password (sink, "Password");
    zstr_sendx (auth, "PLAIN", passfilepath, NULL);
    zsock_wait (auth);
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        true);
    assert (success);

    //  Test negative case (bad client password)
    zstr_sendx (proxy, "PLAIN", "FRONTEND", NULL);
    zsock_wait (proxy);
    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);
    zsock_wait (proxy);
    s_bind_test_sockets (proxy, &frontend, &backend);
    zsock_set_plain_username (faucet, "admin");
    zsock_set_plain_password (faucet, "Bogus");
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        false);
    assert (!success);

    if (zsys_has_curve ()) {
        //  We'll create two new certificates and save the client public
        //  certificate on disk
        zcert_t *server_cert = zcert_new ();
        assert (server_cert);
        zcert_t *client_cert = zcert_new ();
        assert (client_cert);
        const char *public_key = zcert_public_txt (server_cert);
        const char *secret_key = zcert_secret_txt (server_cert);

        //  Try CURVE authentication

        //  Test without setting-up any authentication
        zstr_sendx (proxy, "CURVE", "FRONTEND", public_key, secret_key, NULL);
        zsock_wait (proxy);
        zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);
        zsock_wait (proxy);
        s_bind_test_sockets (proxy, &frontend, &backend);
        zcert_apply (client_cert, faucet);
        zsock_set_curve_serverkey (faucet, public_key);
        success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,
            verbose, false);
        assert (!success);

        //  Test CURVE_ALLOW_ANY
        zstr_sendx (proxy, "CURVE", "FRONTEND", public_key, secret_key, NULL);
        zsock_wait (proxy);
        s_bind_test_sockets (proxy, &frontend, &backend);
        zcert_apply (client_cert, faucet);
        zsock_set_curve_serverkey (faucet, public_key);
        zstr_sendx (auth, "CURVE", CURVE_ALLOW_ANY, NULL);
        zsock_wait (auth);
        success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,
            verbose, true);
        assert (success);

        //  Test with client certificate file in authentication folder
        zstr_sendx (proxy, "CURVE", "FRONTEND", public_key, secret_key, NULL);
        zsock_wait (proxy);
        zstr_sendx (proxy, "CURVE", "BACKEND", public_key, secret_key, NULL);
        zsock_wait (proxy);
        s_bind_test_sockets (proxy, &frontend, &backend);
        zcert_apply (client_cert, faucet);
        zsock_set_curve_serverkey (faucet, public_key);
        zcert_apply (client_cert, sink);
        zsock_set_curve_serverkey (sink, public_key);
        zcert_save_public (client_cert, certfilepath);
        zstr_sendx (auth, "CURVE", basedirpath, NULL);
        zsock_wait (auth);
        success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,
            verbose, true);
        assert (success);

        zcert_destroy (&server_cert);
        zcert_destroy (&client_cert);
    }

    //  Remove the authenticator and check a normal connection works
    zactor_destroy (&auth);
    s_bind_test_sockets (proxy, &frontend, &backend);
    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,
        true);
    assert (success);

    // Test heartbeat functionality

    s_create_test_sockets (&proxy, &faucet, &sink, verbose);
    s_create_endpoints (proxy, &frontend, &backend);
    zclock_sleep (200);

    // Set the heartbeat interval to much longer than the TTL so that
    // the socket times out on the remote side
    zstr_sendx (proxy, "FRONTEND", "PULL", frontend,
        "HEARTBEAT_IVL", "300", "HEARTBEAT_TTL", "100", NULL);
    zsock_wait (proxy);

    zstr_sendx (proxy, "BACKEND", "PUSH", backend, NULL);
    zsock_wait (proxy);

    // Setup a monitor to make sure we disconnect
    zactor_t *monitor = zactor_new (zmonitor, faucet);
    assert (monitor);
    if (verbose)
        zstr_sendx (monitor, "VERBOSE", NULL);
    zstr_sendx (monitor, "LISTEN", "CONNECTED", "DISCONNECTED",  NULL);
    zstr_sendx (monitor, "START", NULL);
    zsock_wait (monitor);

    // Connect the faucet
    int rc = zsock_connect (faucet, "%s", frontend);
    assert (rc == 0);

    zmsg_t *msg = zmsg_recv (monitor);
    assert (msg);
    char *event = zmsg_popstr (msg);
    assert (streq (event, "CONNECTED"));
    zmsg_destroy (&msg);
    zstr_free (&event);

    // time to settle ~300ms
    zclock_sleep (300);

    // Now we should have been disconnected
    msg = zmsg_recv (monitor);
    assert (msg);
    event = zmsg_popstr (msg);
    assert (streq (event, "DISCONNECTED"));
    zmsg_destroy (&msg);
    zstr_free (&event);

    //  Cleanup
    zactor_destroy (&monitor);
    zsock_destroy (&faucet);
    zsock_destroy (&sink);
    zactor_destroy (&proxy);
    zstr_free (&frontend);
    zstr_free (&backend);

    //  Delete temporary directory and test files
    zsys_file_delete (passfilepath);
    zsys_file_delete (certfilepath);
    zsys_dir_delete (basedirpath);
#endif

#if defined (__WINDOWS__)
    zsys_shutdown();
#endif
    //  @end
    printf ("OK\n");
}

<MSG> problem : zproxy self test fails under windows

solution : replace ipc with inproc (only for windows)

<DFF> @@ -547,10 +547,18 @@ zproxy_test (bool verbose)
     proxy = zactor_new (zproxy, NULL);
     assert (proxy);
 
-    sink = zsock_new_sub (">ipc://backend", "whatever");
-    assert (sink);
-
-    zstr_sendx (proxy, "BACKEND", "XPUB", "ipc://backend", NULL);
+#ifdef  WIN32
+	sink = zsock_new_sub(">inproc://backend", "whatever");
+#else
+	sink = zsock_new_sub (">ipc://backend", "whatever");
+#endif //  WIN32
+	assert (sink);
+
+#ifdef WIN32
+	zstr_sendx (proxy, "BACKEND", "XPUB", "inproc://backend", NULL);
+#else
+	zstr_sendx(proxy, "BACKEND", "XPUB", "ipc://backend", NULL);
+#endif
     zsock_wait (proxy);
 
     zsock_destroy(&sink);

{'commit': 'diff --git a/src/zproxy.c b/src/zproxy.c\nindex 00a7363c..29bd5899 100644\n--- a/src/zproxy.c\n+++ b/src/zproxy.c\n@@ -547,10 +547,18 @@ zproxy_test (bool verbose)\n     proxy = zactor_new (zproxy, NULL);\n     assert (proxy);\n \n-    sink = zsock_new_sub (">ipc://backend", "whatever");\n-    assert (sink);\n-\n-    zstr_sendx (proxy, "BACKEND", "XPUB", "ipc://backend", NULL);\n+#ifdef  WIN32\r\n+\tsink = zsock_new_sub(">inproc://backend", "whatever");\n+#else\r\n+\tsink = zsock_new_sub (">ipc://backend", "whatever");\r\n+#endif //  WIN32\r\n+\tassert (sink);\n+\n+#ifdef WIN32\n+\tzstr_sendx (proxy, "BACKEND", "XPUB", "inproc://backend", NULL);\n+#else\n+\tzstr_sendx(proxy, "BACKEND", "XPUB", "ipc://backend", NULL);\n+#endif\n     zsock_wait (proxy);\n \n     zsock_destroy(&sink);\n', 'commit_subject': 'problem : zproxy self test fails under windows', 'repo_name': 'zeromq/czmq', 'committer': {'name': 'Bruno Bodin', 'email': '0317d452586bff9a8d49cafbf7db3e3b7023a08c@gmail.com', 'time_sec': '1484597851', 'tz_offset': '60', 'date': {'seconds': '1484597851'}}, 'commit_tree': 'e9e003365d0e6644abebb1d160c4802e959ed112', 'commit_message': 'problem : zproxy self test fails under windows\n\nsolution : replace ipc with inproc (only for windows)\n', 'language_name': 'Ruby', 'license': 'mpl-2.0', 'hunks': '@@ -547,10 +547,18 @@ zproxy_test (bool verbose)\n     proxy = zactor_new (zproxy, NULL);\n     assert (proxy);\n \n-    sink = zsock_new_sub (">ipc://backend", "whatever");\n-    assert (sink);\n-\n-    zstr_sendx (proxy, "BACKEND", "XPUB", "ipc://backend", NULL);\n+#ifdef  WIN32\r\n+\tsink = zsock_new_sub(">inproc://backend", "whatever");\n+#else\r\n+\tsink = zsock_new_sub (">ipc://backend", "whatever");\r\n+#endif //  WIN32\r\n+\tassert (sink);\n+\n+#ifdef WIN32\n+\tzstr_sendx (proxy, "BACKEND", "XPUB", "inproc://backend", NULL);\n+#else\n+\tzstr_sendx(proxy, "BACKEND", "XPUB", "ipc://backend", NULL);\n+#endif\n     zsock_wait (proxy);\n \n     zsock_destroy(&sink);\n', 'addition_count': 12, 'deletion_count': 4, 'src_file': 'a/src/zproxy.c', 'tgt_file': 'b/src/zproxy.c', 'file_extension': '.c', 'before_file': '/*  =========================================================================\n    zproxy - run a steerable proxy in the background\n\n    Copyright (c) the Contributors as noted in the AUTHORS file.\n    This file is part of CZMQ, the high-level C binding for 0MQ:\n    http://czmq.zeromq.org.\n\n    This Source Code Form is subject to the terms of the Mozilla Public\n    License, v. 2.0. If a copy of the MPL was not distributed with this\n    file, You can obtain one at http://mozilla.org/MPL/2.0/.\n    =========================================================================\n*/\n\n/*\n@header\n    A zproxy actor switches messages between a frontend and a backend socket.\n    It acts much like the zmq_proxy_steerable method, though it makes benefit\n    of CZMQ\'s facilities, to be somewhat simpler to set-up.\n@discuss\n    This class replaces zproxy_v2, and is meant for applications that use the\n    CZMQ v3 API (meaning, zsock).\n@end\n*/\n\n#include "czmq_classes.h"\n\ntypedef enum proxy_socket {\n    NONE = -1,\n    FRONTEND,\n    BACKEND,\n    SOCKETS\n}\nproxy_socket;\n\ntypedef struct {\n    int ivl;\n    int timeout;\n    int ttl;\n} zproxy_heartbeat_t;\n\n#define AUTH_NONE  0\n#define AUTH_PLAIN 1\n#define AUTH_CURVE 2\n\n//  --------------------------------------------------------------------------\n//  The self_t structure holds the state for one actor instance\n\ntypedef struct {\n    zsock_t *pipe;              //  Actor command pipe\n    zpoller_t *poller;          //  Socket poller\n    zsock_t *frontend;          //  Frontend socket\n    zsock_t *backend;           //  Backend socket\n    zsock_t *capture;           //  Capture socket\n    int auth_type [SOCKETS];    //  Auth type for sockets\n    char *domain [SOCKETS];     //  Auth domains for sockets\n    char *public_key [SOCKETS]; //  Public keys for sockets\n    char *secret_key [SOCKETS]; //  Secret keys for sockets\n    bool terminated;            //  Did caller ask us to quit?\n    bool verbose;               //  Verbose logging enabled?\n} self_t;\n\n\nstatic self_t *\ns_self_new (zsock_t *pipe)\n{\n    self_t *self = (self_t *) zmalloc (sizeof (self_t));\n    assert (self);\n    self->pipe = pipe;\n    self->poller = zpoller_new (self->pipe, NULL);\n    assert (self->poller);\n    return self;\n}\n\nstatic void\ns_self_destroy (self_t **self_p)\n{\n    assert (self_p);\n    if (*self_p) {\n        self_t *self = *self_p;\n        zpoller_destroy (&self->poller);\n        zsock_destroy (&self->frontend);\n        zsock_destroy (&self->backend);\n        zsock_destroy (&self->capture);\n        int index;\n        for (index = 0; index < SOCKETS; index++) {\n            zstr_free (&self->domain [index]);\n            zstr_free (&self->public_key [index]);\n            zstr_free (&self->secret_key [index]);\n        }\n        freen (self);\n        *self_p = NULL;\n    }\n}\n\nstatic zsock_t *\ns_self_create_socket (self_t *self, char *type_name, char *endpoints, zproxy_heartbeat_t *heartbeat, proxy_socket selected_socket)\n{\n    //  This array matches ZMQ_XXX type definitions\n    assert (ZMQ_PAIR == 0);\n    char *type_names [] = {\n        "PAIR", "PUB", "SUB", "REQ", "REP",\n        "DEALER", "ROUTER", "PULL", "PUSH",\n        "XPUB", "XSUB", type_name\n    };\n    //  We always match type at least at end of table\n    int index;\n    for (index = 0; strneq (type_name, type_names [index]); index++) ;\n    if (index > ZMQ_XSUB) {\n        zsys_error ("zproxy: invalid socket type \'%s\'", type_name);\n        return NULL;\n    }\n    zsock_t *sock = zsock_new (index);\n    if (sock) {\n#if (ZMQ_VERSION_MAJOR == 4)\n        if (self->domain [selected_socket]) {\n            // Apply authentication domain\n            zsock_set_zap_domain (sock, self->domain [selected_socket]);\n        }\n        if (self->auth_type [selected_socket] == AUTH_PLAIN) {\n            // Enable plain authentication\n            zsock_set_plain_server (sock, 1);\n        }\n        else\n        if (self->auth_type [selected_socket] == AUTH_CURVE) {\n            // Apply certificate keys\n            char *public_key = self->public_key [selected_socket];\n            assert (public_key);\n            char *secret_key = self->secret_key [selected_socket];\n            assert (secret_key);\n            zsock_set_curve_publickey (sock, public_key);\n            zsock_set_curve_secretkey (sock, secret_key);\n\n            // Enable curve authentication\n            zsock_set_curve_server (sock, 1);\n        }\n        if (heartbeat->ivl != 0) {\n            if (self->verbose) {\n                zsys_info ("zproxy - setting heartbeat ivl=%d, timeout=%d, ttl=%d",\n                    heartbeat->ivl, heartbeat->timeout, heartbeat->ttl);\n            }\n            zsock_set_heartbeat_ivl (sock, heartbeat->ivl);\n            zsock_set_heartbeat_timeout (sock, heartbeat->timeout);\n            zsock_set_heartbeat_ttl (sock, heartbeat->ttl);\n        }\n#endif\n        if (zsock_attach (sock, endpoints, true)) {\n            zsys_error ("zproxy: invalid endpoints \'%s\'", endpoints);\n            zsock_destroy (&sock);\n        }\n    }\n    return sock;\n}\n\nstatic const char *\ns_self_selected_socket_name (proxy_socket selected_socket)\n{\n    switch (selected_socket) {\n        case FRONTEND:\n            return "FRONTEND";\n        case BACKEND:\n            return "BACKEND";\n        default:\n            return "UNDEFINED";\n    }\n}\n\nstatic const char *\ns_self_selected_socket_auth (int auth_type)\n{\n    switch (auth_type) {\n        case AUTH_PLAIN:\n            return "PLAIN";\n        case AUTH_CURVE:\n            return "CURVE";\n        default:\n            return "NONE";\n    }\n}\n\nstatic proxy_socket\ns_self_selected_socket (zmsg_t *request)\n{\n    char *socket_name = zmsg_popstr (request);\n    assert (socket_name);\n\n    proxy_socket socket = NONE;\n\n    if (streq (socket_name, "FRONTEND"))\n        socket = FRONTEND;\n    else\n    if (streq (socket_name, "BACKEND"))\n        socket = BACKEND;\n    else {\n        zsys_error ("zproxy: invalid proxy socket selection: %s", socket_name);\n        assert (false);\n    }\n    zstr_free (&socket_name);\n    return socket;\n}\n\nstatic int\ns_get_int_value (zmsg_t *request)\n{\n    char *value = zmsg_popstr (request);\n    assert (value);\n    char *endp = NULL;\n    int int_value = strtol (value, &endp, 0);\n    assert (endp != value); // no digits found\n    zstr_free (&value);\n    return int_value;\n}\n\nstatic void\ns_parse_heartbeat_command (zmsg_t *request, zproxy_heartbeat_t *heartbeat) {\n    char *setting = zmsg_popstr (request);\n    while (setting != NULL) {\n        if (streq (setting, "HEARTBEAT_IVL"))\n            heartbeat->ivl =  s_get_int_value (request);\n        else\n        if (streq (setting, "HEARTBEAT_TIMEOUT"))\n            heartbeat->timeout = s_get_int_value (request);\n        else\n        if (streq (setting, "HEARTBEAT_TTL"))\n            heartbeat->ttl = s_get_int_value (request);\n        zstr_free (&setting);\n        setting = zmsg_popstr (request);\n    }\n}\n\nstatic void\ns_self_configure (self_t *self, zsock_t **sock_p, zmsg_t *request, proxy_socket selected_socket)\n{\n    char *type_name = zmsg_popstr (request);\n    assert (type_name);\n    char *endpoints = zmsg_popstr (request);\n    assert (endpoints);\n\n    zproxy_heartbeat_t heartbeat = { 0, }; // default value 0 for all members\n    s_parse_heartbeat_command (request, &heartbeat);\n\n    if (self->verbose)\n        zsys_info ("zproxy: - %s type=%s attach=%s authentication=%s",\n            s_self_selected_socket_name (selected_socket), type_name, endpoints,\n            s_self_selected_socket_auth (self->auth_type [selected_socket]));\n    assert (*sock_p == NULL);\n    *sock_p = s_self_create_socket (self, type_name, endpoints, &heartbeat, selected_socket);\n    assert (*sock_p);\n\n#ifdef CZMQ_BUILD_DRAFT_API\n    if (strcmp (type_name, "SUB") || strcmp (type_name, "XSUB")) {\n        char *topic;\n        while ((topic = zmsg_popstr (request)) != NULL) {\n            zsock_set_subscribe (*sock_p, topic);\n            zstr_free (&topic);\n        }\n    }\n#endif // CZMQ_BUILD_DRAFT_API\n\n    zstr_free (&type_name);\n    zstr_free (&endpoints);\n}\n\nstatic void\ns_self_add_to_poller_when_configured (self_t *self)\n{\n    if (self->frontend && self->backend) {\n        zpoller_add(self->poller, self->frontend);\n        zpoller_add(self->poller, self->backend);\n    }\n}\n\n//  --------------------------------------------------------------------------\n//  Handle a command from calling application\n\nstatic int\ns_self_handle_pipe (self_t *self)\n{\n    //  Get the whole message off the pipe in one go\n    zmsg_t *request = zmsg_recv (self->pipe);\n    if (!request)\n        return -1;                  //  Interrupted\n\n    char *command = zmsg_popstr (request);\n    assert (command);\n    if (self->verbose)\n        zsys_info ("zproxy: API command=%s", command);\n\n    if (streq (command, "FRONTEND")) {\n        s_self_configure (self, &self->frontend, request, FRONTEND);\n        s_self_add_to_poller_when_configured (self);\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "BACKEND")) {\n        s_self_configure (self, &self->backend, request, BACKEND);\n        s_self_add_to_poller_when_configured (self);\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "CAPTURE")) {\n        self->capture = zsock_new (ZMQ_PUSH);\n        assert (self->capture);\n        char *endpoint = zmsg_popstr (request);\n        assert (endpoint);\n        int rc = zsock_connect (self->capture, "%s", endpoint);\n        assert (rc == 0);\n        zstr_free (&endpoint);\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "PAUSE")) {\n        zpoller_destroy (&self->poller);\n        self->poller = zpoller_new (self->pipe, NULL);\n        assert (self->poller);\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "RESUME")) {\n        zpoller_destroy (&self->poller);\n        self->poller = zpoller_new (self->pipe, self->frontend, self->backend, NULL);\n        assert (self->poller);\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "VERBOSE")) {\n        self->verbose = true;\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "DOMAIN")) {\n        proxy_socket selected_socket = s_self_selected_socket (request);\n        char *domain = zmsg_popstr (request);\n        assert (domain);\n        zstr_free (&self->domain [selected_socket]);\n        self->domain [selected_socket] = domain;\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "PLAIN")) {\n        proxy_socket selected_socket = s_self_selected_socket (request);\n        self->auth_type [selected_socket] = AUTH_PLAIN;\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "CURVE")) {\n        proxy_socket selected_socket = s_self_selected_socket (request);\n        self->auth_type [selected_socket] = AUTH_CURVE;\n        char *public_key = zmsg_popstr (request);\n        assert (public_key);\n        char *secret_key = zmsg_popstr (request);\n        assert (secret_key);\n        zstr_free (&self->public_key [selected_socket]);\n        zstr_free (&self->secret_key [selected_socket]);\n        self->public_key [selected_socket] = public_key;\n        self->secret_key [selected_socket] = secret_key;\n        zsock_signal (self->pipe, 0);\n    }\n    else\n    if (streq (command, "$TERM"))\n        self->terminated = true;\n    else {\n        zsys_error ("zproxy: - invalid command: %s", command);\n        assert (false);\n    }\n    zstr_free (&command);\n    zmsg_destroy (&request);\n    return 0;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Switch messages from an input socket to an output socket until there are\n//  no messages left waiting. We use this loop rather than zmq_poll, to reduce\n//  the cost of polling, which is non-trivial on some boxes (OS/X, mainly).\n\nstatic void\ns_self_switch (self_t *self, zsock_t *input, zsock_t *output)\n{\n    //  We use the low-level libzmq API for best performance\n    void *zmq_input = zsock_resolve (input);\n    void *zmq_output = zsock_resolve (output);\n    void *zmq_capture = self->capture? zsock_resolve (self->capture): NULL;\n\n    zmq_msg_t msg;\n    zmq_msg_init (&msg);\n    while (true) {\n        if (zmq_recvmsg (zmq_input, &msg, ZMQ_DONTWAIT) == -1)\n            break;      //  Presumably EAGAIN\n        int send_flags = zsock_rcvmore (input)? ZMQ_SNDMORE: 0;\n        if (zmq_capture) {\n            zmq_msg_t dup;\n            zmq_msg_init (&dup);\n            zmq_msg_copy (&dup, &msg);\n            if (zmq_sendmsg (zmq_capture, &dup, send_flags) == -1)\n                zmq_msg_close (&dup);\n        }\n        if (zmq_sendmsg (zmq_output, &msg, send_flags) == -1) {\n            zmq_msg_close (&msg);\n            break;\n        }\n    }\n}\n\n\n//  --------------------------------------------------------------------------\n//  zproxy() implements the zproxy actor interface\n\nvoid\nzproxy (zsock_t *pipe, void *unused)\n{\n    self_t *self = s_self_new (pipe);\n    assert (self);\n    //  Signal successful initialization\n    zsock_signal (pipe, 0);\n\n    while (!self->terminated) {\n        zsock_t *which = (zsock_t *) zpoller_wait (self->poller, -1);\n        if (zpoller_terminated (self->poller))\n            break;          //  Interrupted\n        else\n        if (which == self->pipe)\n            s_self_handle_pipe (self);\n        else\n        if (which == self->frontend)\n            s_self_switch (self, self->frontend, self->backend);\n        else\n        if (which == self->backend)\n            s_self_switch (self, self->backend, self->frontend);\n    }\n    s_self_destroy (&self);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Selftest\n\n#define LOCALENDPOINT "tcp://127.0.0.1:%d"\n\nstatic int\ns_get_available_port (void)\n{\n    int port_nbr = -1;\n    int attempts = 0;\n\n    //  Choosing a random port for better results in case multiple tests are running\n    //  in parallel on the same machine, such as during CI runs\n    while (port_nbr == -1 && attempts++ < 10) {\n        port_nbr = 49152 + randof (16383);\n        zsock_t *server = zsock_new (ZMQ_PUSH);\n        assert (server);\n        port_nbr = zsock_bind (server, LOCALENDPOINT, port_nbr);\n        zsock_destroy (&server);\n    }\n    if (port_nbr < 0) {\n        zsys_error ("zproxy: failed to find an available port number");\n        assert (false);\n    }\n    return port_nbr;\n}\n\n#if (ZMQ_VERSION_MAJOR == 4)\n\nstatic void\ns_create_test_sockets (zactor_t **proxy, zsock_t **faucet, zsock_t **sink, bool verbose)\n{\n    zsock_destroy (faucet);\n    zsock_destroy (sink);\n    zactor_destroy (proxy);\n    *faucet = zsock_new (ZMQ_PUSH);\n    assert (*faucet);\n    *sink = zsock_new (ZMQ_PULL);\n    assert (*sink);\n    *proxy = zactor_new (zproxy, NULL);\n    assert (*proxy);\n    if (verbose) {\n        zstr_sendx (*proxy, "VERBOSE", NULL);\n        zsock_wait (*proxy);\n    }\n}\n\n//  Checks whether client can connect to server\n//  expect_success is used to synchronise the connect as sleeps are unreliable\nstatic bool\ns_can_connect (zactor_t **proxy, zsock_t **faucet, zsock_t **sink, const char *frontend, const char *backend, bool verbose, bool expect_success)\n{\n    assert (*faucet);\n    assert (*sink);\n    assert (frontend);\n    assert (backend);\n\n    int rc = zsock_connect (*faucet, "%s", frontend);\n    assert (rc == 0);\n    rc = zsock_connect (*sink, "%s", backend);\n    assert (rc == 0);\n    if (expect_success) {\n        zstr_send (*faucet, "Hello, World");\n        char *hello = zstr_recv (*sink);\n        assert (hello);\n        assert (streq (hello, "Hello, World"));\n        zstr_free (&hello);\n    }\n    zframe_t *frame = zframe_from ("Hello, World");\n    rc = zframe_send (&frame, *faucet, expect_success ? 0 : ZFRAME_DONTWAIT);\n    assert (rc == 0 || !expect_success);\n    if (!expect_success && rc == -1)\n        zframe_destroy (&frame);\n\n    zpoller_t *poller = zpoller_new (*sink, NULL);\n    assert (poller);\n    bool success = (zpoller_wait (poller, 400) == *sink);\n    zpoller_destroy (&poller);\n    s_create_test_sockets (proxy, faucet, sink, verbose);\n\n    return success;\n}\n\nstatic void\ns_create_endpoints (zactor_t *proxy, char **frontend, char **backend)\n{\n    zstr_free (frontend);\n    zstr_free (backend);\n    assert (proxy);\n\n    srandom ((uint) (time (NULL) ^ *(int *) proxy));\n    *frontend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());\n    *backend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());\n}\n\nstatic void\ns_bind_test_sockets (zactor_t *proxy, char **frontend, char **backend)\n{\n    s_create_endpoints (proxy, frontend, backend);\n\n    //  Wait a moment to make sure ports have time to unbind...\n    zclock_sleep(200);\n    zstr_sendx (proxy, "FRONTEND", "PULL", *frontend, NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "BACKEND", "PUSH", *backend, NULL);\n    zsock_wait (proxy);\n}\n#endif\n\nvoid\nzproxy_test (bool verbose)\n{\n    printf (" * zproxy: ");\n    proxy = zactor_new (zproxy, NULL);\n    assert (proxy);\n\n    sink = zsock_new_sub (">ipc://backend", "whatever");\n    assert (sink);\n\n    zstr_sendx (proxy, "BACKEND", "XPUB", "ipc://backend", NULL);\n    zsock_wait (proxy);\n\n    zsock_destroy(&sink);\n    assert (faucet);\n    zsock_t *sink = zsock_new_pull (">inproc://backend");\n    assert (sink);\n\n    //  Send some messages and check they arrived\n    char *hello, *world;\n    zstr_sendx (faucet, "Hello", "World", NULL);\n    zstr_recvx (sink, &hello, &world, NULL);\n    assert (streq (hello, "Hello"));\n    assert (streq (world, "World"));\n    zstr_free (&hello);\n    zstr_free (&world);\n\n    //  Test pause/resume functionality\n    zstr_sendx (proxy, "PAUSE", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (faucet, "Hello", "World", NULL);\n    zsock_set_rcvtimeo (sink, 100);\n    zstr_recvx (sink, &hello, &world, NULL);\n    assert (!hello && !world);\n\n    zstr_sendx (proxy, "RESUME", NULL);\n    zsock_wait (proxy);\n    zstr_recvx (sink, &hello, &world, NULL);\n    assert (streq (hello, "Hello"));\n    assert (streq (world, "World"));\n    zstr_free (&hello);\n    zstr_free (&world);\n\n    //  Test capture functionality\n    zsock_t *capture = zsock_new_pull ("inproc://capture");\n    assert (capture);\n\n    //  Switch on capturing, check that it works\n    zstr_sendx (proxy, "CAPTURE", "inproc://capture", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (faucet, "Hello", "World", NULL);\n    zstr_recvx (sink, &hello, &world, NULL);\n    assert (streq (hello, "Hello"));\n    assert (streq (world, "World"));\n    zstr_free (&hello);\n    zstr_free (&world);\n\n    zstr_recvx (capture, &hello, &world, NULL);\n    assert (streq (hello, "Hello"));\n    assert (streq (world, "World"));\n    zstr_free (&hello);\n    zstr_free (&world);\n\n    zsock_destroy (&faucet);\n    zsock_destroy (&sink);\n    zsock_destroy (&capture);\n    zactor_destroy (&proxy);\n\n    //  Test socket creation dependency\n    proxy = zactor_new (zproxy, NULL);\n    assert (proxy);\n\n    char *frontend = NULL;\n    char *backend = NULL;\n    backend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());\n    zclock_sleep (200);\n    sink = zsock_new_sub (backend, "whatever");\n    assert (sink);\n\n    zstr_sendx (proxy, "BACKEND", "XPUB", backend, NULL);\n    zsock_wait (proxy);\n\n    zsock_destroy(&sink);\n    zactor_destroy(&proxy);\n    zstr_free (&backend);\n\n#ifdef CZMQ_BUILD_DRAFT_API\n    //  Create and configure our proxy with PUB/SUB to test subscriptions\n    proxy = zactor_new (zproxy, NULL);\n    assert (proxy);\n    if (verbose) {\n        zstr_sendx (proxy, "VERBOSE", NULL);\n        zsock_wait (proxy);\n    }\n    zstr_sendx (proxy, "FRONTEND", "SUB", "inproc://frontend", "He", "b", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "BACKEND", "PUB", "inproc://backend", NULL);\n    zsock_wait (proxy);\n\n    //  Connect application sockets to proxy\n    faucet = zsock_new_pub (">inproc://frontend");\n    assert (faucet);\n    sink = zsock_new_sub (">inproc://backend", "");\n    assert (sink);\n\n    //  Send some messages and check they arrived\n    zstr_sendx (faucet, "Hello", "World", NULL);\n    // since SUB is binding, subscription might be lost see:\n    // https://github.com/zeromq/libzmq/issues/2267\n    zsock_set_rcvtimeo (sink, 100);\n    hello = zstr_recv (sink);\n    if (hello) {\n        assert (streq (hello, "Hello"));\n        world = zstr_recv (sink);\n        assert (streq (world, "World"));\n        zstr_free (&hello);\n        zstr_free (&world);\n    }\n\n    zsock_destroy (&faucet);\n    zsock_destroy (&sink);\n    zactor_destroy(&proxy);\n#endif // CZMQ_BUILD_DRAFT_API\n\n#if (ZMQ_VERSION_MAJOR == 4)\n    // Test authentication functionality\n    const char *basedirpath = "src/selftest-rw/.test_zproxy";\n    const char *passfilepath = "src/selftest-rw/.test_zproxy/password-file";\n    const char *certfilepath = "src/selftest-rw/.test_zproxy/mycert.txt";\n\n    // Make sure old aborted tests do not hinder us\n    zdir_t *dir = zdir_new (basedirpath, NULL);\n    if (dir) {\n        zdir_remove (dir, true);\n        zdir_destroy (&dir);\n    }\n    zsys_file_delete (passfilepath);\n    zsys_file_delete (certfilepath);\n    zsys_dir_delete (basedirpath);\n\n    //  Create temporary directory for test files\n    zsys_dir_create (basedirpath);\n\n    //  Check there\'s no authentication\n    s_create_test_sockets (&proxy, &faucet, &sink, verbose);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    bool success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,\n        verbose, true);\n    assert (success);\n\n    //  Install the authenticator\n    zactor_t *auth = zactor_new (zauth, NULL);\n    assert (auth);\n    if (verbose) {\n        zstr_sendx (auth, "VERBOSE", NULL);\n        zsock_wait (auth);\n    }\n\n    //  Check there\'s no authentication on a default NULL server\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        true);\n    assert (success);\n\n    //  When we set a domain on the server, we switch on authentication\n    //  for NULL sockets, but with no policies, the client connection\n    //  will be allowed.\n    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);\n    zsock_wait (proxy);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        true);\n    assert (success);\n\n    //  Block 127.0.0.1, connection should fail\n    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);\n    zsock_wait (proxy);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    zstr_sendx (auth, "DENY", "127.0.0.1", NULL);\n    zsock_wait (auth);\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        false);\n    assert (!success);\n\n    //  Allow our address, which overrides the block list\n    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "DOMAIN", "BACKEND", "global", NULL);\n    zsock_wait (proxy);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    zstr_sendx (auth, "ALLOW", "127.0.0.1", NULL);\n    zsock_wait (auth);\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        true);\n    assert (success);\n\n    //  Try PLAIN authentication\n\n    //  Test negative case (no server-side passwords defined)\n    zstr_sendx (proxy, "PLAIN", "FRONTEND", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);\n    zsock_wait (proxy);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    zsock_set_plain_username (faucet, "admin");\n    zsock_set_plain_password (faucet, "Password");\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        false);\n    assert (!success);\n\n    //  Test positive case (server-side passwords defined)\n    FILE *password = fopen (passfilepath, "w");\n    assert (password);\n    fprintf (password, "admin=Password\\n");\n    fclose (password);\n    zstr_sendx (proxy, "PLAIN", "FRONTEND", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "PLAIN", "BACKEND", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "DOMAIN", "BACKEND", "global", NULL);\n    zsock_wait (proxy);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    zsock_set_plain_username (faucet, "admin");\n    zsock_set_plain_password (faucet, "Password");\n    zsock_set_plain_username (sink, "admin");\n    zsock_set_plain_password (sink, "Password");\n    zstr_sendx (auth, "PLAIN", passfilepath, NULL);\n    zsock_wait (auth);\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        true);\n    assert (success);\n\n    //  Test negative case (bad client password)\n    zstr_sendx (proxy, "PLAIN", "FRONTEND", NULL);\n    zsock_wait (proxy);\n    zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);\n    zsock_wait (proxy);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    zsock_set_plain_username (faucet, "admin");\n    zsock_set_plain_password (faucet, "Bogus");\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        false);\n    assert (!success);\n\n    if (zsys_has_curve ()) {\n        //  We\'ll create two new certificates and save the client public\n        //  certificate on disk\n        zcert_t *server_cert = zcert_new ();\n        assert (server_cert);\n        zcert_t *client_cert = zcert_new ();\n        assert (client_cert);\n        const char *public_key = zcert_public_txt (server_cert);\n        const char *secret_key = zcert_secret_txt (server_cert);\n\n        //  Try CURVE authentication\n\n        //  Test without setting-up any authentication\n        zstr_sendx (proxy, "CURVE", "FRONTEND", public_key, secret_key, NULL);\n        zsock_wait (proxy);\n        zstr_sendx (proxy, "DOMAIN", "FRONTEND", "global", NULL);\n        zsock_wait (proxy);\n        s_bind_test_sockets (proxy, &frontend, &backend);\n        zcert_apply (client_cert, faucet);\n        zsock_set_curve_serverkey (faucet, public_key);\n        success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,\n            verbose, false);\n        assert (!success);\n\n        //  Test CURVE_ALLOW_ANY\n        zstr_sendx (proxy, "CURVE", "FRONTEND", public_key, secret_key, NULL);\n        zsock_wait (proxy);\n        s_bind_test_sockets (proxy, &frontend, &backend);\n        zcert_apply (client_cert, faucet);\n        zsock_set_curve_serverkey (faucet, public_key);\n        zstr_sendx (auth, "CURVE", CURVE_ALLOW_ANY, NULL);\n        zsock_wait (auth);\n        success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,\n            verbose, true);\n        assert (success);\n\n        //  Test with client certificate file in authentication folder\n        zstr_sendx (proxy, "CURVE", "FRONTEND", public_key, secret_key, NULL);\n        zsock_wait (proxy);\n        zstr_sendx (proxy, "CURVE", "BACKEND", public_key, secret_key, NULL);\n        zsock_wait (proxy);\n        s_bind_test_sockets (proxy, &frontend, &backend);\n        zcert_apply (client_cert, faucet);\n        zsock_set_curve_serverkey (faucet, public_key);\n        zcert_apply (client_cert, sink);\n        zsock_set_curve_serverkey (sink, public_key);\n        zcert_save_public (client_cert, certfilepath);\n        zstr_sendx (auth, "CURVE", basedirpath, NULL);\n        zsock_wait (auth);\n        success = s_can_connect (&proxy, &faucet, &sink, frontend, backend,\n            verbose, true);\n        assert (success);\n\n        zcert_destroy (&server_cert);\n        zcert_destroy (&client_cert);\n    }\n\n    //  Remove the authenticator and check a normal connection works\n    zactor_destroy (&auth);\n    s_bind_test_sockets (proxy, &frontend, &backend);\n    success = s_can_connect (&proxy, &faucet, &sink, frontend, backend, verbose,\n        true);\n    assert (success);\n\n    // Test heartbeat functionality\n\n    s_create_test_sockets (&proxy, &faucet, &sink, verbose);\n    s_create_endpoints (proxy, &frontend, &backend);\n    zclock_sleep (200);\n\n    // Set the heartbeat interval to much longer than the TTL so that\n    // the socket times out on the remote side\n    zstr_sendx (proxy, "FRONTEND", "PULL", frontend,\n        "HEARTBEAT_IVL", "300", "HEARTBEAT_TTL", "100", NULL);\n    zsock_wait (proxy);\n\n    zstr_sendx (proxy, "BACKEND", "PUSH", backend, NULL);\n    zsock_wait (proxy);\n\n    // Setup a monitor to make sure we disconnect\n    zactor_t *monitor = zactor_new (zmonitor, faucet);\n    assert (monitor);\n    if (verbose)\n        zstr_sendx (monitor, "VERBOSE", NULL);\n    zstr_sendx (monitor, "LISTEN", "CONNECTED", "DISCONNECTED",  NULL);\n    zstr_sendx (monitor, "START", NULL);\n    zsock_wait (monitor);\n\n    // Connect the faucet\n    int rc = zsock_connect (faucet, "%s", frontend);\n    assert (rc == 0);\n\n    zmsg_t *msg = zmsg_recv (monitor);\n    assert (msg);\n    char *event = zmsg_popstr (msg);\n    assert (streq (event, "CONNECTED"));\n    zmsg_destroy (&msg);\n    zstr_free (&event);\n\n    // time to settle ~300ms\n    zclock_sleep (300);\n\n    // Now we should have been disconnected\n    msg = zmsg_recv (monitor);\n    assert (msg);\n    event = zmsg_popstr (msg);\n    assert (streq (event, "DISCONNECTED"));\n    zmsg_destroy (&msg);\n    zstr_free (&event);\n\n    //  Cleanup\n    zactor_destroy (&monitor);\n    zsock_destroy (&faucet);\n    zsock_destroy (&sink);\n    zactor_destroy (&proxy);\n    zstr_free (&frontend);\n    zstr_free (&backend);\n\n    //  Delete temporary directory and test files\n    zsys_file_delete (passfilepath);\n    zsys_file_delete (certfilepath);\n    zsys_dir_delete (basedirpath);\n#endif\n\n#if defined (__WINDOWS__)\n    zsys_shutdown();\n#endif\n    //  @end\n    printf ("OK\\n");\n}\n', 'message': 'problem : zproxy self test fails under windows\n\nsolution : replace ipc with inproc (only for windows)\n', 'author_name': 'Bruno Bodin', 'lang': 'c'}