-
Notifications
You must be signed in to change notification settings - Fork 75
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
RxRuby and concurrency #82
Comments
|
Is there a scheduler that transports the events to a thread != the main thread?
I tried to implement an own Scheduler but failed. The code is in the lines of this:
Looper is a tread that works on a Queue, enqueue would put the action to this queue (ignoring the due_time and state for now) and then action should be called from the Looper-Thread. Could you give me some pointers on what i miss? I set up the whole chain like this:
|
I see what you mean 😄
Could you give me what result(output) or error you got from the code? |
Hi @minamimonji, thanks for following up. I prepared this stripped down version, perhaps you could have a look. It tries to mimic java Executors. require 'rx_ruby'
# simple thing that takes stuff from a queue and runs it
class Executor
def initialize
@queue = Queue.new
@t = Thread.new do
while true
begin
next_runnable = @queue.pop
puts "calling invoke on #{next_runnable}"
next_runnable.invoke
rescue Exception => e
puts e.backtrace
puts e
end
end
end
end
def enqueue(runnable)
@queue.push(runnable)
end
end
class T
def method_missing(name, *args, &block)
puts "called #{name} with #{args}"
end
end
# that is my scheduler that is connected to an executor
class MyScheduler
def initialize(executor)
@executor = executor
end
def schedule_recursive_with_state(state, action)
dt = Time.now.to_i
#############################
# usually i would expect to pass in state here as second parameter!
res = RxRuby::ScheduledItem.new(self, T.new, dt, &action)
@executor.enqueue(res)
end
end
executor = Executor.new
scheduler = MyScheduler.new(executor)
#event every second
source = RxRuby::Observable.timer(0, 1)
#the events should be handled in my executor -> observe_on with a std. subscriber at the end
source.observe_on(scheduler).subscribe(->(x){puts "onEvent #{x}"}, ->(e){puts "error #{e}"}, ->(){puts "finished"})
sleep(10) Here I have 2 problems:
|
I tried more require 'rx_ruby'
# simple thing that takes stuff from a queue and runs it
class Executor
def initialize
@queue = Queue.new
@t = Thread.new do
puts "ttt"
puts "ExecutorThread #{Thread.current}"
while true
begin
next_runnable = @queue.pop
puts "calling invoke on #{next_runnable}"
next_runnable.invoke
rescue Exception => e
puts e.backtrace
puts e
end
end
end
end
def enqueue(runnable)
@queue.push(runnable)
end
end
# that is my scheduler that is connected to an executor
class MyScheduler
def initialize(executor)
@executor = executor
end
def schedule_recursive_with_state(state, action)
dt = Time.now.to_i
res = RxRuby::ScheduledItem.new(self, T.new, dt, &action)
@executor.enqueue(res)
end
end
executor = Executor.new
scheduler = MyScheduler.new(executor)
#event every second
source = RxRuby::Observable.timer(0, 1)
#the events should be handled in my executor -> observe_on with a std. subscriber at the end
source.map do |x|
puts "in map #{Thread.current}"
x
end.observe_on(scheduler).subscribe(->(x){puts "onEvent #{x} in thread #{Thread.current}"}, ->(e){puts "error #{e}"}, ->(){puts "finished"})
sleep(10) the debug output shows, that there are more events produced, the first event already arrived in the correct thread in the subscriber, but after this, something goes wrong. could you please help me out here? |
@gizmomogwai Thanks for more info. |
@minamimonji thank ... i appreciate your efforts (a lot!!!) |
@gizmomogwai Could you run your code above with class MyScheduler
include RxRuby::Scheduler
def initialize(executor)
@executor = executor
end
def schedule_with_state(state, action)
dt = Time.now.to_i
res = RxRuby::ScheduledItem.new(self, state, dt, &action)
@executor.enqueue(res)
RxRuby::Subscription.empty
end
end The main differences are including the existing |
@minamimonji thanks a lot ... your suggestion solves my problem! |
@gizmomogwai You might want to clean up objects, or threads when def schedule_with_state(state, action)
#...
@executor.enqueue(res)
RxRuby::Subscription.create{ @executor.terminate } # code in the block called when unsubscribe is called.
# Note that you have to define Executor#terminate method to kill the thread `@t` in the `Executor`.
end |
Wow ... thanks a lot ... that really helps! |
There is already ticket #71, but I am stumbling over something else.
The text was updated successfully, but these errors were encountered: