client.c

#include <stdio.h>
#include <stdlib.h>

#include "mdp_client.h"

#define SAFEFREE(x)                                                            \
  if (x) {                                                                     \
    free(x);                                                                   \
    x = NULL;                                                                  \
  }

int main() {

  char service[] = "bb-test";
  char endpoint[] = "ipc:///tmp/bbtest.ipc";
  char request_str[128];
  char *cmd = NULL, *reply = NULL;
  int i = 0, loops = 10;
  mdp_client_t **clients = NULL;
  zmsg_t *request = NULL;


  /* Create array of ptr for <loop> clients */
  clients = calloc(loops, sizeof(mdp_client_t *));
  assert(clients != NULL);

  /* create <loops> client sessions and send a request on each */
  for (i = 0; i < loops; i++) {
    /* create a new MDP client session */
    clients[i] = mdp_client_new(endpoint);
    if (!clients[i]) {
      fprintf(stderr, "Error %s\r\n", mdp_client_reason(clients[i]));
      exit(-1);
    }
    /* create new request message */
    request = zmsg_new();
    assert(request != NULL);
    memset(request_str, 0, 128);
    sprintf(request_str, "Request %d", i);
    zmsg_addstr(request, request_str);
    /* send the message as an MDP client request */
    if(mdp_client_request(clients[i], service, &request) ==0 ) {
      fprintf(stdout, "%s sent\r\n", request_str);
    } else {
      fprintf(stderr, "%s NOT SENT (%s)\r\n", request_str, mdp_client_reason(clients[i]));
    }

    zmsg_destroy(&request);

    /* If I add sleep time here, so the worker can process the
     * request and send the reply back, it works just fine.
     * As soon as a drop all requests to the broker, the worker gets
     * stuck at zsock_recv() stuck after processing only one, or a
     * subset of the requests )
     * */
    //sleep(1);
  }

  /*  collect the replies       */
  for (i = 0; i < loops; i++) {

    /* create a message pipe to read the replies */
    zsock_t *client_sock = mdp_client_msgpipe(clients[i]);
    assert(client_sock);
    /* set receive timeout (60s) */
    zsock_set_rcvtimeo(client_sock, 10000);
    /* get the message as "ss" (string and string) into cmd and reply*/
    if (zsock_recv(client_sock, "ss", &cmd, &reply) == 0) {
      fprintf(stdout, "Received: %s: %s\r\n", cmd, reply);
    } else {
      fprintf(stderr, "Failed to receive reply %s\r\n",
              mdp_client_reason(clients[i]));
    }

    /* close the message pipe */
    zmq_close(client_sock);

    /* destroy the client session */
    if (clients[i]) {
      mdp_client_destroy(&clients[i]);
    }

    SAFEFREE(cmd);
    SAFEFREE(reply);
  }

  return 0;
}