Сервер привязан к порту 5555

Сервер будет привязан к порту 5555. В отличие от других примеров, я оставлю ваш сервер фиксированной частью и не буду менять его при подключении к MontitoredQueue. Однако такой обмен не является проблемой и не создаст никаких проблем (при условии, что вы правильно настроите MonitoredQueue).

MonitoredQueue привязан к порту 4444, подключен к порту 5555, публикует трафик на порту 7777

MonitoredQueue находится между клиентом и сервером. Он прослушивает порт 4444, отправляет запросы на сервер и отвечает клиенту. При этом любое проходящее сообщение будет опубликовано с соответствующим префиксом «in» или «out» на сокете PUB. Позже мы увидим, что они будут содержать не только префикс и запрос/ответ, но и личность клиента.

Клиент подключается к порту 4444

Клиент мог напрямую подключаться к серверу по порту 5555, но это не позволяло нам перехватывать трафик. По этой причине мы подключим клиента к порту 4444, где находится MonitoredQueue, ожидающий сервера и прослушивания.

Вы увидите, что клиенту и серверу не придется менять строчку кода для участия в этом обмене.

Реальный код

server.py

В нашем случае сервер ожидает строку, которую можно преобразовать в целое число, и возвращает строку с удвоенным значением.

import zmq

def double_server(server_url="tcp://*:5555"):
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind(server_url)
    print "server started..."
    while True:
        req = socket.recv()
        print "server received request", req
        result = str(2*int(req))
        socket.send(result)
        print "server replied with", result

if __name__ == "__main__":
    double_server()

client.py

Наш клиент попытается 5 раз запросить какой-либо результат на порту 4444 на локальном хосте.

import zmq

def client(server_url="tcp://localhost:4444"):
    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    # socket.setsockopt(zmq.IDENTITY, "client_id_abc") # see Conclusions
    socket.connect(server_url)

    for i in range(5):
        print "request", i
        socket.send(str(i))
        res = socket.recv()
        print i, "result: ", res

if __name__ == "__main__":
    client()

Вы можете попробовать прямо сейчас подключиться к порту 5555, чтобы убедиться, что он работает, но для нашего прослушивания он должен общаться с MonitoredQueue.

monitor.py

Вот и вся магия. pyzmq уже предоставляет устройство MonitoredQueue, поэтому мы можем просто взять его и использовать.

import zmq
from zmq.devices.monitoredqueuedevice import MonitoredQueue
from zmq.utils.strtypes import asbytes

def monitoredqueue(frontend_url="tcp://*:4444", server_url="tcp://localhost:5555", capture_url="tcp://*:7777"):
    mondev = MonitoredQueue(zmq.ROUTER, zmq.DEALER, zmq.PUB, asbytes("in"), asbytes("out"))
    mondev.bind_in(frontend_url)
    mondev.connect_out(server_url)
    mondev.bind_mon(capture_url)
    mondev.setsockopt_in(zmq.HWM, 1)
    mondev.start()
    print "monitored queue started"

if __name__ == "__main__":
    monitoredqueue()

Примечание о типах сокетов и псевдонимах:

• zmq.ROUTER раньше назывался zmq.XREP
• zmq.DEALER раньше назывался zmq.XREQ
• эти псевдонимы все еще работают.

MonitoredQueue будет публиковать каждое сообщение, проходящее через сокет zmq.PUB на порту 7777. Эти сообщения будут иметь префикс «in» и «out», а также будут содержать один фрейм со строкой идентификатора. Эта идентификационная строка назначается сокетом ROUTER и во время обмена уникальна для всех подключенных клиентов. Эта идентификация является частью так называемого конверта и представляет собой сообщение запроса/ответа, ограниченное пустым фреймом (как скоро будет видно).

monitorclient.py

Этот клиент монитора здесь только для того, чтобы показать, как получить перехваченную информацию.

Он подписывается на порт 7777, обслуживаемый монитором (MonitoredQueue), и распечатывает его. Важно использовать составное сообщение, иначе мы упустим часть информации.

import zmq

def monitorclient(server_url="tcp://localhost:7777"):
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.connect(server_url)
    socket.setsockopt(zmq.SUBSCRIBE, "")
    print "started monitoring client"

    while True:
        res = socket.recv_multipart()
        print res

if __name__ == "__main__":
    monitorclient()

Запустить его

Нам понадобится 4 открытые консоли, в каждой мы запустим по одному python скрипту

Сначала запустите сервер:

$ python server.py

Начать мониторинг очереди

$ python monitor.py

Запустить клиент, чтение перехваченных сообщений

$ python monitorclient.py

Наконец, запустите клиент, пытающийся получить ответ от сервера, проксируемого MonitoredQueue.

$ python client.py
request 0
0 result:  0
request 1
1 result:  2
request 2
2 result:  4
request 3
3 result:  6
request 4
4 result:  8

Результаты соответствуют ожиданиям.

Теперь проверьте вывод server.py:

$ python server.py
server received request 0
server replied with 0
server received request 1
server replied with 2
server received request 2
server replied with 4
server received request 3
server replied with 6
server received request 4
server replied with 8

Ничего удивительного, все идет хорошо.

Наш monitor.py ничего не выводит, нам нужно проверить вывод из monitorclient.py

$ python monitorclient.py 
started monitoring client
['in', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '0']
['out', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '0']
['in', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '1']
['out', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '2']
['in', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '2']
['out', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '4']
['in', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '3']
['out', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '6']
['in', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '4']
['out', '\x00\xc4\x84\x1c\xf2\xc2.@\xd3\x86cN\x0e\x06\x7f\xaf\x0b', '', '8']

Здесь вы видите распечатку всех 10 сообщений, 5 запросов, 5 ответов.

Каждый имеет структуру [prefix, identity, emptyframe, message], где

• prefix либо "входит", либо "выходит"
• identity — это строка, назначенная конкретному клиенту MonitoredQueues. Каждый раз, когда клиент подключается, это удостоверение может меняться. В качестве бонуса мы можем подключить несколько клиентов и при этом иметь возможность различать разные клиенты. Если вам нужны определенные идентификаторы клиентов, см. строку с комментариями в client.py с socket.setsockopt(zmq.IDENTITY, "client_id_abc"). Если вы раскомментируете его, вы увидите "client_id_abc" как идентификатор вашего клиента.
• emptyframe рассматривается как '' и отделяет конверт от данных сообщения.
• message — это то, что спросил клиент или какой сервер ответил.

Выводы

• сниффинг работает, и PyZMQ уже предлагает устройство MonitoredQueue для этой цели
• с zmq.PUB сниффинг не будет блокировать связь, вы можете просто проигнорировать сниффинговые данные и все будет работать.
• для производства было бы целесообразно сделать MonitoredQueue фиксированной частью системы, таким образом привязав его к известному IP-адресу и порту. Это потребует изменения на сервере, который должен будет подключиться (вместо текущей привязки). Такое изменение тривиально и не влияет на остальной код и поведение. Если у вас есть только одна конечная точка для мониторинга, вы также можете встроить монитор в сервер (для этого потребуется 2 потока, один для сервера, другой для монитора).
• zmq отличный "Лего" для такого рода задач.