python - 让 Pynng 和 socket 相互对话

Question

长话短说

我使用 pynng 启动了一个服务器，然后来自 Python 标准库 socket 的客户端 将尝试向它发送消息。

问题是 client 可以发送消息，但是 server 却忽略了它。因此，它不起作用。

我错过了什么吗？一些低级协议(protocol)设置？一些终止符？

我这样做的原因是我将构建一个使用 pynng 充当服务器的 Python 脚本。然后一个非 Python 程序(我假设它了解基本的 TCP 协议(protocol))将尝试与这个 Python 服务器对话。因此，我正在使用恕我直言，我可以操作的最原始的套接字库，即标准库中的 socket 模块。

详情

我会在讨论时展示代码片段，但我会在最后展示完整的最小代码示例。

我正在尝试使用 pynng 启动服务器

def server():
    with pynng.Pair0(listen=f'tcp://{HOST:s}:{PORT:d}', recv_timeout=10000) as s:
        print("Server running")
        data = s.recv()  # Blocks forever here
        print(data)

然后，看起来像这样的客户端将尝试连接到它:

def client():
    with socket.create_connection(address=(HOST, PORT), timeout=5) as s:
        print("Client connected")
        s.sendall(b'Hello world')
        print("Client sent message")

我在使用threading 时将它们放在一起:

def main():
    srv = threading.Thread(target=server)
    cli = threading.Thread(target=client)

    srv.start()
    cli.start()

    srv.join()
    cli.join()

最小工作代码

总而言之，这是最低限度的工作代码:

import socket
import pynng
import threading

HOST = "127.0.0.1"
PORT = 65432

def main():
    srv = threading.Thread(target=server)
    cli = threading.Thread(target=client)

    srv.start()
    cli.start()

    srv.join()
    cli.join()

def server():
    with pynng.Pair0(listen=f'tcp://{HOST:s}:{PORT:d}', recv_timeout=10000) as s:
        print("Server running")
        data = s.recv()  # Blocks forever here
        print("Message received")
        print(data)

def client():
    with socket.create_connection(address=(HOST, PORT), timeout=5) as s:
        print("Client connected")
        s.sendall(b'Hello world')
        print("Client sent message")


if __name__ == "__main__":
    main()

然后我在终端中运行这个

$ python main.py

看来 server 无法recv 消息，因此 recv 尝试在 10000 毫秒处超时。

Server running
Client connected
Client sent message
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/home/kmonisit/miniconda3/envs/engg/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/home/kmonisit/miniconda3/envs/engg/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "main.py", line 39, in server
    data = s.recv()  # Blocks forever here
  File "/home/kmonisit/miniconda3/envs/engg/lib/python3.8/site-packages/pynng/nng.py", line 454, in recv
    check_err(ret)
  File "/home/kmonisit/miniconda3/envs/engg/lib/python3.8/site-packages/pynng/exceptions.py", line 201, in check_err
    raise exc(string, err)
pynng.exceptions.Timeout: Timed out

Answer 1

pynng基于 Nanomsg Next Generation , 这是 Scalability Protocols 的实现.可伸缩性协议(protocol)适用于许多不同的传输，包括 tcp，但 bare sockets不兼容。 然而，只要稍加努力和努力，它们就可以兼容。也就是说，如果需要，您可以在纯 Python 中实现可伸缩性协议(protocol)。

首先，我们需要知道wire format是什么；谢天谢地，这是有记录的in an RFC in the original nanomsg repository . Pair0 客户端的实现如下:

class Pair0:
    """A poor implementation of the Pair0 protocol"""

    def __init__(self, host, port, timeout=None):
        self._sock = socket.create_connection(address=(host, port), timeout=timeout)
        # https://github.com/nanomsg/nanomsg/blob/master/rfc/sp-tcp-mapping-01.txt
        # upon making a connection, both ends are required to send this header
        self._sock.send(b'\x00SP\x00\x00\x10\x00\x00')
        print(self._sock.recv(8))

    def send(self, data):
        # messages are simply "length + payload".  Length is 64-bit in network byte
        # order.
        packed = struct.pack('!Q', len(data))
        self._sock.sendall(packed + data)

    def recv(self):
        size_bytes = self._sock.recv(8)
        (size,) = struct.unpack('!Q', size_bytes)
        received = 0
        parts = []
        while received < size:
            data = self._sock.recv(size - received)
            received += len(data)
            parts.append(data)
        return b''.join(parts)

并集成到您的测试程序中:

import socket
import struct
import pynng
import threading
import time

HOST = "127.0.0.1"
PORT = 65432


def main():
    srv = threading.Thread(target=server)

    srv.start()
    # sleep to give the server time to bind to the address
    time.sleep(0.1)
    _client = Pair0(HOST, PORT, 1)
    _client.send(b'hello pynng')
    _client.send(b'hope everything is going well for you')
    print(_client.recv())
    print(_client.recv())
    srv.join()


def server():
    with pynng.Pair0(listen=f'tcp://{HOST:s}:{PORT:d}', recv_timeout=1000) as s:
        print("Server running")
        for _ in range(2):
            data = s.recv()
            print("Message received")
            print(data)
        s.send(b'hello bad client')
        s.send(b'I hope you are doing okay')


class Pair0:
    """A poor implementation of the Pair0 protocol"""

    def __init__(self, host, port, timeout=None):
        self._sock = socket.create_connection(address=(host, port), timeout=timeout)
        # https://github.com/nanomsg/nanomsg/blob/master/rfc/sp-tcp-mapping-01.txt
        # upon making a connection, both ends are required to send this header
        self._sock.send(b'\x00SP\x00\x00\x10\x00\x00')
        print(self._sock.recv(8))

    def send(self, data):
        # messages are simply "length + payload".  Length is 64-bit in network byte
        # order.
        packed = struct.pack('!Q', len(data))
        self._sock.sendall(packed + data)

    def recv(self):
        size_bytes = self._sock.recv(8)
        (size,) = struct.unpack('!Q', size_bytes)
        received = 0
        parts = []
        while received < size:
            data = self._sock.recv(size - received)
            received += len(data)
            parts.append(data)
        return b''.join(parts)


if __name__ == "__main__":
    main()

现在，这远不及 pynng 中的实现(它依赖于底层的 nng 实现)那么健壮。 nng 在边缘条件下做正确的事情™，包括丢失网络、处理多个客户端、跟踪状态机、处理 SIGINT 等。这也是一个不完整的实现，因为它没有 bind 等.

免责声明:我是 pynng 的作者。