From 9b88fa39b37965b22c8f5da70992637e9d9d2579 Mon Sep 17 00:00:00 2001
From: Jon Seymour <jon@wildducktheories.com>
Date: Wed, 13 Apr 2016 22:41:13 +1000
Subject: [PATCH] PR #443: Add support to a Validation phase to the task
 lifecycle.

It is better if we can detect and report configuration errors of task
pipeline early, during task definition, rather than deferring such
checks to runtime (when this is possible).

To support such checks, we add a Validate() error method to
the pipeline.Node() interface which gives pipeline nodes a chance
to validate their configuration. Any pipeline containing nodes
that do not pass these test cannot be saved.

Signed-off-by: Jon Seymour <jon@wildducktheories.com>
---
 pipeline/node.go     | 7 +++++++
 pipeline/pipeline.go | 6 ++++++
 2 files changed, 13 insertions(+)

diff --git a/pipeline/node.go b/pipeline/node.go
index 29257f737..970291a95 100644
--- a/pipeline/node.go
+++ b/pipeline/node.go
@@ -61,6 +61,9 @@ type Node interface {
 	// The type of output the node provides.
 	Provides() EdgeType
 
+	// Check that the definition of the node is consistent
+	validate() error
+
 	// Helper methods for walking DAG
 	tMark() bool
 	setTMark(b bool)
@@ -167,6 +170,10 @@ func (n *node) Provides() EdgeType {
 	return n.provides
 }
 
+func (n *node) validate() error {
+	return nil
+}
+
 func (n *node) dot(buf *bytes.Buffer) {
 	for _, c := range n.children {
 		buf.Write([]byte(fmt.Sprintf("%s -> %s;\n", n.Name(), c.Name())))
diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go
index 63ffedf66..9c4cc127e 100644
--- a/pipeline/pipeline.go
+++ b/pipeline/pipeline.go
@@ -60,6 +60,12 @@ func CreatePipeline(script string, sourceEdge EdgeType, scope *tick.Scope, deadm
 			return nil, fmt.Errorf("source edge type must be either Stream or Batch not %s", sourceEdge)
 		}
 	}
+	if err = p.Walk(
+		func(n Node) error {
+			return n.validate()
+		}); err != nil {
+		return nil, err
+	}
 	return p, nil
 }