Merge pull request #20 from fm-tibco/master

update go mod; add activity cleanup

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
 )

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=

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
+}