Twisted spawnProcess，将一个进程的输出发送到另一个进程的输入 [英] Twisted spawnProcess, send output of one process to input of another

问题描述

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

I am trying to use twisted spawnProcess to replicate the behavior of something like this:

cat <input.txt | wc -w

这只是两个命令的示例，实际上我有自己的进程(比如 python 或 bash 脚本或外部程序)，其中每个进程从 stdin 读取并写入 stdout.就像上面的例子一样，我想将 stdout 从一个进程传送到另一个进程的 stdin，我想使用 spawnProcess 来做到这一点.我在这里使用了一些提示:

This is just an example of two commands, in reality I have my own processes (say python or bash scripts or external programs) where each process reads from stdin and writes to stdout. Just like the above example, I want to pipe stdout from one process to stdin of another, and I want to do this using spawnProcess. I used some hints here:

用 spawnProcess 扭曲管道两个进程

但我无法让它工作.它只是在从第二个 spawnProcess 协议的 stdin 读取时挂起.我的代码在下面.我做错了什么?我究竟如何才能实现这个目标?从第一个内部调用第二个 spawnProcess 是否更好?

but I can't get it to work. It just hangs when reading from stdin on the second spawnProcess protocol. My code is below.What am I doing wrong? How exactly can I achieve this objective? Is it better to call the second spawnProcess from within the first?

#!/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 复制管道，因此您需要关闭它们.

In the case of piping through cat | wc, spawnProcess duplicates the pipes so you need to close them.

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")

这篇关于Twisted spawnProcess，将一个进程的输出发送到另一个进程的输入的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！