如何在Twisted [autobahn] websocket服务器中实时流式输出?
我想使用subprocess.Popen()来执行一个C程序,并实时输出它的输出并将它发送给客户端。但是,输出缓冲并在执行结束时一起发送(Blocking性质)。我怎样才能实时接收输出,然后立即将其发送到Twisted Autobahn中。如何在Twisted [autobahn] websocket服务器中实时流式输出?
def onConnect(self, request):
try:
self.cont_name = ''.join(random.choice(string.lowercase) for i in range(5))
self.file_name = self.cont_name
print("Connecting...")
except Exception:
print("Failed"+str(Exception))
def onOpen(self):
try:
print("open")
except Exception:
print("Couldn't create container")
def onMessage(self, payload,isBinary=False):
cmd = "docker exec "+self.cont_name+" /tmp/./"+self.file_name
a = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, bufsize=1)
for line in iter(a.stdout.readline, b''):
line = line.encode('utf8')
self.sendMessage(line)
def onClose(self, wasClean, code, reason):
try:
print("Closed container...")
except Exception:
print(str(Exception))
当使用子过程执行该命令搬运工,C代码的整个输出在一次,而不是因为它发生返回。对于前:
#include <stdio.h>
#include <unistd.h>
int main(){
int i=0;
for(i=0;i<5;i++){
fflush(stdout);
printf("Rounded\n");
sleep(3);
}
}
在容器中运行之后,程序应该在3秒到客户端返回“四舍五入”。然而,它最终会在执行结束时发送所有'Rounded'。
的不当行为来自环路在该方法中:
def onMessage(self, payload,isBinary=False):
cmd = "docker exec "+self.cont_name+" /tmp/./"+self.file_name
a = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, bufsize=1)
for line in iter(a.stdout.readline, b''):
line = line.encode('utf8')
self.sendMessage(line)
Twisted是一个协作多任务处理系统。默认情况下,所有内容都在单个线程中运行(“反应器线程”)。这意味着所有代码必须定期(并且通常很快)放弃控制,以便其他代码(应用程序代码或Twisted实现代码)有机会运行。该函数中的循环从子进程读取并尝试使用高速公路API发送数据 - 一遍又一遍,从不放弃控制。
阻塞来自Popen对象的读取也可能导致问题。您不知道读取会阻塞多长时间,因此您不知道多长时间会阻止其他代码在反应器线程中运行。你既可以将您的POPEN读取到一个新的线程,他们不会阻止反应堆螺纹:
def onMessage(self, payload,isBinary=False):
cmd = "docker exec "+self.cont_name+" /tmp/./"+self.file_name
popen_in_thread(
lambda line: reactor.callFromThread(
lambda: self.sendMessage(line.encode("utf-8"))
),
[cmd], shell=True, stdout=subprocess.PIPE, bufsize=1
)
def popen_in_thread(callback, *args, **kwargs):
def threaded():
a = subprocess.Popen(*args, **kwargs)
for line in iter(a.stdout.readline, b''):
callback(line)
reactor.callInThread(threaded)
或者,更好,使用双绞线自身的流程支持:
def onMessage(self, payload,isBinary=False):
class ProcessLinesToMessages(ProcessProtocol):
def outReceived(self, output):
buf = self.buf + output
lines = buf.splitlines()
self.buf = lines.pop()
for line in lines:
self.sendMessage(line.encode("utf-8"))
while True:
line, self.buf = self.buf.splitline
reactor.spawnProcess(
ProcessLinesToMessages(),
"docker",
[
"docker",
"exec",
self.cont_name,
"/tmp/./ + self.file_name,
],
)
(既不版本进行测试,希望这个想法很清楚)
可以在这种情况下使用? –
Deferred表示单个未来结果。您想要传输正在进行的流程中的数据。这两件事情没有很好地吻合。不过,您可能想使用管道
你的问题广义上讲是有道理的,但是*广泛地说,它不是很清楚你遇到的问题在哪里,或者有什么建议可以帮助你克服它。你可以包含一些示例代码吗?最好按照http://sscce.org/的说法。 –
@ Jean-PaulCalderone我添加了一些代码。请确认一下。谢谢 –