From 820315de8ff7a63a6bfe1fc78d8d428d720e4310 Mon Sep 17 00:00:00 2001
From: Frank Martinez <fmartinez@tibc.com>
Date: Wed, 1 May 2019 10:52:48 -0400
Subject: [PATCH] update go mod; add activity cleanup

---
 go.mod                 |  5 ++++-
 go.sum                 |  4 ++--
 pipeline/definition.go | 18 ++++++++++++++++++
 3 files changed, 24 insertions(+), 3 deletions(-)

diff --git a/go.mod b/go.mod
index b96ac6b..ab0e811 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,9 @@
 module github.com/project-flogo/stream
 
 require (
-	github.com/project-flogo/core v0.9.0-beta.2
+	github.com/pkg/errors v0.8.1 // indirect
+	github.com/project-flogo/core v0.9.0-rc.1
 	github.com/stretchr/testify v1.3.0
+	go.uber.org/atomic v1.3.2 // indirect
+	go.uber.org/multierr v1.1.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 7e5bc07..a576543 100644
--- a/go.sum
+++ b/go.sum
@@ -4,8 +4,8 @@ github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/project-flogo/core v0.9.0-beta.2 h1:L5VLOfF3YfAFVJwn/tL3x0835sZIedrHcrPkYxHbPl4=
-github.com/project-flogo/core v0.9.0-beta.2/go.mod h1:dzmBbQfNNC0g0KClKYQxxGJLe53MHafg75Vhmw2TW8U=
+github.com/project-flogo/core v0.9.0-rc.1 h1:0dALJoxHI/T4Ao340cHfNg5Yf5uM/bT4Y7u+5UYXfIQ=
+github.com/project-flogo/core v0.9.0-rc.1/go.mod h1:dzmBbQfNNC0g0KClKYQxxGJLe53MHafg75Vhmw2TW8U=
 github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
diff --git a/pipeline/definition.go b/pipeline/definition.go
index e66dfa5..abd9ef3 100644
--- a/pipeline/definition.go
+++ b/pipeline/definition.go
@@ -1,9 +1,12 @@
 package pipeline
 
 import (
+	"github.com/project-flogo/core/activity"
 	"github.com/project-flogo/core/data/mapper"
 	"github.com/project-flogo/core/data/metadata"
 	"github.com/project-flogo/core/data/resolve"
+	"github.com/project-flogo/core/support"
+	"github.com/project-flogo/core/support/log"
 )
 
 type DefinitionConfig struct {
@@ -43,3 +46,18 @@ func (d *Definition) Metadata() *metadata.IOMetadata {
 func (d *Definition) Name() string {
 	return d.name
 }
+
+func (d *Definition) Cleanup() error {
+	for _, stage := range d.stages {
+		if !activity.IsSingleton(stage.act) {
+			if needsCleanup, ok := stage.act.(support.NeedsCleanup); ok {
+				err := needsCleanup.Cleanup()
+				if err != nil {
+					log.RootLogger().Warnf("Error cleaning up activity '%s' in pipeline '%s' : ", activity.GetRef(stage.act), d.name, err)
+				}
+			}
+		}
+	}
+
+	return nil
+}