I use ZeroMQ in PHP for time sensitive realtime signaling. First, some theory: Internet Engineering Task Force (IETF) Request for Comments (RFC) 793 and RFC 1122 specify TCP. In this situation, the property of interest is the user timeout. User timeout is best explained in RFC 5482, which is a proposal to implement standards for communicating user timeout settings between applications:
"The Transmission Control Protocol (TCP) specification [RFC0793] defines a local, per-connection "user timeout" parameter that
specifies the maximum amount of time that transmitted data may remain
unacknowledged before TCP will forcefully close the corresponding
connection. Applications can set and change this parameter with OPEN
and SEND calls. If an end-to-end connectivity disruption lasts longer
than the user timeout, a sender will receive no acknowledgments for
any transmission attempt, including keep-alives, and it will close the
TCP connection when the user timeout occurs...
In the absence of an application-specified user timeout, the TCP
specification [RFC0793] defines a default user timeout of 5 minutes.
The Host Requirements RFC [RFC1122] refines this definition by
introducing two thresholds, R1 and R2 (R2 > R1), that control the
number of retransmission attempts for a single segment. It suggests
that TCP should notify applications when R1 is reached for a
segment,and close the connection when R2 is reached. [RFC1122] also
defines the recommended values for R1 (3 retransmissions) and R2 (100
seconds), noting that R2 for SYN segments should be at least 3
minutes."
There's likely a lot of equipment between the server and client; it's unlikely that you control all of it (unless LAN, of course). Since there is no current standard for agreeing on timeout settings across applications and gateways; and firewalls are reported to close connections, it's best to send 'keep alive' data at regular intervals. Additionally, many other problems could break the connection, and, similarly, the application could fail. Therefore, a keep alive heartbeat implementation helps create a robust application that keeps the connection open and promptly identifies problems.
Heartbeating
I usually use a heartbeat interval between 1 to 5 seconds, depending on the application. The client counts missed heartbeats and responds accordingly if the signaling is down (e.g. fallback to AJAX polling, alert the user, restart a script, etc.). Remember, publisher-subscriber messages are not queued. If the subscriber misses it, it missed it. If you want the server to be able to confirm receipt and client connection status, additional communication processes would be necessary. If the client is down for any amount of time, you cannot be sure if all messages were received. The ZeroMQ Guide has substantial discussion about heartbeating and robust application designs.
"Heartbeating solves the problem of knowing whether a peer is alive or
dead. This is not an issue specific to ZeroMQ. TCP has a long timeout
(30 minutes or so), that means that it can be impossible to know
whether a peer has died, been disconnected, or gone on a weekend to
Prague with a case of vodka, a redhead, and a large expense account."
Alright, here's a code example using your code:
Server.php
<?php
$context = new ZMQContext();
$publisher = new ZMQSocket($context, ZMQ::SOCKET_PUB);
$publisher->bind("tcp://*:5563");
$i = 0; // Iteration counter to trigger message
while (true) {
$publisher->send('HB'); // Distribute the heartbeat (all clients will subscribe)
sleep (5); // Heartbeat interval
$i++ // Clock that ticks up ~5 seconds at a time (alternatively,
// you could set a variable to store the timestamp of the last time the
// messages (A & B) were sent and then check to see if 5 minutes has elapsed).
if ($i >= 60) {
// 300 seconds--or perhaps a hair more--have elapsed, send the messages
$i = 0; // Rest the clock.
$publisher->send("A", ZMQ::MODE_SNDMORE);
$publisher->send("We don't want to see this");
$publisher->send("B", ZMQ::MODE_SNDMORE);
$publisher->send("We would like to see this");
} // Don't forget to close if statement.
}
Client.php
<?php
$context = new ZMQContext();
$subscriber = new ZMQSocket($context, ZMQ::SOCKET_SUB);
$SERVERIP = 'Where is the server?'; // Make sure this is correct and no firewall!
$subscriber->connect("tcp://$SERVERIP:5563");
$subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, "B");
$subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, 'HB'); // Subscribe to the heartbeat
$heartMonitor = time(); // Time we started listening.
while (true) {
$address = $subscriber->recv();
$more = $subscriber->getSockOpt(ZMQ::SOCKOPT_RCVMORE); // See if this is a multipart message
if ($more) {
// Message has second part--normally there would be a loop for variable length messages.
$contents = $subscriber->recv();
}
if ($address == 'HB') {
// Message is the heartbeat, would send this to additional
// application components and have them listen for heartbeat as well.
$heartMonitor = time(); // Time last heartbeat received in seconds
}else{
// Message is your content message; Do stuff.
printf ("[%s] %s%s", $address, $contents, PHP_EOL);
$heartMonitor = time(); // It doesn't matter which message was just received, both confirm connection.
}
$timeout = 20; // At least 3 heartbeat signals were missed.
if ($heartMonitor < (time() - $timeout)) {
// Connection lost. Alert user, fallback, or attempt reconnect, etc. here.
break;
}
} // Don't forget to close while loop.
Finally, there are some ZeroMQ options for modifying TCP settings; however, I am not sure how well this is supported. Additionally, it does not provide reliability checks at the application level like does heartbeating.
So it is best to keep the connection open (don't want to miss messages and this is actually where a lot of the latency reduction comes from compared to other protocols); you don't want to be reconnecting a lot. Additionally, there is no reliable way to change TCP settings on servers, network equipment, and clients. The best approach is to confirm communication at the application level.
I hope this helps and sets you in the right direction. I have not tested this code, but I will try to tomorrow (and then edit this post).
