如何在Twisted [autobahn] websocket服务器中实时流式输出？

Question

我想使用subprocess.Popen（）来执行一个C程序，并实时输出它的输出并将它发送给客户端。但是，输出缓冲并在执行结束时一起发送（Blocking性质）。我怎样才能实时接收输出，然后立即将其发送到Twisted Autobahn中。

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'。

Answer 1

的不当行为来自环路在该方法中：

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, 
     ], 
    ) 

（既不版本进行测试，希望这个想法很清楚）