Class: WebFunction::Pipeline

Inherits:
Object
  • Object
show all
Defined in:
lib/web_function/pipeline.rb

Overview

A pipeline is a sequence of steps that are executed in order.

Examples:

pipeline = WebFunction::Pipeline.new("https://pipe.example/exec")
pipeline.add_step({ url: "https://a", headers: {}, body: {} })
pipeline.add_step({ url: "https://b", headers: {}, body: {} })
pipeline.execute(returns: :all) # => [{ "a" => 1 }, { "b" => 2 }]

Instance Method Summary collapse

Constructor Details

#initialize(url) ⇒ Pipeline

Returns a new instance of Pipeline.



13
14
15
16
17
 
# File 'lib/web_function/pipeline.rb', line 13

def initialize(url)
  @url = url
  @steps = []
  @promises = []
end

Instance Method Details

#add_step(step) ⇒ Promise

Adds a step to the pipeline.

Parameters:

  • step (Hash) —

    The step to add

Returns:

  • (Promise) —

    A new Promise instance



25
26
27
28
29
30
31
32
33
 
# File 'lib/web_function/pipeline.rb', line 25

def add_step(step)
  n = @promises.count
  promise = Promise.new(self, "$[#{n}]")

  @steps << step
  @promises << promise

  promise
end

#execute(returns: :all) ⇒ Object

Executes the pipeline.

Parameters:

  • returns (String, Symbol) (defaults to: :all) —

    The return type or a JSONPath expression to return a specific value.

Returns:

  • (Object) —

    The response returned by the pipeline.



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/web_function/pipeline.rb', line 41

def execute(returns: :all)
  case returns
  when :all
    responses = Request.execute(@url, args: {
      steps: @steps,
      returns: "$",
    },
    )

    responses.each_with_index do |response, index|
      @promises[index].value = response
    end

    reset!

    responses
  when :last
    response = Request.execute(@url, args: {
      steps: @steps,
      returns: "$[-1:]",
    },
    )

    @promises.last.value = response

    reset!

    response
  else
    response = Request.execute(@url, args: {
      steps: @steps,
      returns: returns,
    },
    )

    reset!

    response
  end
end

#reset! ⇒ void

This method returns an undefined value.

Resets the pipeline.



86
87
88
89
 
# File 'lib/web_function/pipeline.rb', line 86

def reset!
  @steps = []
  @promises = []
end