harn-stdlib 0.10.177

Embedded Harn standard library source catalog
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

pub type WorkflowJoinPolicy = {strategy?: string, min_completed?: int?, require_all_inputs?: bool}

pub type WorkflowScheduleEdge = {from: string, to: string, branch?: string?, label?: string?}

type WorkflowScheduleGraph = {edges?: list<WorkflowScheduleEdge>, ...dict}

type WorkflowJoinConfig = {
  node_id?: string,
  node?: {join_policy?: WorkflowJoinPolicy, ...dict},
  graph?: WorkflowScheduleGraph,
  completed_nodes?: list<string>,
}

pub type WorkflowJoinReadiness = {ready: bool, required: int, completed_inputs: int}

type WorkflowNextEdgesConfig = {graph?: WorkflowScheduleGraph, current?: string, branch?: string?}

fn __schedule_incoming_edges(
  graph: WorkflowScheduleGraph,
  node_id: string,
) -> list<WorkflowScheduleEdge> {
  let incoming: list<WorkflowScheduleEdge> = []
  for edge in graph.edges ?? [] {
    if edge.to == node_id {
      incoming.push(edge)
    }
  }
  return incoming
}

fn __schedule_required_inputs(policy: WorkflowJoinPolicy, incoming_count: int) -> int {
  if policy.min_completed != nil {
    return policy.min_completed
  }
  if (policy.require_all_inputs ?? false) || (policy.strategy ?? "all") == "all" {
    return incoming_count
  }
  return 1
}

/**
 * workflow_join_readiness.
 *
 * @effects: []
 * @errors: []
 * @api_stability: experimental
 */
pub fn workflow_join_readiness(config: WorkflowJoinConfig? = nil) -> WorkflowJoinReadiness {
  const options = config ?? {}
  const node = options.node ?? {}
  const graph = options.graph ?? {}
  const completed_nodes = options.completed_nodes ?? []
  const incoming = __schedule_incoming_edges(graph, options.node_id ?? "")
  const required = __schedule_required_inputs(node.join_policy ?? {}, len(incoming))
  let completed_inputs = 0
  for edge in incoming {
    if contains(completed_nodes, edge.from) {
      completed_inputs = completed_inputs + 1
    }
  }
  return {
    ready: completed_inputs >= required,
    required: required,
    completed_inputs: completed_inputs,
  }
}

/**
 * workflow_next_edges.
 *
 * @effects: []
 * @errors: []
 * @api_stability: experimental
 */
pub fn workflow_next_edges(config: WorkflowNextEdgesConfig? = nil) -> list<WorkflowScheduleEdge> {
  const options = config ?? {}
  const graph = options.graph ?? {}
  const current = options.current ?? ""
  const branch = options.branch
  let exact: list<WorkflowScheduleEdge> = []
  let default_edges: list<WorkflowScheduleEdge> = []
  for edge in graph.edges ?? [] {
    if edge.from != current {
      continue
    }
    if edge.branch == branch {
      exact.push(edge)
    }
    if edge.branch == nil {
      default_edges.push(edge)
    }
  }
  if len(exact) > 0 || branch == nil {
    return exact
  }
  return default_edges
}