Schedulers don't appear to be workingfor me, tried several configurations.

[garethhu]

I have two functions that deal with adding to the subject

function source!(src::Source{T}) where T Subject(T, scheduler = AsyncScheduler()) do subject start_consume!(src, subject) end end

function start_consume!(src::Source{T}, consumer) where T while true msg = src.pollFn() if !isnothing(msg) next!(consumer, msg) else yield() end end end

When I replace Subject(T, scheduler = AsyncScheduler()) with make(T) everything works. However I want to run multiple queues on the same scheduler, I have tried running queues by looping through them and using @spawn and this didn't work it ran the first queue but the second did not execute at all. Consequently, my new approach is to give both queues the same scheduler so that the work is distributed between them. But currently, I can't get one queue running with a scheduler.

[bvdmitri]

Hi @garethhu,

I'm not sure if I understand what you are trying to achieve and I also cannot execute your code because I do not have an access to the Source structure. Also this piece of code cannot work since Subject does not accept a function as a first argument to their constructor (make(T) does)

Subject(T) do subject 
    ...
end

Consider using just:

subject = Subject(T)

But to answer your question I can clarify some things about schedulers in Rocket package. They are indeed considered to be an experimental feature and poorly documented. But in any case AsyncScheduler is not supposed to run different sources on the same async queue. It just simply spawns a new Julia Task for every new subscriber and there is no way to run multiple sources on the same queue using AsyncScheduler. Rocket.jl does not implement asynchronous queues and uses Julia internal scheduler to schedule different Tasks if needed.

What you can probably try is to use scheduler = PostponeScheduler().

scheduler = PostponeScheduler()
subject1 = Subject(T, scheduler = scheduler)
subject2 = Subject(T, scheduler = scheduler)

It synchronously adds incoming messages and subscriptions to one single queue. You can release queue with release!(scheduler) (executes actions added before calling release!) or with wait(scheduler) (executes actions added both before and after calling wait).