Class: Ruote::Exp::FlowExpression
- Inherits:
-
Object
- Object
- Ruote::Exp::FlowExpression
- Defined in:
- lib/ruote/exp/ro_on_x.rb,
lib/ruote/exp/ro_timers.rb,
lib/ruote/exp/ro_persist.rb,
lib/ruote/exp/ro_filters.rb,
lib/ruote/exp/ro_variables.rb,
lib/ruote/exp/ro_attributes.rb,
lib/ruote/exp/flow_expression.rb
Overview
Ruote is a process definition interpreter. It doesn't directly “read” process definitions, it relies on a parser/generator to produce “abstract syntax trees” that look like
[ expression_name, { ... attributes ... }, [ children_expressions ] ]
The nodes (and leaves) in the trees are expressions. This is the base class for all expressions.
The most visible expressions are “define”, “sequence” and “participant”. Think :
pdef = Ruote.process_definition do
sequence do
participant :ref => 'customer'
participant :ref => 'accounting'
participant :ref => 'logistics'
end
end
Each node is an expression…
the states of an expression
nil
the normal state
'cancelling'
the expression and its children are getting cancelled
'dying'
the expression and its children are getting killed
'failed'
the expression has finishing
'failing'
the expression just failed and it's cancelling its children
'timing_out'
the expression just timedout and it's cancelling its children
'paused'
the expression is paused, it will store downstream messages and play them only when a 'resume' message comes from upstream.
Direct Known Subclasses
AddBranchesExpression, ApplyExpression, AwaitExpression, CancelProcessExpression, CommandExpression, CommandedExpression, ConcurrenceExpression, CronExpression, DefineExpression, EchoExpression, EqualsExpression, ErrorExpression, FilterExpression, ForgetExpression, IfExpression, ListenExpression, LoseExpression, NoOpExpression, OnErrorExpression, OnceExpression, ParticipantExpression, ReadExpression, RedoExpression, RefExpression, RegisterpExpression, ReserveExpression, RestoreExpression, SaveExpression, SequenceExpression, StallExpression, SubprocessExpression, UndoExpression, WaitExpression
Defined Under Namespace
Classes: HandlerEntry
Constant Summary collapse
- COMMON_ATT_KEYS =
%w[ if unless forget lose flank timeout timers on_error on_cancel on_timeout
Instance Attribute Summary collapse
-
#context ⇒ Object
readonly
Returns the value of attribute context.
-
#error ⇒ Object
Mostly used when the expression is returned via Ruote::Engine#ps(wfid) or Ruote::Engine#processes().
Class Method Summary collapse
-
.do_action(context, msg) ⇒ Object
Called by the worker when it has something to do for a FlowExpression.
-
.dummy(h) ⇒ Object
Returns a dummy expression.
-
.fetch(context, fei) ⇒ Object
Fetches an expression from the storage and readies it for service.
-
.from_h(context, h) ⇒ Object
Instantiates expression back from hash.
-
.names(*exp_names) ⇒ Object
Keeping track of names and aliases for the expression.
Instance Method Summary collapse
-
#ancestor?(fei) ⇒ Boolean
Returns true if the given fei points to an expression in the parent chain of this expression.
-
#applied_workitem ⇒ Object
Returns a one-off Ruote::Workitem instance (the applied workitem).
-
#att(keys, values, opts = {}) ⇒ Object
Returns the value for attribute 'key', this value should be present in the array list 'values'.
-
#att_text(workitem = h.applied_workitem) ⇒ Object
Equivalent to #attribute_text, but will return nil if there is no attribute whose values is nil.
-
#attribute(n, workitem = h.applied_workitem, options = {}) ⇒ Object
Looks up the value for attribute n.
-
#attribute_text(workitem = h.applied_workitem) ⇒ Object
Given something like.
-
#attributes ⇒ Object
Returns the attributes of this expression (like { 'ref' => 'toto' } or { 'timeout' => '2d' }..
-
#await(att, msg) ⇒ Object
If the expression has an :await attribute, the expression gets into a special “awaiting” state until the condition in the value of :await gets triggered and the trigger calls resume on the expression.
-
#cancel(flavour) ⇒ Object
This default implementation cancels all the [registered] children of this expression.
-
#cancel_flanks(flavour) ⇒ Object
Emits a cancel message for each flanking expression (if any).
-
#cfei_at(i) ⇒ Object
Given an index, returns the child fei (among the currently registered children feis) whose fei.expid ends with this index (whose child_id is equal to that index).
-
#child_id ⇒ Object
Returns the child_id for this expression.
-
#child_ids ⇒ Object
Returns the list of child_ids (last part of the fei.expid) for the currently registered (active) children.
-
#compile_atts(opts = {}) ⇒ Object
Returns a Hash containing all attributes set for an expression with their values resolved.
-
#compile_variables ⇒ Object
Returns a fresh hash of all the variables visible from this expression.
-
#debug_id ⇒ Object
Outputs ids like “0_2!d218c1b”, no wfid, only <expid>!<subid>[0, 7].
-
#deflate(err, level = 0) ⇒ Object
Given this expression and an error, deflates the error into a hash (serializable).
-
#do(action, msg) ⇒ Object
Wraps a call to “apply”, “reply”, etc…
-
#do_apply(msg) ⇒ Object
Called by the worker when it has just created this FlowExpression and wants to apply it.
-
#do_cancel(msg) ⇒ Object
The raw handling of messages passed to expressions (the fine handling is done in the #cancel method).
-
#do_fail(msg) ⇒ Object
Called when handling an on_error, will place itself in a 'failing' state and cancel the children (when the reply from the children comes back, the on_error will get triggered).
-
#do_pause(msg) ⇒ Object
Expression received a “pause” message.
-
#do_persist ⇒ Object
Make sure to persist (retry if necessary).
-
#do_reply(msg) ⇒ Object
(also: #do_receive)
Wraps #reply (does the administrative part of the reply work).
-
#do_reply_to_parent(workitem, delete = true) ⇒ Object
The essence of the reply_to_parent job…
-
#do_resume(msg) ⇒ Object
Will “unpause” the expression (if it was paused), and trigger any 'paused_replies' (replies that came while the expression was paused).
-
#do_unpersist ⇒ Object
Make sure to unpersist (retry if necessary).
-
#fei ⇒ Object
Returns the Ruote::FlowExpressionId for this expression.
-
#handle_on_error(msg, error) ⇒ Object
Looks up parent with on_error attribute and triggers it.
-
#has_attribute(*args) ⇒ Object
(also: #has_att)
Given a list of attribute names, returns the first attribute name for which there is a value.
-
#initial_persist ⇒ Object
Persists and fetches the _rev identifier from the storage.
-
#initialize(context, hash) ⇒ FlowExpression
constructor
A new instance of FlowExpression.
-
#is_concurrent? ⇒ Boolean
Concurrent expressions (expressions that apply more than one child at a time) are supposed to return true here.
-
#iterative_var_lookup(k) ⇒ Object
TODO : rdoc rewrite needed.
-
#launch_sub(pos, subtree, opts = {}) ⇒ Object
Launches a subprocesses (usually called from the #apply of certain expression implementations..
- #lookup_val(att_options = {}) ⇒ Object
-
#lookup_val_prefix(prefix, att_options = {}) ⇒ Object
prefix = 'on' => will lookup on, on_val, on_value, on_v, on_var, on_variable, on_f, on_fld, on_field…
-
#lookup_variable(var, prefix = nil) ⇒ Object
(also: #v, #lv)
Looks up the value of a variable in expression tree (seen from a leaf, it looks more like a stack than a tree).
-
#name ⇒ Object
Returns the name of this expression, like 'sequence', 'participant', 'cursor', etc…
-
#parent ⇒ Object
Fetches the parent expression, or returns nil if there is no parent expression.
-
#parent_id ⇒ Object
Returns the Ruote::FlowExpressionIf of the parent expression, or nil if there is no parent expression.
-
#pause_on_apply(msg) ⇒ Object
Called by #do_apply when msg == 'paused'.
- #persist_or_raise ⇒ Object (also: #persist)
-
#reply(workitem) ⇒ Object
A default implementation for all the expressions.
-
#reply_to_parent(workitem, delete = true) ⇒ Object
FlowExpression call this method when they're done and they want their parent expression to take over (it will end up calling the #reply of the parent expression).
-
#root(stubborn = false) ⇒ Object
An expensive method, looks up all the expressions with the same wfid in the storage (for some storages this is not expensive at all), and determine the root of this expression.
-
#root_id(stubborn = false) ⇒ Object
Returns the fei of the root expression of this expression.
-
#set_variable(var, val, override = false) ⇒ Object
Sets a variable to a given value.
-
#tree ⇒ Object
Returns the current version of the tree (returns the updated version if it got updated..
-
#tree_children ⇒ Object
Returns the “AST” view on the children of this expression…
- #try_persist ⇒ Object
- #try_unpersist ⇒ Object
- #unpersist_or_raise ⇒ Object (also: #unpersist)
-
#unset_variable(var, override = false) ⇒ Object
Unbinds a variables.
-
#update_tree(t = nil) ⇒ Object
Updates the tree of this expression.
-
#variables ⇒ Object
A shortcut to the variables held in the expression (nil if none held).
-
#wfid ⇒ Object
Returns the workflow instance id of the workflow this expression belongs to.
Methods included from WithMeta
Methods included from WithH
Constructor Details
#initialize(context, hash) ⇒ FlowExpression
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
# File 'lib/ruote/exp/flow_expression.rb', line 122 def initialize(context, hash) @context = context @msg = nil # contains generally the msg the expression got instantiated for self.h = hash h._id ||= Ruote.to_storage_id(h.fei) h['type'] ||= 'expressions' h.name ||= self.class.expression_names.first h.children ||= [] h.applied_workitem['fei'] = h.fei h.created_time ||= Ruote.now_to_utc_s h.on_cancel ||= attribute(:on_cancel) h.on_error ||= attribute(:on_error) h.on_timeout ||= attribute(:on_timeout) h.on_terminate ||= attribute(:on_terminate) end |
Instance Attribute Details
#context ⇒ Object (readonly)
Returns the value of attribute context
112 113 114 |
# File 'lib/ruote/exp/flow_expression.rb', line 112 def context @context end |
#error ⇒ Object
Mostly used when the expression is returned via Ruote::Engine#ps(wfid) or Ruote::Engine#processes(). If an error occurred for this flow expression, #ps will set this error field so that it yields the ProcessError.
So, for short, usually, this attribute yields nil.
120 121 122 |
# File 'lib/ruote/exp/flow_expression.rb', line 120 def error @error end |
Class Method Details
.do_action(context, msg) ⇒ Object
Called by the worker when it has something to do for a FlowExpression.
304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 |
# File 'lib/ruote/exp/flow_expression.rb', line 304 def self.do_action(context, msg) fei = msg['fei'] action = msg['action'] if action == 'reply' && fei['engine_id'] != context.engine_id # # the reply has to go to another engine, let's locate the # 'engine participant' and give it the workitem/reply # # see ft_37 for a test/example engine_participant = context.plist.lookup(fei['engine_id'], msg['workitem']) raise( "no EngineParticipant found under name '#{fei['engine_id']}'" ) unless engine_participant engine_participant.reply(fei, msg['workitem']) return end # normal case fexp = nil n = context.storage.class.name.match(/Couch/) ? 3 : 1 # n.times do |i| if fexp = fetch(context, msg['fei']); break; end sleep 0.028 unless i == (n - 1) end # # Simplify that once ruote-couch behaves fexp.do(action, msg) if fexp end |
.dummy(h) ⇒ Object
Returns a dummy expression. Only used by the error_handler service.
58 59 60 61 62 63 64 |
# File 'lib/ruote/exp/ro_on_x.rb', line 58 def self.dummy(h) fe = self.allocate fe.instance_variable_set(:@h, h) fe end |
.fetch(context, fei) ⇒ Object
Fetches an expression from the storage and readies it for service.
277 278 279 280 281 282 283 284 |
# File 'lib/ruote/exp/flow_expression.rb', line 277 def self.fetch(context, fei) return nil if fei.nil? fexp = context.storage.get('expressions', Ruote.to_storage_id(fei)) fexp ? from_h(context, fexp) : nil end |
.from_h(context, h) ⇒ Object
Instantiates expression back from hash.
266 267 268 269 270 271 272 273 |
# File 'lib/ruote/exp/flow_expression.rb', line 266 def self.from_h(context, h) return self.new(nil, h) unless context exp_class = context.expmap.expression_class(h['name']) exp_class.new(context, h) end |
.names(*exp_names) ⇒ Object
Keeping track of names and aliases for the expression
292 293 294 295 296 |
# File 'lib/ruote/exp/flow_expression.rb', line 292 def self.names(*exp_names) exp_names = exp_names.collect { |n| n.to_s } (:expression_names) { exp_names } end |
Instance Method Details
#ancestor?(fei) ⇒ Boolean
Returns true if the given fei points to an expression in the parent chain of this expression.
866 867 868 869 870 871 872 873 874 |
# File 'lib/ruote/exp/flow_expression.rb', line 866 def ancestor?(fei) fei = fei.to_h if fei.respond_to?(:to_h) return false unless h.parent_id return true if h.parent_id == fei parent.ancestor?(fei) end |
#applied_workitem ⇒ Object
Returns a one-off Ruote::Workitem instance (the applied workitem).
240 241 242 243 |
# File 'lib/ruote/exp/flow_expression.rb', line 240 def applied_workitem @awi ||= Ruote::Workitem.new(h.applied_workitem) end |
#att(keys, values, opts = {}) ⇒ Object
Returns the value for attribute 'key', this value should be present in the array list 'values'. If not, the default value is returned. By default, the default value is the first element of 'values'.
75 76 77 78 79 80 81 82 |
# File 'lib/ruote/exp/ro_attributes.rb', line 75 def att(keys, values, opts={}) default = opts[:default] || values.first val = Array(keys).collect { |key| attribute(key) }.compact.first.to_s values.include?(val) ? val : default end |
#att_text(workitem = h.applied_workitem) ⇒ Object
Equivalent to #attribute_text, but will return nil if there is no attribute whose values is nil.
139 140 141 142 143 144 |
# File 'lib/ruote/exp/ro_attributes.rb', line 139 def att_text(workitem=h.applied_workitem) text = attributes.keys.find { |k| attributes[k] == nil } text ? dsub(text.to_s, workitem) : nil end |
#attribute(n, workitem = h.applied_workitem, options = {}) ⇒ Object
Looks up the value for attribute n.
48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 |
# File 'lib/ruote/exp/ro_attributes.rb', line 48 def attribute(n, workitem=h.applied_workitem, ={}) n = n.to_s default = [:default] escape = [:escape] string = [:to_s] || [:string] v = attributes[n] v = if v == nil default elsif escape v else dsub(v, workitem) end v = v.to_s if v and string v end |
#attribute_text(workitem = h.applied_workitem) ⇒ Object
Given something like
sequence do
participant 'alpha'
end
in the context of the participant expression
attribute_text()
will yield 'alpha'.
Note : an empty text returns '', not the nil value.
129 130 131 132 133 134 |
# File 'lib/ruote/exp/ro_attributes.rb', line 129 def attribute_text(workitem=h.applied_workitem) text = attributes.keys.find { |k| attributes[k] == nil } dsub(text.to_s, workitem) end |
#attributes ⇒ Object
Returns the attributes of this expression (like { 'ref' => 'toto' } or { 'timeout' => '2d' }.
920 921 922 923 |
# File 'lib/ruote/exp/flow_expression.rb', line 920 def attributes tree[1] end |
#await(att, msg) ⇒ Object
If the expression has an :await attribute, the expression gets into a special “awaiting” state until the condition in the value of :await gets triggered and the trigger calls resume on the expression.
436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 |
# File 'lib/ruote/exp/flow_expression.rb', line 436 def await(att, msg) action, condition = Ruote::Exp::AwaitExpression.extract_await_ac(:await => att) raise ::ArgumentError.new( ":await does not understand #{att.inspect}" ) if action == nil msg.merge!('flavour' => 'awaiting') h.state = 'awaiting' h.paused_apply = msg persist_or_raise @context.tracker.add_tracker( h.fei['wfid'], action, Ruote.to_storage_id(h.fei), condition, { '_auto_remove' => true, 'action' => 'resume', 'fei' => h.fei, 'flavour' => 'awaiting' }) end |
#cancel(flavour) ⇒ Object
This default implementation cancels all the [registered] children of this expression.
727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 |
# File 'lib/ruote/exp/flow_expression.rb', line 727 def cancel(flavour) return reply_to_parent(h.applied_workitem) if h.children.empty? # # there are no children, nothing to cancel, let's just reply to # the parent expression do_persist || return # # before firing the cancel message to the children # # if the do_persist returns false, it means it failed, implying this # expression is stale, let's return, thus discarding this cancel message children.each do |child_fei| # # let's send a cancel message to each of the children # # maybe some of them are gone or have not yet been applied, anyway, # the messages are sent @context.storage.put_msg( 'cancel', 'fei' => child_fei, 'parent_id' => h.fei, # indicating that this is a "cancel child" 'flavour' => flavour) end end |
#cancel_flanks(flavour) ⇒ Object
Emits a cancel message for each flanking expression (if any).
709 710 711 712 713 714 715 716 717 718 719 720 721 722 |
# File 'lib/ruote/exp/flow_expression.rb', line 709 def cancel_flanks(flavour) return unless h.flanks h.flanks.each do |flank_fei| @context.storage.put_msg( 'cancel', 'fei' => flank_fei, 'parent_id' => h.fei, # indicating that this is a "cancel child", well... 'flavour' => flavour) end end |
#cfei_at(i) ⇒ Object
Given an index, returns the child fei (among the currently registered children feis) whose fei.expid ends with this index (whose child_id is equal to that index).
Returns nil if not found or a child fei as a Hash.
251 252 253 254 |
# File 'lib/ruote/exp/flow_expression.rb', line 251 def cfei_at(i) children.find { |cfei| Ruote.extract_child_id(cfei) == i } end |
#child_id ⇒ Object
Returns the child_id for this expression. (The rightmost part of the fei.expid).
172 173 174 175 |
# File 'lib/ruote/exp/flow_expression.rb', line 172 def child_id fei.child_id end |
#child_ids ⇒ Object
Returns the list of child_ids (last part of the fei.expid) for the currently registered (active) children.
259 260 261 262 |
# File 'lib/ruote/exp/flow_expression.rb', line 259 def child_ids children.collect { |cfei| Ruote.extract_child_id(cfei) } end |
#compile_atts(opts = {}) ⇒ Object
Returns a Hash containing all attributes set for an expression with their values resolved.
108 109 110 111 112 113 |
# File 'lib/ruote/exp/ro_attributes.rb', line 108 def compile_atts(opts={}) attributes.keys.each_with_object({}) { |k, r| r[dsub(k)] = attribute(k, h.applied_workitem, opts) } end |
#compile_variables ⇒ Object
Returns a fresh hash of all the variables visible from this expression.
This is used mainly when forgetting an expression.
44 45 46 47 48 49 50 |
# File 'lib/ruote/exp/ro_variables.rb', line 44 def compile_variables vars = h.parent_id ? parent.compile_variables : {} vars.merge!(h.variables) if h.variables vars end |
#debug_id ⇒ Object
Outputs ids like “0_2!d218c1b”, no wfid, only <expid>!<subid>[0, 7]
39 40 41 42 |
# File 'lib/ruote/exp/ro_persist.rb', line 39 def debug_id "#{h.fei['expid']}!#{h.fei['subid'][0, 7]}" end |
#deflate(err, level = 0) ⇒ Object
Given this expression and an error, deflates the error into a hash (serializable).
37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
# File 'lib/ruote/exp/ro_on_x.rb', line 37 def deflate(err, level=0) { 'fei' => h.fei, 'at' => err.respond_to?(:at) ? err.at : Ruote.now_to_utc_s, 'class' => err.class.to_s, 'message' => err., 'trace' => err.backtrace, 'details' => err.respond_to?(:ruote_details) ? err.ruote_details : nil, 'deviations' => err.respond_to?(:deviations) ? err.deviations : nil, 'tree' => tree } rescue => errr # could degenerate (and get stopped by a SystemStackError) deflate(errr, level + 1) end |
#do(action, msg) ⇒ Object
Wraps a call to “apply”, “reply”, etc… Makes sure to set @msg with a deep copy of the msg before.
346 347 348 349 350 351 |
# File 'lib/ruote/exp/flow_expression.rb', line 346 def do(action, msg) @msg = Ruote.fulldup(msg) send("do_#{action}", msg) end |
#do_apply(msg) ⇒ Object
Called by the worker when it has just created this FlowExpression and wants to apply it.
356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 |
# File 'lib/ruote/exp/flow_expression.rb', line 356 def do_apply(msg) if msg['state'] == 'paused' return pause_on_apply(msg) end if msg['flavour'].nil? && (aw = attribute(:await)) return await(aw, msg) end unless Condition.apply?(attribute(:if), attribute(:unless)) return do_reply_to_parent(h.applied_workitem) end pi = h.parent_id reply_immediately = false if attribute(:scope).to_s == 'true' h.variables ||= {} end if attribute(:forget).to_s == 'true' h.variables = compile_variables h.parent_id = nil h.forgotten = true reply_immediately = true elsif attribute(:lose).to_s == 'true' h.lost = true elsif msg['flanking'] or (attribute(:flank).to_s == 'true') h.flanking = true reply_immediately = true end if reply_immediately and pi @context.storage.put_msg( 'reply', 'fei' => pi, 'workitem' => Ruote.fulldup(h.applied_workitem), 'flanking' => h.flanking) end filter consider_tag consider_timers apply end |
#do_cancel(msg) ⇒ Object
The raw handling of messages passed to expressions (the fine handling is done in the #cancel method).
667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 |
# File 'lib/ruote/exp/flow_expression.rb', line 667 def do_cancel(msg) flavour = msg['flavour'] return if h.state == 'cancelling' && flavour != 'kill' # cancel on cancel gets discarded return if h.state == 'failed' && flavour == 'timeout' # do not timeout expressions that are "in error" (failed) h.state = case flavour when 'kill' then 'dying' when 'timeout' then 'timing_out' else 'cancelling' end if h.state == 'timing_out' h.applied_workitem['fields']['__timed_out__'] = [ h.fei, Ruote.now_to_utc_s, tree.first, compile_atts ] elsif h.state == 'cancelling' if t = msg['on_cancel'] h.on_cancel = t elsif ra_opts = msg['re_apply'] ra_opts = {} if ra_opts == true ra_opts['tree'] ||= tree h.on_re_apply = ra_opts end end cancel(flavour) end |
#do_fail(msg) ⇒ Object
Called when handling an on_error, will place itself in a 'failing' state and cancel the children (when the reply from the children comes back, the on_error will get triggered).
760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 |
# File 'lib/ruote/exp/flow_expression.rb', line 760 def do_fail(msg) h.state = 'failing' h.applied_workitem = msg['workitem'] if h.children.size < 1 reply_to_parent(h.applied_workitem) else flavour = msg['immediate'] ? 'kill' : nil persist_or_raise h.children.each do |i| @context.storage.put_msg('cancel', 'fei' => i, 'flavour' => flavour) end end end |
#do_pause(msg) ⇒ Object
Expression received a “pause” message. Will put the expression in the “paused” state and then pass the message to the children.
If the expression is in a non-nil state (failed, timed_out, …), the message will be ignored.
787 788 789 790 791 792 793 794 795 796 797 798 |
# File 'lib/ruote/exp/flow_expression.rb', line 787 def do_pause(msg) return if h.state != nil h.state = 'paused' do_persist || return h.children.each { |i| @context.storage.put_msg('pause', 'fei' => i) } unless msg['breakpoint'] end |
#do_persist ⇒ Object
Make sure to persist (retry if necessary).
106 107 108 109 |
# File 'lib/ruote/exp/ro_persist.rb', line 106 def do_persist do_p(true) end |
#do_reply(msg) ⇒ Object Also known as: do_receive
Wraps #reply (does the administrative part of the reply work).
606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 |
# File 'lib/ruote/exp/flow_expression.rb', line 606 def do_reply(msg) workitem = msg['workitem'] fei = workitem['fei'] removed = h.children.delete(fei) # accept without any check ? if msg['flanking'] (h.flanks ||= []) << fei if (not removed) # then it's a timer do_persist return end end if ut = msg['updated_tree'] ct = tree.dup ct.last[Ruote::FlowExpressionId.child_id(fei)] = ut update_tree(ct) end if h.state == 'paused' (h.paused_replies ||= []) << msg do_persist elsif h.state != nil # failing or timing out ... if h.children.size < 1 reply_to_parent(workitem) else #persist_or_raise # for the updated h.children do_persist end else # vanilla reply reply(workitem) end end |
#do_reply_to_parent(workitem, delete = true) ⇒ Object
The essence of the reply_to_parent job…
477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 |
# File 'lib/ruote/exp/flow_expression.rb', line 477 def do_reply_to_parent(workitem, delete=true) # propagate the cancel "flavour" back, so that one can know # why a branch got cancelled. flavour = if @msg.nil? nil elsif @msg['action'] == 'cancel' @msg['flavour'] || 'cancel' elsif h.state.nil? nil else @msg['flavour'] end # deal with the timers and the schedules %w[ timeout_schedule_id job_id ].each do |sid| @context.storage.delete_schedule(h[sid]) if h[sid] end # # legacy schedule ids, to be removed for ruote 2.4.0 @context.storage.delete_schedule(h.schedule_id) if h.schedule_id # # time-driven exps like cron, wait and once now all use h.schedule_id h.timers.each do |schedule_id, action| @context.storage.delete_schedule(schedule_id) end if h.timers # cancel flanking expressions if any cancel_flanks(h.state == 'dying' ? 'kill' : nil) # trigger or vanilla reply if h.state == 'failing' # on_error is implicit (#do_fail got called) trigger('on_error', workitem) elsif h.state == 'cancelling' && h.on_cancel trigger('on_cancel', workitem) elsif h.state == 'cancelling' && h.on_re_apply trigger('on_re_apply', workitem) elsif h.state == 'timing_out' && h.on_timeout trigger('on_timeout', workitem) elsif h.state == nil && h.on_reply trigger('on_reply', workitem) elsif h.flanking && h.state.nil? # # do vanish do_unpersist elsif h.lost && h.state.nil? # # do not reply, sit here (and wait for cancellation probably) do_persist elsif h.trigger && workitem['fields']["__#{h.trigger}__"] # # the "second take" trigger(h.trigger, workitem) else # vanilla reply filter(workitem) if h.state.nil? f = h.state.nil? && attribute(:vars_to_f) Ruote.set(workitem['fields'], f, h.variables) if f workitem['sub_wf_name'] = h.applied_workitem['sub_wf_name'] workitem['sub_wf_revision'] = h.applied_workitem['sub_wf_revision'] leave_tag(workitem) if h.tagname (do_unpersist || return) if delete # remove expression from storage if h.parent_id && ! h.attached @context.storage.put_msg( 'reply', 'fei' => h.parent_id, 'workitem' => workitem.merge!('fei' => h.fei), 'updated_tree' => h.updated_tree, # nil most of the time 'flavour' => flavour) else @context.storage.put_msg( (h.forgotten || h.attached) ? 'ceased' : 'terminated', 'wfid' => h.fei['wfid'], 'fei' => h.fei, 'workitem' => workitem, 'variables' => h.variables, 'flavour' => flavour) if h.state.nil? && h.on_terminate == 'regenerate' && ! (h.forgotten || h.attached) then @context.storage.put_msg( 'regenerate', 'wfid' => h.fei['wfid'], 'tree' => h.original_tree, 'workitem' => workitem, 'variables' => h.variables, 'flavour' => flavour) #'stash' => end end end end |
#do_resume(msg) ⇒ Object
Will “unpause” the expression (if it was paused), and trigger any 'paused_replies' (replies that came while the expression was paused).
803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 |
# File 'lib/ruote/exp/flow_expression.rb', line 803 def do_resume(msg) return unless h.state == 'paused' || h.state == 'awaiting' h.state = nil m = h.delete('paused_apply') return do_apply(m) if m # if it's a paused apply, pipe it directly to #do_apply replies = h.delete('paused_replies') || [] do_persist || return h.children.each { |i| @context.storage.put_msg('resume', 'fei' => i) } # resume children replies.each { |m| @context.storage.put_msg(m.delete('action'), m) } # trigger replies end |
#do_unpersist ⇒ Object
Make sure to unpersist (retry if necessary).
113 114 115 116 |
# File 'lib/ruote/exp/ro_persist.rb', line 113 def do_unpersist do_p(false) end |
#fei ⇒ Object
Returns the Ruote::FlowExpressionId for this expression.
146 147 148 149 |
# File 'lib/ruote/exp/flow_expression.rb', line 146 def fei Ruote::FlowExpressionId.new(h.fei) end |
#handle_on_error(msg, error) ⇒ Object
Looks up parent with on_error attribute and triggers it
68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 |
# File 'lib/ruote/exp/ro_on_x.rb', line 68 def handle_on_error(msg, error) return false if h.state == 'failing' err = deflate(error) oe_parent = lookup_on_error(err) return false unless oe_parent # no parent with on_error attribute found handler = oe_parent.local_on_error(err) return false if handler.to_s == '' # empty on_error handler nullifies ancestor's on_error workitem = msg['workitem'] workitem['fields']['__error__'] = err immediate = if handler.is_a?(String) !! handler.match(/^!/) elsif handler.is_a?(Array) !! handler.first.to_s.match(/^!/) else false end # NOTE: why not pass the handler in the msg? # no, because of HandlerEntry (not JSON serializable) @context.storage.put_msg( 'fail', 'fei' => oe_parent.h.fei, 'workitem' => workitem, 'immediate' => immediate) true # yes, error is being handled. end |
#has_attribute(*args) ⇒ Object Also known as: has_att
Given a list of attribute names, returns the first attribute name for which there is a value.
37 38 39 40 41 42 |
# File 'lib/ruote/exp/ro_attributes.rb', line 37 def has_attribute(*args) args.each { |a| a = a.to_s; return a if attributes[a] != nil } nil end |
#initial_persist ⇒ Object
Persists and fetches the _rev identifier from the storage.
Only used by the worker when creating the expression.
48 49 50 51 52 53 54 55 56 57 |
# File 'lib/ruote/exp/ro_persist.rb', line 48 def initial_persist r = @context.storage.put(@h, :update_rev => true) #t = Thread.current.object_id.to_s[-3..-1] #puts "+ per #{debug_id} #{tree[0]} r#{h._rev} t#{t} -> #{r.class}" #Ruote.p_caller('+ per') raise_or_return('initial_persist failed', r) end |
#is_concurrent? ⇒ Boolean
Concurrent expressions (expressions that apply more than one child at a time) are supposed to return true here.
233 234 235 236 |
# File 'lib/ruote/exp/flow_expression.rb', line 233 def is_concurrent? false end |
#iterative_var_lookup(k) ⇒ Object
TODO : rdoc rewrite needed
This method is mostly used by the worker when looking up a process name or participant name bound under a variable.
119 120 121 122 123 124 125 126 |
# File 'lib/ruote/exp/ro_variables.rb', line 119 def iterative_var_lookup(k) v = lookup_variable(k) return [ k, v ] unless (v.is_a?(String) or v.is_a?(Symbol)) iterative_var_lookup(v) end |
#launch_sub(pos, subtree, opts = {}) ⇒ Object
Launches a subprocesses (usually called from the #apply of certain expression implementations.
831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 |
# File 'lib/ruote/exp/flow_expression.rb', line 831 def launch_sub(pos, subtree, opts={}) i = h.fei.merge( 'subid' => Ruote.generate_subid(h.fei.inspect), 'expid' => pos) if ci = opts[:child_id] i['subid'] = "#{i['subid']}k#{ci}" end #p '=== launch_sub ===' #p [ :launcher, h.fei['expid'], h.fei['subid'], h.fei['wfid'] ] #p [ :launched, i['expid'], i['subid'], i['wfid'] ] forget = opts[:forget] register_child(i) unless forget variables = ( forget ? compile_variables : {} ).merge!(opts[:variables] || {}) @context.storage.put_msg( 'launch', 'fei' => i, 'parent_id' => forget ? nil : h.fei, 'tree' => subtree, 'workitem' => opts[:workitem] || h.applied_workitem, 'variables' => variables, 'forgotten' => forget) end |
#lookup_val(att_options = {}) ⇒ Object
96 97 98 99 100 101 102 103 |
# File 'lib/ruote/exp/ro_attributes.rb', line 96 def lookup_val(={}) lval( VV, s_cartesian(%w[ v var variable ], VV), s_cartesian(%w[ f fld field ], VV), ) end |
#lookup_val_prefix(prefix, att_options = {}) ⇒ Object
prefix = 'on' => will lookup on, on_val, on_value, on_v, on_var, on_variable, on_f, on_fld, on_field…
87 88 89 90 91 92 93 94 |
# File 'lib/ruote/exp/ro_attributes.rb', line 87 def lookup_val_prefix(prefix, ={}) lval( [ prefix ] + [ 'val', 'value' ].map { |s| "#{prefix}_#{s}" }, %w[ v var variable ].map { |s| "#{prefix}_#{s}" }, %w[ f fld field ].map { |s| "#{prefix}_#{s}" }, ) end |
#lookup_variable(var, prefix = nil) ⇒ Object Also known as: v, lv
Looks up the value of a variable in expression tree (seen from a leaf, it looks more like a stack than a tree)
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 81 82 83 84 85 |
# File 'lib/ruote/exp/ro_variables.rb', line 55 def lookup_variable(var, prefix=nil) var, prefix = split_prefix(var, prefix) if prefix == '//' return @context.storage.get_engine_variable(var) end if prefix == '/' && par = parent return par.lookup_variable(var, prefix) end if h.variables and Ruote.has_key?(h.variables, var) return Ruote.lookup(h.variables, var) end if h.parent_id && h.parent_id['engine_id'] == @context.engine_id # # do not lookup variables in a remote engine ... (return parent.lookup_variable(var, prefix)) rescue nil # if the lookup fails (parent gone) then rescue and let go end k = var.split('.').first vars = { k => @context.storage.get_engine_variable(k) } # # engine var lookup might be expensive, only lookup the var we need Ruote.lookup(vars, var) end |
#name ⇒ Object
Returns the name of this expression, like 'sequence', 'participant', 'cursor', etc…
912 913 914 915 |
# File 'lib/ruote/exp/flow_expression.rb', line 912 def name tree[0] end |
#parent ⇒ Object
Fetches the parent expression, or returns nil if there is no parent expression.
180 181 182 183 184 185 |
# File 'lib/ruote/exp/flow_expression.rb', line 180 def parent h.parent_id ? Ruote::Exp::FlowExpression.fetch(@context, h.parent_id) : nil end |
#parent_id ⇒ Object
Returns the Ruote::FlowExpressionIf of the parent expression, or nil if there is no parent expression.
162 163 164 165 166 167 |
# File 'lib/ruote/exp/flow_expression.rb', line 162 def parent_id h.parent_id ? Ruote::FlowExpressionId.new(h.parent_id) : nil end |
#pause_on_apply(msg) ⇒ Object
Called by #do_apply when msg == 'paused'. Covers the “apply/launch it but it's immediately paused” case. Freezes the apply message in h.paused_apply and saves the expression.
421 422 423 424 425 426 427 428 429 |
# File 'lib/ruote/exp/flow_expression.rb', line 421 def pause_on_apply(msg) msg['state'] = nil h.state = 'paused' h.paused_apply = msg persist_or_raise end |
#persist_or_raise ⇒ Object Also known as: persist
91 92 93 94 |
# File 'lib/ruote/exp/ro_persist.rb', line 91 def persist_or_raise p_or_raise(true) end |
#reply(workitem) ⇒ Object
A default implementation for all the expressions.
659 660 661 662 |
# File 'lib/ruote/exp/flow_expression.rb', line 659 def reply(workitem) reply_to_parent(workitem) end |
#reply_to_parent(workitem, delete = true) ⇒ Object
FlowExpression call this method when they're done and they want their parent expression to take over (it will end up calling the #reply of the parent expression).
Expression implementations are free to override this method. The common behaviour is in #do_reply_to_parent.
470 471 472 473 |
# File 'lib/ruote/exp/flow_expression.rb', line 470 def reply_to_parent(workitem, delete=true) do_reply_to_parent(workitem, delete) end |
#root(stubborn = false) ⇒ Object
An expensive method, looks up all the expressions with the same wfid in the storage (for some storages this is not expensive at all), and determine the root of this expression. It does this by recursively going from an expression to its parent, starting with this expression. The root is when the parent can't be reached.
By default, this method will always return an expression, but if stubborn is set to true and the top expression points to a gone parent then nil will be returned. The default (stubborn=true) is probably what you want anyway.
199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 |
# File 'lib/ruote/exp/flow_expression.rb', line 199 def root(stubborn=false) previous = nil current = @h exps = @context.storage.find_expressions( h.fei['wfid'] ).each_with_object({}) { |exp, h| h[exp['fei']] = exp } while current && current['parent_id'] previous = current current = exps[current['parent_id']] end current ||= previous unless stubborn current ? Ruote::Exp::FlowExpression.from_h(@context, current) : nil end |
#root_id(stubborn = false) ⇒ Object
Returns the fei of the root expression of this expression. The result is an instance of Ruote::FlowExpressionId.
Uses #root behind the scenes, hence the stubborn option.
225 226 227 228 |
# File 'lib/ruote/exp/flow_expression.rb', line 225 def root_id(stubborn=false) root(stubborn).fei end |
#set_variable(var, val, override = false) ⇒ Object
Sets a variable to a given value. (will set at the appropriate level).
98 99 100 101 102 103 |
# File 'lib/ruote/exp/ro_variables.rb', line 98 def set_variable(var, val, override=false) fexp, v = locate_set_var(var, override) || locate_var(var) fexp.un_set_variable(:set, v, val, (fexp.h.fei != h.fei)) if fexp end |
#tree ⇒ Object
Returns the current version of the tree (returns the updated version if it got updated.
883 884 885 886 |
# File 'lib/ruote/exp/flow_expression.rb', line 883 def tree h.updated_tree || h.original_tree end |
#tree_children ⇒ Object
Returns the “AST” view on the children of this expression…
927 928 929 930 |
# File 'lib/ruote/exp/flow_expression.rb', line 927 def tree_children tree[2] end |
#try_persist ⇒ Object
59 60 61 62 63 64 65 66 67 68 69 |
# File 'lib/ruote/exp/ro_persist.rb', line 59 def try_persist r = @context.storage.put(@h) #t = Thread.current.object_id.to_s[-3..-1] #puts "+ per #{debug_id} #{tree[0]} r#{h._rev} t#{t} -> #{r.class}" #p self.h.children.collect { |i| Ruote.sid(i) } #Ruote.p_caller('+ per') r end |
#try_unpersist ⇒ Object
71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
# File 'lib/ruote/exp/ro_persist.rb', line 71 def try_unpersist r = @context.storage.delete(@h) #t = Thread.current.object_id.to_s[-3..-1] #puts "- unp #{debug_id} #{tree[0]} r#{h._rev} t#{t} -> #{r.class}" #Ruote.p_caller('- unp') return r if r #if h.has_error err = @context.storage.get('errors', "err_#{Ruote.to_storage_id(h.fei)}") @context.storage.delete(err) if err #end # removes any error in the journal for this expression # since it will now be gone, no need to keep track of its errors nil end |
#unpersist_or_raise ⇒ Object Also known as: unpersist
96 97 98 99 |
# File 'lib/ruote/exp/ro_persist.rb', line 96 def unpersist_or_raise p_or_raise(false) end |
#unset_variable(var, override = false) ⇒ Object
Unbinds a variables.
107 108 109 110 111 112 |
# File 'lib/ruote/exp/ro_variables.rb', line 107 def unset_variable(var, override=false) fexp, v = locate_set_var(var, override) || locate_var(var) fexp.un_set_variable(:unset, v, nil, (fexp.h.fei != h.fei)) if fexp end |
#update_tree(t = nil) ⇒ Object
Updates the tree of this expression
update_tree(t)
will set the updated tree to t
update_tree
will copy (deep copy) the original tree as the updated_tree.
Adding a child to a sequence expression :
seq.update_tree
seq.updated_tree[2] << [ 'participant', { 'ref' => 'bob' }, [] ]
seq.do_persist
904 905 906 907 |
# File 'lib/ruote/exp/flow_expression.rb', line 904 def update_tree(t=nil) h.updated_tree = t || Ruote.fulldup(h.original_tree) end |
#variables ⇒ Object
A shortcut to the variables held in the expression (nil if none held).
35 36 37 38 |
# File 'lib/ruote/exp/ro_variables.rb', line 35 def variables @h['variables'] end |
#wfid ⇒ Object
Returns the workflow instance id of the workflow this expression belongs to.
154 155 156 157 |
# File 'lib/ruote/exp/flow_expression.rb', line 154 def wfid h.fei['wfid'] end |