Streaming-HTTP-Antwort mit Python "Anfragen" Bibliothek lesen

Question

Ich versuche, einen Ereignisstrom von der Kubernetes API mit dem requests-Modul zur Verfügung gestellt zu konsumieren. Ich habe in, was wie ein Pufferungsproblem scheint: das requests Modul scheint um ein Ereignis zu verzögern.

Ich habe Code, der etwa wie folgt aussieht:

r = requests.get('http://localhost:8080/api/v1beta1/watch/services', 
       stream=True) 

for line in r.iter_lines(): 
    print 'LINE:', line 

Als Kubernetes Ereignisbenachrichtigungen sendet, wird dieser Code nur Anzeige das letzte Ereignis emittiert wird, wenn ein neues Ereignis kommt, das es fast macht völlig nutzlos für Code, der auf den Dienst antworten muss Ereignisse hinzufügen/löschen.

Ich habe dies durch curl in einem Subprozess Laichen gelöst statt mit die requests Bibliothek:

p = subprocess.Popen(['curl', '-sfN', 
         'http://localhost:8080/api/watch/services'], 
        stdout=subprocess.PIPE, 
        bufsize=1) 

for line in iter(p.stdout.readline, b''): 
    print 'LINE:', line 

Dies funktioniert, aber auf Kosten einer gewissen Flexibilität. Gibt es eine Möglichkeit, dieses Pufferproblem mit der Bibliothek requests zu vermeiden?

Answer 1

Dieses Verhalten ist aufgrund einer fehlerhaften Implementierung der iter_lines -Methode in der requests-Bibliothek.

iter_lines iteriert über den Antwortinhalt in chunk_size Blöcke von Daten mit dem Iterator. Wenn es weniger als chunk_size Bytes von Daten für von dem entfernten Server Lesen (die in der Regel der Fall sein wird, wenn die letzte Zeile des Ausgangs Lesen), wird der Lesevorgang blockiert, bis chunk_size Bytes Daten verfügbar sind.

Ich habe meine eigene iter_lines Routine geschrieben, die richtig arbeitet:

import os 


def iter_lines(fd, chunk_size=1024): 
    '''Iterates over the content of a file-like object line-by-line.''' 

    pending = None 

    while True: 
     chunk = os.read(fd.fileno(), chunk_size) 
     if not chunk: 
      break 

     if pending is not None: 
      chunk = pending + chunk 
      pending = None 

     lines = chunk.splitlines() 

     if lines and lines[-1]: 
      pending = lines.pop() 

     for line in lines: 
      yield line 

    if pending: 
     yield(pending) 

für einen Puffer zu füllen Das funktioniert, weil os.read kehren weniger als chunk_size Bytes von Daten, anstatt zu warten.