keepalive_test.py

#!/usr/bin/python3
#-*- coding: utf-8 -*-

# This is just a fixed version of https://gist.github.com/moio/13030dbf575f998cdd6bcc430da6c4fe

import subprocess
import sys
import zmq

SERVER_ADDRESS = "tcp://127.0.0.1:4505"

print("libzmq-%s" % zmq.zmq_version())
print("pyzmq-%s" % zmq.pyzmq_version())
print("Python-%s" % sys.version)

context = zmq.Context()
socket = context.socket(zmq.XSUB)

print("TCP_KEEPALIVE value: %s" % socket.getsockopt(zmq.TCP_KEEPALIVE))

socket.connect(SERVER_ADDRESS)
subprocess.call("netstat -eepto | grep 4505 | grep root | grep ESTABLISHED", shell=True)

socket = context.socket(zmq.XSUB)
socket.setsockopt(zmq.TCP_KEEPALIVE, 1)
socket.setsockopt(zmq.TCP_KEEPALIVE_IDLE, -1)
print("TCP_KEEPALIVE value: %s" % socket.getsockopt(zmq.TCP_KEEPALIVE))

socket.connect("tcp://127.0.0.1:4505")
subprocess.call("netstat -eepto | grep 4505 | grep root | grep ESTABLISHED", shell=True)