在Python中一次读取stdin行

Question

我正在尝试通过发送一系列命令并捕获输出来测试特定的聊天程序。

我如何通过：

sleep 2
echo test
sleep 2
echo test1

我已经试过了：

(sleep 2; echo test; sleep 2; echo test1) | python3 test.py

但它只打印第一部分，而第二部分我什么也没得到。相反，它进入了一个无限循环。

python程序的代码是：

import sys, select

while True:
  socket_list = [sys.stdin]
  read_sockets, write_sockets, error_sockets = select.select(socket_list, [], [])
  for sock in read_sockets:
    message = sys.stdin.readline()
    sys.stdout.write("> %s: ")
    sys.stdout.flush()

我应该提到这不是完整的程序，但是它是帮助重新创建完全相同的效果的部分。

Answer 1

你的例子进入循环，因为在echo test1 标准输入关闭后read()总是返回空。此外，您不应该使用另一个阻塞调用select。如果您要在多个对象上使用它，您仍然可能被readline 前一个事件阻止。

要从stdin 中读取一行，您不需要使用 select.select(). 可能是实现这一目标的最简单方法：

for line in sys.stdin:
    print(line)

如果您想使用其中stdin一个文件或文件列表作为程序的参数，Python 的 fileinput模块是开箱即用的解决方案。

如果您想要/需要使用select.select()：

import os, sys, select

buffer = ""
while True:
  select.select([sys.stdin.fileno()], [], [])
  read = os.read(sys.stdin.fileno(), 512)

  # empty read: EOF
  if len(read) == 0:
    # buffer might not be empty
    if len(buffer) > 0:
      sys.stdout.write(buffer + "\n")

    break

  # find newlines
  parts = read.split("\n")
  buffer += parts.pop(0)

  while len(parts) > 0:
    sys.stdout.write(buffer + "\n")
    buffer = parts.pop(0)

当一个对象准备好时， select.select 解除阻塞。在这种情况下，唯一的对象是stdin。对于多个对象，您需要检查其返回值select.select以找出哪个对象已准备就绪。os.read()用于从stdin进行非阻塞读取 （或者：可以stdin.read(1)一次读取一个字符）。 read.split("\n")用于查找换行符，请注意，单次读取可能会产生不止一行。