、Observer.createに他のパラメータを渡すにはどうしたらいいですか？

Question

説明のためにRxPYを使用しています。

私は関数からオブザーバブルを作成したいが、その関数はパラメータを取らなければならない。この特定の例は、ランダムな間隔で、私がそれに送るたくさんのあらかじめ定義されたティッカーの1つを返す必要があります。これまでの私の解決策は、クロージャを使用することです：

from __future__ import print_function 

from rx import Observable 
import random 
import string 
import time 

def make_tickers(n = 300, s = 123): 
    """ generates up to n unique 3-letter strings geach makde up of uppsercase letters""" 
    random.seed(s) 
    tickers = [''.join(random.choice(string.ascii_uppercase) for _ in range(3)) for y in range(n)] 
    tickers = list(set(tickers)) # unique 
    print(len(tickers)) 
    return(tickers) 

def spawn_prices_fn(tickers): 
    """ returns a function that will return a random element 
    out of tickers every 20-100 ms, and takes an observable parameter """ 

    def spawner(observer): 
     while True: 
      next_tick = random.choice(tickers) 
      observer.on_next(next_tick) 
      time.sleep(random.randint(20, 100)/1000.0) 

    return(spawner) 


if __name__ == "__main__": 
    spawned = spawn_prices_fn(make_tickers()) 
    xx = Observable.create(spawned) 
    xx.subscribe(lambda s: print(s)) 

もっと簡単な方法はありますか？クロージャを必要としないObservable.createの最初のパラメタ関数にさらにパラメータを送ることはできますか？標準的なアドバイスは何ですか？

Answer 1

これはさまざまな方法で行うことができますが、ここではコードをあまり変更しないソリューションの1つがあります。生成テロップ 注意も私もわずかにするためにコードを調整rx様

以上であることがいくつかのrxマジックと組み合わせ、単一の文字列を生成する機能に分割することができるflake8幸せ

from __future__ import print_function 

import random 
import string 
import time 

from rx import Observable 


def make_tickers(n=300, s=123): 
    """ 
    Generates up to n unique 3-letter strings each made up of uppercase letters 
    """ 
    random.seed(s) 
    tickers = [''.join(random.choice(string.ascii_uppercase) for _ in range(3)) 
       for y in range(n)] 
    tickers = list(set(tickers)) # unique 
    print(len(tickers)) 
    return(tickers) 


def random_picker(tickers): 
    ticker = random.choice(tickers) 
    time.sleep(random.randint(20, 100)/1000.0) 
    return ticker 


if __name__ == "__main__": 
    xx = Observable\ 
     .repeat(make_tickers())\ 
     .map(random_picker)\ 
     .subscribe(lambda s: print(s)) 

又はmake_tickersない溶液：

from __future__ import print_function 

import random 
import string 
import time 

from rx import Observable 


def random_picker(tickers): 
    ticker = random.choice(tickers) 
    time.sleep(random.randint(20, 100)/1000.0) 
    return ticker 


if __name__ == "__main__": 
    random.seed(123) 
    Observable.range(1, 300)\ 
     .map(lambda _: ''.join(random.choice(string.ascii_uppercase) 
           for _ in range(3)))\ 
     .reduce(lambda x, y: x + [y], [])\ 
     .do_while(lambda _: True)\ 
     .map(random_picker)\ 
     .subscribe(lambda s: print(s)) 

time.sleepがから離れて移動することができ0だがコードはやや複雑になるだろう