/*

   *  Hello World server

   *  Binds REP socket to tcp://*:5555

   *  Expects "Hello" from client, replies with "World"

   * @author Ian Barber

   */

   $context = new ZMQContext(1);

   //  Socket to talk to clients

   $responder = new ZMQSocket($context, ZMQ::SOCKET_REP);

   $responder->bind("tcp://*:5555");

   while(true) {

       //  Wait for next request from client

       $request = $responder->recv();

       printf ("Received request: [%s]\n", $request);

       //  Do some 'work'

       sleep (1);

       //  Send reply back to client

       $responder->send("World");    

}

   /*

   *  Hello World client

   *  Connects REQ socket to tcp://localhost:5555

   *  Sends "Hello" to server, expects "World" back

   * @author Ian Barber

   */

   $context = new ZMQContext();

   //  Socket to talk to server

   echo "Connecting to hello world server…\n";

   $requester = new ZMQSocket($context, ZMQ::SOCKET_REQ);

   $requester->connect("tcp://localhost:5555");

   for($request_nbr = 0; $request_nbr != 10; $request_nbr++) {

       printf ("Sending request %d…\n", $request_nbr);

       $requester->send("Hello");

       $reply = $requester->recv();

       printf ("Received reply %d: [%s]\n", $request_nbr, $reply);

}

   /*

   *  Weather update server

   *  Binds PUB socket to tcp://*:5556

   *  Publishes random weather updates

   * @author Ian Barber

   */

   //  Prepare our context and publisher

   $context = new ZMQContext();

   $publisher = $context->getSocket(ZMQ::SOCKET_PUB);

   $publisher->bind("tcp://*:5556");

   $publisher->bind("ipc://weather.ipc");

   while (true) {

       //  Get values that will fool the boss

       $zipcode     = mt_rand(0, 100000);

       $temperature = mt_rand(-80, 135);

       $relhumidity = mt_rand(10, 60);

       //  Send message to all subscribers

       $update = sprintf ("%05d %d %d", $zipcode, $temperature, $relhumidity);

       $publisher->send($update);

   }

   /*

   *  Weather update client

   *  Connects SUB socket to tcp://localhost:5556

   *  Collects weather updates and finds avg temp in zipcode

   * @author Ian Barber

   */

   $context = new ZMQContext();

   //  Socket to talk to server

   echo "Collecting updates from weather server…", PHP_EOL;

   $subscriber = new ZMQSocket($context, ZMQ::SOCKET_SUB);

   $subscriber->connect("tcp://localhost:5556");

   //  Subscribe to zipcode, default is NYC, 10001

   $filter = $_SERVER['argc'] > 1 ? $_SERVER['argv'][1] : "10001";

   $subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, $filter);

   //  Process 100 updates

   $total_temp = 0;

   for ($update_nbr = 0; $update_nbr < 100; $update_nbr++) {

       $string = $subscriber->recv();

       sscanf ($string, "%d %d %d", $zipcode, $temperature, $relhumidity);

       $total_temp += $temperature;

   }

   printf ("Average temperature for zipcode '%s' was %dF\n",

       $filter, (int) ($total_temp / $update_nbr));

   /*

   *  Weather proxy device

   * @author Ian Barber

   */

   $context = new ZMQContext();

   //  This is where the weather server sits

   $frontend = new ZMQSocket($context, ZMQ::SOCKET_SUB);

   $frontend->connect("tcp://192.168.55.210:5556");

   //  This is our public endpoint for subscribers

   $backend = new ZMQSocket($context, ZMQ::SOCKET_PUB);

   $backend->bind("tcp://10.1.1.0:8100");

   //  Subscribe on everything

   $frontend->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, "");

   //  Shunt messages out to our own subscribers

   while(true) {

       while(true) {

           //  Process all parts of the message

           $message = $frontend->recv();

           $more = $frontend->getSockOpt(ZMQ::SOCKOPT_RCVMORE);

           $backend->send($message, $more ? ZMQ::SOCKOPT_SNDMORE : 0);

           if(!$more) {

               break; // Last message part

           }

       }

}