Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions workflow/builder.go
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,7 @@ func (wb *Builder) AddDirectEdge(source ExecutorBinding, target ExecutorBinding,
Index: wb.edgeIdx(),
}
applyEdgeOptions(&edge, opts)
wb.edges[source.ID] = append(wb.edges[source.ID], edge)
wb.addEdgeForSource(source.ID, edge)
wb.conditionlessConnections = append(wb.conditionlessConnections, conn)
return wb
}
Expand Down Expand Up @@ -176,7 +176,7 @@ func (wb *Builder) AddFanOutEdge(source ExecutorBinding, targets []ExecutorBindi
Index: wb.edgeIdx(),
}
applyEdgeOptions(&edge, opts)
wb.edges[source.ID] = append(wb.edges[source.ID], edge)
wb.addEdgeForSource(source.ID, edge)
return wb
}

Expand Down Expand Up @@ -205,7 +205,7 @@ func (wb *Builder) AddFanInBarrierEdge(sources []ExecutorBinding, target Executo
}
applyEdgeOptions(&edge, opts)
for _, id := range sourceIDs {
wb.edges[id] = append(wb.edges[id], edge)
wb.addEdgeForSource(id, edge)
}
return wb
}
Expand Down Expand Up @@ -354,6 +354,10 @@ func (wb *Builder) edgeIdx() int {
return wb.edgeCount
}

func (wb *Builder) addEdgeForSource(sourceID string, edge Edge) {
wb.edges[sourceID] = append(wb.edges[sourceID], edge)
}

// validateTypeCompatibility checks that sent message types of source executors
// are compatible with input types of target executors for every edge. It
// creates temporary executor instances to inspect their protocols. If an
Expand Down
Loading