I have a custom Broadway Producer that fetches from a queue. As new messages/jobs are put in the queue I’d like to notify a custom producer to fetch its latent demand from the queue so it can deliver to the Processors.
def MyCustomProducer do ... @impl true def handle_cast(:job_enqueued, state) do handle_receive_messages(state) end end
def JobQueue do ... def enqueue(job) do # put in queue # get pid of producer? GenStage.cast(some_pid, :job_enqueued) end end
I figured the most straight forward way would be to cast to the Producer pid notifying it to fetch messages, but I’m not sure how to find the Producer pid given it’s managed/wrapped by Broadway. Should I be using something like
Broadway.push_messages/2 in this use case?