问题描述
我正在尝试使用扭曲的 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 协议的 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,将一个进程的输出发送到另一个进程的输入的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!