python - RxPy 将值传递给观察者

Question

有没有办法根据用户输入将值传递给观察者（这意味着传递的值并不总是固定的）？

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())

我试图四处寻找以了解 RxPy，但网络中几乎没有任何示例......

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())  

启动和输入结果：

abc
Received abc

def
Received def

Done!

使用 .停止您的输入流Ctrl+D。