FRP пример с циклом событий или пулом потоков

Question

Мой вопрос в конечном счете для библиотеки ruby ​​rx, хотя любой пример на любом языке был бы с удовольствием оценен. В основном я хочу запланировать каждую операцию для существующего цикла событий (или пула потоков, если на то пошло). Я думаю, это должно быть сделано планировщиком. Я не нашел ни одного примера планировщика, отправляющего рекурсивные операции в цикл событий, и именно поэтому я спрашиваю. Вот список для рубиновых ого:

https://github.com/ReactiveX/RxRuby/tree/master/lib/rx/concurrency

Почему цикл обработки событий? Поскольку я хочу добавить операции ввода-вывода, которые работают внутри цикла событий и используют параллелизм. Что-то вроде этого:

Rx::Observable.from_enumerable(hosts). 
    map { |h| HTTP.connect(h) }. 
    map{|host| host.get("http://myservice/somelist.txt") }. 
    on_next { |html| parse(html).each_line.....} # you get the idea 

Answer 1

Это обычно делается с помощью планировщика, и я бы ожидать, что порт RubyRx включил EventloopScheduler.

Вы можете поставить в очередь/планировщик их на него с ObserveOn оператора

Rx::Observable.from_enumerable(hosts). 
    observeOn(els). # you have declared els somewhere else as an EventLoopScheduler instance 
    map { |h| HTTP.connect(h) }. 
    map{|host| host.get("http://myservice/somelist.txt") }. 
    on_next { |html| parse(html).each_line.....} # you get the idea 

или вы могли бы добавить параллелизм на карте

Rx::Observable.from_enumerable(hosts). 
    observeOn(els). # you have declared els somewhere else as an EventLoopScheduler instance 
    map { |h| HTTP.connect(h) }. 
    flatmap{|host| Rx::Observable.start(host.get("http://myservice/somelist.txt"), els) }. 
    on_next { |html| parse(html).each_line.....} # you get the idea 

Я надеюсь, что код может работать (я C#/JS)