扭曲的spawnProcess，将一个进程的输出发送到另一个进程的输入

我正在尝试使用扭曲的spawnProcess来复制类似这样的行为：

cat <input.txt | wc -w

这只是两个命令的示例，实际上我有自己的进程（例如 python 或 bash 脚本或外部程序），其中每个进程从 stdin 读取并写入 stdout。就像上面的示例一样，我想将标准输出从一个进程传输到另一个进程的标准输入，并且我想使用spawnProcess 来完成此操作。我在这里使用了一些提示：

用spawnProcess扭曲管道两个进程 https://stackoverflow.com/questions/23012265/twisted-pipe-two-processes-with-spawnprocess

但我无法让它发挥作用。当从第二个spawnProcess协议上的标准输入读取时，它只是挂起。我的代码如下。我做错了什么？我究竟怎样才能实现这个目标？从第一个spawnProcess中调用第二个spawnProcess是否更好？

#!/usr/bin/env python

from twisted.internet import protocol
from twisted.internet import reactor
import re
import os
import sys

class CatPP(protocol.ProcessProtocol):
    def __init__(self,input_data):
        self.input_data=input_data
        self.data = ""

    def connectionMade(self):
        print "connectionMade in CatPP! Now writing to stdin of cat"
        print "   writing this data: %s" % self.input_data
        self.transport.write(self.input_data+'\n')
        print "   closing stdin"
        self.transport.closeStdin() # tell them we're done
        print "   stdin closed"

    def outReceived(self, data):
        print "outReceived from cat! with %d bytes!" % len(data)
        self.data = self.data + data
        print "    received this: %s" % self.data

    def errReceived(self, data):
        print "errReceived from cat! with %d bytes!" % len(data)

    def inConnectionLost(self):
        print "inConnectionLost for cat! stdin is closed! (we probably did it)"

    def outConnectionLost(self):
        print "outConnectionLost for cat! The child closed their stdout!"
        # now is the time to examine what they wrote
        print "I saw cat write this:", self.data

    def errConnectionLost(self):
        print "errConnectionLost for cat! The child closed their stderr."

    def processExited(self, reason):
        print "processExited for cat, status %d" % (reason.value.exitCode,)

    def processEnded(self, reason):
        print "processEnded for cat, status %d" % (reason.value.exitCode,)

class WcPP(protocol.ProcessProtocol):
    def __init__(self):
        self.data = ""

    def connectionMade(self):
        print "connectionMade! Now reading from pipe to get stdin for wp"
        print "    reading from stdin"
        txt = sys.stdin.read()
        print "  Read this from stdin: %s" % (txt,)
        self.transport.write(txt)
        self.transport.closeStdin() # tell them we're done

    def outReceived(self, data):
        print "outReceived from cat! with %d bytes!" % len(data)
        self.data = self.data + data

    def errReceived(self, data):
        print "errReceived from cat! with %d bytes!" % len(data)

    def inConnectionLost(self):
        print "inConnectionLost for cat! stdin is closed! (we probably did it)"

    def outConnectionLost(self):
        print "outConnectionLost for cat! The child closed their stdout!"
        # now is the time to examine what they wrote
        print "Final output:", self.data
        #(dummy, lines, words, chars, file) = re.split(r'\s+', self.data)
        #print "I saw %s lines" % lines

    def errConnectionLost(self):
        print "errConnectionLost for cat! The child closed their stderr."

    def processExited(self, reason):
        print "processExited for cat, status %d" % (reason.value.exitCode,)

    def processEnded(self, reason):
        print "processEnded for cat, status %d" % (reason.value.exitCode,)
        reactor.stop()

readPipe, writePipe = os.pipe()

handle=open('junkin.txt','r')
cat_txt=handle.read()
handle.close()

pp1 = CatPP(cat_txt)
pp2 = WcPP()
reactor.spawnProcess(pp1, "cat", ["cat"], {}, childFDs={1: writePipe})
reactor.spawnProcess(pp2, "wc", ["wc", "-w"], {},childFDs={0: readPipe})
reactor.run()
try:
    os.close(readPipe)
except:
    print "Exception closing readPipe"
try:
    os.close(writePipe)
except:
    print "Exception closing writePipe"

这是一个工作示例。

配管通过时cat | wc, spawnProcess重复管道，因此您需要关闭它们。

from twisted.internet import protocol
from twisted.internet import reactor
import os

class Writer(protocol.ProcessProtocol):
  def __init__(self, data):
    self.data = data
  def connectionMade(self):
    print "Writer -- connection made"
    self.transport.writeToChild(0, self.data)
    self.transport.closeChildFD(0)
  def childDataReceived(self, fd, data):
    pass
  def processEnded(self, status):
    pass

class Reader(protocol.ProcessProtocol):
  def __init__(self):
    pass
  def connectionMade(self):
    print "Reader -- connection made"
    pass
  def childDataReceived(self, fd, data):
    print "Reader -- childDataReceived"
    self.received = data
  def processEnded(self, status):
    print "process ended, got:", self.received

class WriteRead(protocol.ProcessProtocol):
  def __init__(self, data):
    self.data = data
  def connectionMade(self):
    self.transport.writeToChild(0, self.data)
    self.transport.closeChildFD(0)
  def childDataReceived(self, fd, data):
    self.received = data
    print "got data:", data
  def processEnded(self, status):
    print "process ended - now what?"

def test1(data):
    # just call wc
    p2 = reactor.spawnProcess(WriteRead(data), "wc", ["wc"], env=None, childFDs={0: "w", 1: "r"})
    reactor.run()

def test2(data):
    rfd, wfd = os.pipe()
    p1 = reactor.spawnProcess(Writer(data), "cat", ["cat"], env=None, childFDs={0:"w", 1: wfd })
    p2 = reactor.spawnProcess(Reader(),     "wc", ["wc", "-w"], env=None, childFDs={0: rfd, 1: "r"})
    os.close(rfd)
    os.close(wfd)
    reactor.run()

test2("this is a test")