RxPy Wert an Beobachter übergeben

Question

Gibt es eine Möglichkeit, den Wert an den Beobachter entsprechend der Benutzereingabe zu übergeben (was bedeutet, dass der übergebene Wert nicht immer fest ist)?

from rx import Observable, Observer 

def push_five_strings(observer,value): 
     observer.on_next(value) 
     #observer.on_next("Alpha") 
     observer.on_completed() 


class PrintObserver(Observer): 

    def on_next(self, value): 
     print("Received {0}".format(value)) 

    def on_completed(self): 
     print("Done!") 

    def on_error(self, error): 
     print("Error Occurred: {0}".format(error)) 

strings = [("Alpha", "Beta", "Gamma", "Delta", "Epsilon")] 
for i in strings: 
     push_five_strings(strings) #e.g. getting the values to push in one string at a time from a list of strings 
#push_five_strings("Gamma") 
#push_five_strings("Alpha") 
#push_five_strings("Beta") 
#push_five_strings("Delta") 

source = Observable.create(push_five_strings) 
#source = Observable.from_(["Alpha", "Beta", "Gamma", "Delta", "Epsilon"]) 
#source = Observable.from_([value]) 


source.subscribe(PrintObserver()) 

Ich habe versucht, versucht, die Suche um RxPy zu verstehen, aber es gibt kaum Beispiele um im Netz ...

Answer 1

from rx import Observable, Observer                      
import sys                            

class PrintObserver(Observer):                       

    def on_next(self, value):                        
     print("Received {0}".format(value))                    

    def on_completed(self):                        
     print("Done!")                         

    def on_error(self, error):                       
     print("Error Occurred: {0}".format(error))                  

Observable.from_(sys.stdin).subscribe(PrintObserver()) 

Starten und Typisierungsergebnisse in:

abc 
Received abc 

def 
Received def 

Done! 

Stoppen Sie Ihren Eingabestream mit Ctrl+D.