Using a DAG for describing the dependencies between jobs to run in a Broadway pipeline is the direction I went as well. My use case was to modify a Natural Language Processing pipeline at run-time, but I think the approach is pretty flexible for other contexts.

The gist of the feature was a Dynamically Supervised GenServer maintains the NLP Job/DAGs and dispatches or modifies them when necessary. The WIP of the DAG can be found here. There’s also ETS for a job queue and caching that isn’t ready for multi-node, but it’s worth a look.

I was really happy that all the Broadway pipeline had to do was this:

@impl true
def handle_message(_processor, %Message{} = message, _context) do
  {:ok, ran_job} = Job.run(message.data)
  %Message{message | data: ran_job}
end

I’m planning on pulling this Job module and some supporting contracts into a library, but for now I need to get a feel for the use cases.

Right now the default behavior is to enqueue dependent jobs with the parent as the input upon success, but there are optimizations that a Broadway Processor could do to minimize latency, distribute work, and prevent unnecessary enqueuing, but it works well enough.

I really enjoyed working with Broadway for this project, so I’d love to see where you go with these accessibility use cases.

That looks like a great starting point. I’ll see if I can get a prototype working, based on your code. One thing I’m curious about is the repetition of the map keys as values for the name field, eg google_nlp in:

  google_nlp: %Job{
    name: :google_nlp,
    work: &GoogleNLP.analyze_text/1,
  },

It looks like this builds trees automagically, based on the hierarchy of the data structure, but I’m unclear about how a generalized DAG structure would be specified. Help?

-r

The main reasoning behind using the names as keys is so the code to modify the job pipeline at run-time is easier to write. Just some Map.get calls. The names as keys also helps with preventing duplicate jobs in the same level. I expect there would be a better way maybe by adding more info in the key to aid in retrieval.

The building of the pipeline is done here and dispatching the jobs here. The SessionProcess is just initiated with an empty map it uses to put the top-level jobs in. Each Job has a dependent job to publish to the PubSub currently, but it can be used for about as much arbitrary nesting as you can fit in-memory. Considering I ended up building helper functions to construct the pipeline I expect there’s a better abstraction hiding somewhere.

The gist is using Job.new/1 to build the struct with initial details and Job.add_dependent_job/2 to add something that should run after with the result of the parent. The purpose of the module is to describe the procedure that should take place for running the jobs. I.e. “Run this only after this, but this other thing can run concurrently if you want”.

Nobody has responded to the question about creating a server that can act as an intelligent web cache. Put another way, it needs to be a man-in-the-middle SSL proxy. Can anyone point me to existing work (preferably in Elixir or Ruby) on this sort of thing?

-r