Skip to content

Instantly share code, notes, and snippets.

@hintjens
Created February 26, 2011 11:29
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save hintjens/845127 to your computer and use it in GitHub Desktop.
Save hintjens/845127 to your computer and use it in GitHub Desktop.
Tries to reconnect to server
//
// Hello World server
// Binds REP socket to tcp://*:5555
// Expects "Hello" from client, replies with "World"
//
// Changes for 2.1:
// - added socket close before terminating
//
#include <zmq.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
int main () {
void *context = zmq_init (1);
// Socket to talk to clients
void *responder = zmq_socket (context, ZMQ_REP);
zmq_bind (responder, "tcp://*:5555");
while (1) {
// Wait for next request from client
zmq_msg_t request;
zmq_msg_init (&request);
zmq_recv (responder, &request, 0);
printf ("Received Hello\n");
zmq_msg_close (&request);
// Do some 'work'
sleep (1);
// Send reply back to client
zmq_msg_t reply;
zmq_msg_init_data (&reply, "World", 5, NULL, NULL);
zmq_send (responder, &reply, 0);
zmq_msg_close (&reply);
}
// We never get here but if we did, this would be how we end
zmq_close (responder);
zmq_term (context);
return 0;
}
//
// Client-side pirate
// Use zmq_poll to do a safe request-reply
// To run, start hwserver and then randomly kill/restart it
//
#include "zhelpers.h"
#define REQUEST_TIMEOUT 2500 // msecs
#define REQUEST_RETRIES 3 // Before we abandon
int main () {
void *context = zmq_init (1);
void *client = zmq_socket (context, ZMQ_REQ);
int rc = zmq_connect (client, "tcp://localhost:5555");
assert (rc == 0);
// Socket to talk to server
printf ("Connecting to hello world server...\n");
int retries_left = REQUEST_RETRIES;
while (1) {
rc = s_send (client, "HELLO");
printf ("Trying to contact server, %s\n", rc == 0? "OK": "FAILED");
zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
zmq_poll (items, 1, REQUEST_TIMEOUT * 1000);
if (items [0].revents & ZMQ_POLLIN) {
// We got a reply from the server
char *reply = s_recv (client);
free (reply);
printf ("Server replied\n");
retries_left = REQUEST_RETRIES;
}
else {
// No response from server
if (--retries_left == 0) {
printf ("E: server seems to be offline, abandoning\n");
break;
}
else
printf ("W: no response from server, retrying...\n");
}
}
// Discard any outgoing requests
int linger = 0;
zmq_setsockopt (client, ZMQ_LINGER, &linger, sizeof (linger));
zmq_close (client);
zmq_term (context);
return 0;
}
/* =========================================================================
zhelpers.h
Helper header file for example applications.
-------------------------------------------------------------------------
Copyright (c) 1991-2010 iMatix Corporation <www.imatix.com>
Copyright other contributors as noted in the AUTHORS file.
This file is part of the ZeroMQ Guide: http://zguide.zeromq.org
This is free software; you can redistribute it and/or modify it under the
terms of the GNU Lesser General Public License as published by the Free
Software Foundation; either version 3 of the License, or (at your option)
any later version.
This software is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABIL-
ITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
=========================================================================
*/
#ifndef __ZHELPERS_H_INCLUDED__
#define __ZHELPERS_H_INCLUDED__
// Include a bunch of headers that we will need in the examples
#include <zmq.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <string.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <assert.h>
#include <inttypes.h>
// Bring Windows MSVC up to C99 scratch
#if (defined (__WINDOWS__))
typedef unsigned long ulong;
typedef unsigned int uint;
typedef __int64 int64_t;
#endif
// Provide random number from 0..(num-1)
#define randof(num) (int) ((float) (num) * random () / (RAND_MAX + 1.0))
// Receive 0MQ string from socket and convert into C string
// Caller must free returned string. Returns NULL if the context
// is being terminated.
static char *
s_recv (void *socket) {
zmq_msg_t message;
zmq_msg_init (&message);
if (zmq_recv (socket, &message, 0))
return (NULL);
int size = zmq_msg_size (&message);
char *string = malloc (size + 1);
memcpy (string, zmq_msg_data (&message), size);
zmq_msg_close (&message);
string [size] = 0;
return (string);
}
// Convert C string to 0MQ string and send to socket
static int
s_send (void *socket, char *string) {
int rc;
zmq_msg_t message;
zmq_msg_init_size (&message, strlen (string));
memcpy (zmq_msg_data (&message), string, strlen (string));
rc = zmq_send (socket, &message, 0);
zmq_msg_close (&message);
return (rc);
}
// Sends string as 0MQ string, as multipart non-terminal
static int
s_sendmore (void *socket, char *string) {
int rc;
zmq_msg_t message;
zmq_msg_init_size (&message, strlen (string));
memcpy (zmq_msg_data (&message), string, strlen (string));
rc = zmq_send (socket, &message, ZMQ_SNDMORE);
zmq_msg_close (&message);
return (rc);
}
// Receives all message parts from socket, prints neatly
//
static void
s_dump (void *socket)
{
puts ("----------------------------------------");
while (1) {
// Process all parts of the message
zmq_msg_t message;
zmq_msg_init (&message);
zmq_recv (socket, &message, 0);
// Dump the message as text or binary
char *data = zmq_msg_data (&message);
int size = zmq_msg_size (&message);
int is_text = 1;
int char_nbr;
for (char_nbr = 0; char_nbr < size; char_nbr++)
if ((unsigned char) data [char_nbr] < 32
|| (unsigned char) data [char_nbr] > 127)
is_text = 0;
printf ("[%03d] ", size);
for (char_nbr = 0; char_nbr < size; char_nbr++) {
if (is_text)
printf ("%c", data [char_nbr]);
else
printf ("%02X", (unsigned char) data [char_nbr]);
}
printf ("\n");
int64_t more; // Multipart detection
size_t more_size = sizeof (more);
zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
zmq_msg_close (&message);
if (!more)
break; // Last message part
}
}
// Set simple random printable identity on socket
//
static void
s_set_id (void *socket)
{
char identity [10];
sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));
zmq_setsockopt (socket, ZMQ_IDENTITY, identity, strlen (identity));
}
// Report 0MQ version number
//
static void
s_version (void)
{
int major, minor, patch;
zmq_version (&major, &minor, &patch);
printf ("Current 0MQ version is %d.%d.%d\n", major, minor, patch);
}
// Require at least some specified version
static void
s_version_assert (int want_major, int want_minor)
{
int major, minor, patch;
zmq_version (&major, &minor, &patch);
if (major < want_major
|| (major == want_major && minor < want_minor)) {
printf ("Current 0MQ version is %d.%d\n", major, minor);
printf ("Application needs at least %d.%d - cannot continue\n",
want_major, want_minor);
exit (EXIT_FAILURE);
}
}
// Sleep for a number of milliseconds
static void
s_sleep (int msecs)
{
#if (defined (__WINDOWS__))
Sleep (msecs);
#else
struct timespec t;
t.tv_sec = msecs / 1000;
t.tv_nsec = (msecs % 1000) * 1000000;
int rc = nanosleep (&t, NULL);
assert (rc == 0);
#endif
}
// Return current system clock as milliseconds
static int64_t
s_clock (void)
{
#if (defined (__WINDOWS__))
SYSTEMTIME st;
GetSystemTime (&st);
return (int64_t) st.wSecond * 1000 + st.wMilliseconds;
#else
struct timeval tv;
int rc = gettimeofday (&tv, NULL);
assert (rc == 0);
return (int64_t) (tv.tv_sec * 1000 + tv.tv_usec / 1000);
#endif
}
#endif // __ZHELPERS_H_INCLUDED__
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment