Class: Wurk::Flow::Builder

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/flow/builder.rb

Overview

Declares a flow graph and proves it is one.

Everything here runs before a single key is written. A DAG builder that accepts a cycle deadlocks silently — the parent waits on a dependency that is waiting on the parent, nothing raises, and the flow simply never finishes. The same is true of an unbounded graph, which does not fail so much as bury Redis. So both are refused here, loudly, naming the node.

Validation happens in one pass at #build, not incrementally, because a name may be a forward reference: depends_on: :merge is legal before name: :merge is declared (decision 5 in the slice plan), and that is precisely what makes a cycle expressible and this check load-bearing.

What is not checked here: whether the job class exists, whether its arguments are JSON-native, whether its options are ones Wurk knows. Client owns all three at creation, and a second copy of that check is a second thing to keep in sync.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Builder

Returns a new instance of Builder.



30
31
32
33
34
35
 
# File 'lib/wurk/flow/builder.rb', line 30

def initialize
  @nodes   = []
  @by_name = {}
  @depth   = 0
  @width   = 0
end

Instance Attribute Details

#depth ⇒ Integer (readonly)

Returns number of levels in the graph, longest path first. Set by #build; reading it before then is meaningless.

Returns:

  • (Integer) —

    number of levels in the graph, longest path first. Set by #build; reading it before then is meaningless.



25
26
27
 
# File 'lib/wurk/flow/builder.rb', line 25

def depth
  @depth
end

#width ⇒ Integer (readonly)

Returns the most edges on either side of any one node.

Returns:

  • (Integer) —

    the most edges on either side of any one node.



28
29
30
 
# File 'lib/wurk/flow/builder.rb', line 28

def width
  @width
end

Instance Method Details

#build ⇒ Array<Node>

Resolve, validate and seal the graph.

Returns:

  • (Array<Node>) —

    every node, in topological order — dependencies before dependents, declaration order among equals, so creation and every error message read in the order the caller wrote them.

Raises:



70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/wurk/flow/builder.rb', line 70

def build
  raise InvalidGraph, 'a flow must declare at least one job' if @nodes.empty?

  link!
  order = topological_order
  link_pipes!
  measure!(order)
  check_depth!(order)
  check_width!(order)
  order.each(&:seal!)
  order
end

#job(klass, *args, name: nil, depends_on: nil, pipe: nil, **options) ⇒ Node

Declare one job.

Parameters:

  • klass (Class, String) —

    the job class.

  • args (Array) —

    positional perform_async arguments.

  • name (Symbol, String, nil) (defaults to: nil) —

    a name other nodes may depend on, including from earlier in the block.

  • depends_on (Node, Symbol, String, Array, nil) (defaults to: nil) —

    what must succeed before this node runs — handles returned by earlier #job calls, names, or a mix.

  • pipe (Node, Symbol, String, nil) (defaults to: nil) —

    the one dependency whose stored return value is appended to this node's arguments — a chain link. Implies depends_on:, and cannot be combined with it.

  • options (Hash) —

    job options (queue:, retry:, track:, …), merged into the payload at creation.

Returns:

  • (Node) —

    a handle, usable as another node's depends_on:.



52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/wurk/flow/builder.rb', line 52

def job(klass, *args, name: nil, depends_on: nil, pipe: nil, **options)
  validate_class!(klass)
  validate_pipe!(klass, depends_on, pipe)
  key = register_name!(name)
  check_size!(klass)
  node = Node.new(index: @nodes.size, name: key, klass: klass, args: args, options: options,
                  declared: refs(pipe || depends_on), pipe: pipe)
  @nodes << node
  @by_name[key] = node if key
  node
end