使 pynng 和 socket 相互通信

Question

TL;DR

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

问题是 client 可以发送消息，但 server 没有注意到它。因此，它不起作用。

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

我这样做的原因是我将构建一个使用pynng 充当服务器的Python 脚本。然后一个非 Python 程序（我假设它具有基本 TCP 协议的知识）将尝试与这个 Python 服务器通信。因此，我使用的是恕我直言，我可以操作的最原始的套接字库，标准库中的 socket 模块。

详情

我将在讨论时展示代码 sn-ps，但我将在最后展示完整的最小代码示例。

我正在尝试使用 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 的实现。可扩展性协议适用于许多不同的传输，包括 tcp，但 bare sockets 不兼容。 但是，只要稍加努力，就可以使它们兼容。也就是说，如果需要，您可以在纯 Python 中实现可扩展性协议。

首先，我们需要知道有线格式是什么；谢天谢地，这是记录在 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 在边缘条件下执行 The Right Thing™，包括丢失网络、处理多个客户端、跟踪状态机、处理 SIGINT 等。这也是一个不完整的实现，因为它没有 bind 等。

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