Skip to content

Commit 494fcca

Browse files
abhide-tibcoabhijitWakchaure
abhide-tibco
and
abhijitWakchaure
authored
Flow rerun with io (#54)
* update go mod for flow * update queries for flow state and flow rerun * Fixed go mod file * Removed vendor folder * add queries for flow rerun count * Updated go mod * Fixed error * add vendor to .gitignore * add config.json to .gitignore * add config.json to .gitignore * add flow rerun count and flow input * Updated go mod * Added support for existing tables Dont kill the service immediately but keep running * Added checks for database connected * Fixed error * Fixed max connection issue --------- Co-authored-by: Abhijit Wakchaure <[email protected]>
1 parent fd4768a commit 494fcca

File tree

9 files changed

+325
-87
lines changed

9 files changed

+325
-87
lines changed

‎.gitignore

+2
Original file line numberDiff line numberDiff line change
@@ -47,3 +47,5 @@ tags
4747
.build-cache
4848
submodules/flogo-cicd/.build-cache
4949
./Dockerfile
50+
vendor/
51+
flow-state/config.json

‎flow-state/go.mod

+14-6
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,20 @@
11
module github.com/project-flogo/services/flow-state
22

3-
go 1.16
3+
go 1.18
44

55
require (
6-
github.com/gorilla/websocket v1.4.2
6+
github.com/gorilla/websocket v1.5.0
77
github.com/julienschmidt/httprouter v1.3.0
8-
github.com/lib/pq v1.10.3
9-
github.com/project-flogo/core v1.6.0
10-
github.com/project-flogo/flow v1.6.0
11-
github.com/rs/cors v1.8.0
8+
github.com/lib/pq v1.10.7
9+
github.com/project-flogo/core v1.6.4
10+
github.com/project-flogo/flow v1.6.5-0.20230324065406-53d6cf9cc418
11+
github.com/rs/cors v1.8.3
12+
)
13+
14+
require (
15+
github.com/araddon/dateparse v0.0.0-20190622164848-0fb0a474d195 // indirect
16+
github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect
17+
go.uber.org/atomic v1.6.0 // indirect
18+
go.uber.org/multierr v1.5.0 // indirect
19+
go.uber.org/zap v1.16.0 // indirect
1220
)

‎flow-state/go.sum

+10-30
Original file line numberDiff line numberDiff line change
@@ -5,51 +5,35 @@ github.com/araddon/dateparse v0.0.0-20190622164848-0fb0a474d195/go.mod h1:SLqhdZ
55
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
66
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
77
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
8-
github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
9-
github.com/gin-gonic/gin v1.5.0/go.mod h1:Nd6IXA8m5kNZdNEHMBd93KT+mdY3+bewLgRvmCsR2Do=
10-
github.com/go-playground/locales v0.12.1/go.mod h1:IUMDtCfWo/w/mtMfIE/IG2K+Ey3ygWanZIBtBW0W2TM=
11-
github.com/go-playground/universal-translator v0.16.0/go.mod h1:1AnU7NaIRDWWzGEKwgtJRd2xk99HeFyHw3yid4rvQIY=
12-
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
13-
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
148
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
15-
github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
16-
github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
17-
github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
9+
github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
10+
github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
1811
github.com/julienschmidt/httprouter v1.3.0 h1:U0609e9tgbseu3rBINet9P48AI/D3oJs4dN7jwJOQ1U=
1912
github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
2013
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
21-
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
2214
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
2315
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
24-
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
2516
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
26-
github.com/leodido/go-urn v1.1.0/go.mod h1:+cyI34gQWZcE1eQU7NVgKkkzdXDQHr1dBMtdAPozLkw=
27-
github.com/lib/pq v1.10.3 h1:v9QZf2Sn6AmjXtQeFpdoq/eaNtYP6IN+7lcrygsIAtg=
28-
github.com/lib/pq v1.10.3/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
29-
github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
30-
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
31-
github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
17+
github.com/lib/pq v1.10.7 h1:p7ZhMD+KsSRozJr34udlUrhboJwWAgCg34+/ZZNvZZw=
18+
github.com/lib/pq v1.10.7/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
3219
github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 h1:RWengNIwukTxcDr9M+97sNutRR1RKhG96O6jWumTTnw=
3320
github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826/go.mod h1:TaXosZuwdSHYgviHp1DAtfrULt5eUgsSMsZf+YrPgl8=
3421
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
3522
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
3623
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
3724
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
3825
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
39-
github.com/project-flogo/core v1.4.0/go.mod h1:fapTXUhLxDeAHyb6eMkuwnYswO8FpZJAMat055QVdJE=
40-
github.com/project-flogo/core v1.6.0 h1:+LH226SGU5961xh5H9lp8iUGp5dWPHaMYGu0pTuSLSE=
41-
github.com/project-flogo/core v1.6.0/go.mod h1:fapTXUhLxDeAHyb6eMkuwnYswO8FpZJAMat055QVdJE=
42-
github.com/project-flogo/flow v1.6.0 h1:04La0Tp0QcIuXRXPPncYZ9XwDMLlDdZWxxzQaw92MYg=
43-
github.com/project-flogo/flow v1.6.0/go.mod h1:LesZqWPKWub4EZ71+kib74n5MF7ePlVPAuP0DVtRhkI=
26+
github.com/project-flogo/core v1.6.4 h1:MG4XNXqEmmY3qkPyUOmq5Ph0m8ycknnKcFqsH8avPa0=
27+
github.com/project-flogo/core v1.6.4/go.mod h1:fapTXUhLxDeAHyb6eMkuwnYswO8FpZJAMat055QVdJE=
28+
github.com/project-flogo/flow v1.6.5-0.20230324065406-53d6cf9cc418 h1:jB0fcOjaq5FsgMYKdqVuh+k4JN/T+WEtIdhrVZWh32A=
29+
github.com/project-flogo/flow v1.6.5-0.20230324065406-53d6cf9cc418/go.mod h1:8b0r4IuPb54MFOxKhyq3sPDm080Hzgtv2yVdAA6aEg0=
4430
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
45-
github.com/rs/cors v1.8.0 h1:P2KMzcFwrPoSjkF1WLRPsp3UMLyql8L4v9hQpVeK5so=
46-
github.com/rs/cors v1.8.0/go.mod h1:EBwu+T5AvHOcXwvZIkQFjUN6s8Czyqw12GL/Y0tUyRM=
31+
github.com/rs/cors v1.8.3 h1:O+qNyWn7Z+F9M0ILBHgMVPuB1xTOucVd5gtaYyXBpRo=
32+
github.com/rs/cors v1.8.3/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
4733
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
4834
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
4935
github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
5036
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
51-
github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
52-
github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
5337
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
5438
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
5539
github.com/xeipuuv/gojsonschema v1.1.0/go.mod h1:5yf86TLmAcydyeJq5YvxkGPE2fm/u4myDekKRoLuqhs=
@@ -72,7 +56,6 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
7256
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
7357
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
7458
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
75-
golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
7659
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
7760
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
7861
golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
@@ -81,11 +64,8 @@ golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64
8164
golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
8265
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
8366
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
84-
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
8567
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
8668
gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
87-
gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE=
88-
gopkg.in/go-playground/validator.v9 v9.29.1/go.mod h1:+c9/zcJMFNgbLvly1L1V+PpxWdVbfP1avr/N00E2vyQ=
8969
gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
9070
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
9171
honnef.co/go/tools v0.0.1-2019.2.3 h1:3JgtbtFHMiCmsznwGVTUWbgGov+pVqnlf1dEJTNAXeM=

‎flow-state/server/rest/endpoints.go

+8-5
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
flowEvent "github.com/project-flogo/flow/support/event"
77
"github.com/project-flogo/services/flow-state/event"
88
"github.com/project-flogo/services/flow-state/store/metadata"
9+
"github.com/project-flogo/services/flow-state/store/postgres"
910
"io/ioutil"
1011
"net/http"
1112
"strconv"
@@ -98,11 +99,13 @@ func (se *ServiceEndpoints) getHealthCheck(response http.ResponseWriter, request
9899
se.logger.Debugf("Endpoint[GET:/health] : Called")
99100
switch request.Method {
100101
case http.MethodGet:
101-
if se.stepStore.Status() {
102-
response.WriteHeader(http.StatusOK)
103-
} else {
104-
se.logger.Info("Health check status failed")
105-
response.WriteHeader(515)
102+
103+
status := se.stepStore.Status().(*postgres.DBDetails)
104+
105+
response.Header().Set("Content-Type", "application/json")
106+
response.WriteHeader(http.StatusOK)
107+
if err := json.NewEncoder(response).Encode(status); err != nil {
108+
se.logger.Error(err.Error())
106109
}
107110
default:
108111
response.WriteHeader(http.StatusMethodNotAllowed)

‎flow-state/store/mem/step.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ type StepStore struct {
2525
snapshots sync.Map
2626
}
2727

28-
func (s *StepStore) Status() bool {
28+
func (s *StepStore) Status() interface{} {
2929
//TODO
3030
return true
3131
}

‎flow-state/store/postgres/connection.go

+5
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,11 @@ func decodeTLSParam(tlsparm string) string {
6565
func NewDB(settings map[string]interface{}) (*sql.DB, error) {
6666
var err error
6767

68+
dbMaxConn := settings["maxopenconnection"]
69+
_, err = strconv.Atoi(dbMaxConn.(string))
70+
if err != nil {
71+
settings["maxopenconnection"] = "0"
72+
}
6873
s := &pgConnection{}
6974
err = metadata.MapToStruct(settings, s, false)
7075

‎flow-state/store/postgres/operation.go

+53-11
Original file line numberDiff line numberDiff line change
@@ -14,24 +14,53 @@ import (
1414
)
1515

1616
const (
17-
STEP_INSERT = "INSERT INTO steps (flowinstanceid, stepid, taskname, status, starttime, endtime, stepdata) VALUES ($1,$2,$3,$4,$5,$6,$7);"
18-
SNAPSHOT_INSERT = "INSERT INTO snapshopt (flowinstanceid, hostid, stepid, starttime, endtime, stepdata) VALUES ($1,$2,$3,$4,$5,$6);"
19-
FlowState_UPSERT_RERUN = "INSERT INTO flowstate (flowInstanceId, userId, appName,appVersion, flowName, hostId,startTime,endTime,status) VALUES ($1,$2,$3,$4,$5,$6,$7,$8, $9) ON CONFLICT (flowinstanceid) DO UPDATE SET hostId = EXCLUDED.hostId, flowName = EXCLUDED.flowName, userId = EXCLUDED.userId, status = EXCLUDED.status, starttime=EXCLUDED.starttime,endtime= EXCLUDED.endtime;\n"
20-
UpdateFlowState = "UPDATE flowstate set endtime=$1,status=$2, executiontime=ROUND( ((EXTRACT(EPOCH FROM ($1 - starttime)))*1000) :: numeric , 3) where flowinstanceid = $3;"
17+
STEP_INSERT = "INSERT INTO steps (flowinstanceid, stepid, taskname, status, starttime, endtime, stepdata) VALUES ($1,$2,$3,$4,$5,$6,$7);"
18+
SNAPSHOT_INSERT = "INSERT INTO snapshopt (flowinstanceid, hostid, stepid, starttime, endtime, stepdata) VALUES ($1,$2,$3,$4,$5,$6);"
19+
20+
FlowState_UPSERT_RERUN_v1 = "INSERT INTO flowstate (flowInstanceId, userId, appName,appVersion, flowName, hostId, startTime, endTime, status, rerunofflowinstanceid) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT (flowinstanceid) DO UPDATE SET hostId = EXCLUDED.hostId, flowName = EXCLUDED.flowName, userId = EXCLUDED.userId, status = EXCLUDED.status, starttime=EXCLUDED.starttime,endtime= EXCLUDED.endtime;"
21+
FlowState_UPSERT_RERUN_v2 = "INSERT INTO flowstate (flowInstanceId, userId, appName,appVersion, flowName, hostId, flowInput, flowOutput, rerunCount, startTime, endTime, status, rerunofflowinstanceid) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT (flowinstanceid) DO UPDATE SET hostId = EXCLUDED.hostId, flowName = EXCLUDED.flowName, userId = EXCLUDED.userId, status = EXCLUDED.status, flowInput = EXCLUDED.flowInput, flowOutput = EXCLUDED.flowOutput, rerunCount = EXCLUDED.rerunCount, starttime=EXCLUDED.starttime,endtime= EXCLUDED.endtime;"
22+
UpdateFlowState_v1 = "UPDATE flowstate set endtime=$1,status=$2, executiontime=ROUND( ((EXTRACT(EPOCH FROM ($1 - starttime)))*1000) :: numeric , 3) where flowinstanceid = $3;"
23+
UpdateFlowState_v2 = "UPDATE flowstate set endtime=$1, status=$2, flowOutput=$3, executiontime=ROUND( ((EXTRACT(EPOCH FROM ($1 - starttime)))*1000) :: numeric , 3) where flowinstanceid = $4;"
2124

2225
UpsertSteps = "INSERT INTO steps (flowinstanceid, stepid, taskname, status, starttime, endtime, stepdata, subflowid, flowname, rerun) VALUES($1,$2,$3,$4,$5,$6,$7, $8, $9, $10) ON CONFLICT (flowinstanceid, stepid) DO UPDATE SET status = EXCLUDED.status, starttime=EXCLUDED.starttime,endtime= EXCLUDED.endtime,stepdata=EXCLUDED.stepdata;\n"
2326
DeleteSteps = "DELETE from steps where flowinstanceid = $1 and CAST(stepid as INTEGER) >= $2"
2427

25-
UpsertAppState = "INSERT INTO appstate (userId, appName, persistenceenabled) VALUES($1,$2,$3) ON CONFLICT (userId, appName) DO UPDATE SET persistenceenabled = EXCLUDED.persistenceenabled ;\n"
28+
UpsertAppState = "INSERT INTO appstate (userId, appName, persistenceenabled) VALUES($1,$2,$3) ON CONFLICT (userId, appName) DO UPDATE SET persistenceenabled = EXCLUDED.persistenceenabled ;\n"
29+
IncrementRerunCount = "UPDATE flowstate SET reruncount = reruncount + 1 WHERE flowinstanceid = $1 RETURNING reruncount;"
2630
)
2731

2832
type StatefulDB struct {
29-
db *sql.DB
33+
db *sql.DB
34+
dbDetails *DBDetails
3035
}
3136

3237
func (s *StatefulDB) InsertFlowState(flowState *state.FlowState) (results *ResultSet, err error) {
33-
inputArgs := []interface{}{flowState.FlowInstanceId, flowState.UserId, flowState.AppName, flowState.AppVersion, flowState.FlowName, flowState.HostId, flowState.StartTime, flowState.EndTime, flowState.FlowStats}
34-
return s.insert(FlowState_UPSERT_RERUN, inputArgs)
38+
var flowInputs, flowOutputs []byte
39+
if flowState.FlowInputs == nil {
40+
flowState.FlowInputs = make(map[string]interface{})
41+
}
42+
flowInputs, _ = json.Marshal(flowState.FlowInputs)
43+
44+
//if flowState.FlowOutputs != nil {
45+
// flowOutputs, _ = json.Marshal(flowState.FlowOutputs)
46+
//}
47+
if flowState.OriginalInstanceId != "" {
48+
// update original instance Id and increment rerun count
49+
if s.dbDetails.SmVersion == "2.0" {
50+
_, err = s.update(IncrementRerunCount, []interface{}{flowState.OriginalInstanceId})
51+
if err != nil {
52+
return nil, err
53+
}
54+
}
55+
}
56+
57+
if s.dbDetails.SmVersion == "1.0" {
58+
inputArgs := []interface{}{flowState.FlowInstanceId, flowState.UserId, flowState.AppName, flowState.AppVersion, flowState.FlowName, flowState.HostId, flowState.StartTime, flowState.EndTime, flowState.FlowStats, flowState.OriginalInstanceId}
59+
return s.insert(FlowState_UPSERT_RERUN_v1, inputArgs)
60+
}
61+
inputArgs := []interface{}{flowState.FlowInstanceId, flowState.UserId, flowState.AppName, flowState.AppVersion, flowState.FlowName, flowState.HostId, flowInputs, flowOutputs, flowState.RerunCount, flowState.StartTime, flowState.EndTime, flowState.FlowStats, flowState.OriginalInstanceId}
62+
return s.insert(FlowState_UPSERT_RERUN_v2, inputArgs)
63+
3564
}
3665

3766
func (s *StatefulDB) InsertAppState(appStatedata *metadata.Metadata) (results *ResultSet, err error) {
@@ -40,8 +69,21 @@ func (s *StatefulDB) InsertAppState(appStatedata *metadata.Metadata) (results *R
4069
}
4170

4271
func (s *StatefulDB) UpdateFlowState(flowState *state.FlowState) (results *ResultSet, err error) {
43-
inputArgs := []interface{}{flowState.EndTime, flowState.FlowStats, flowState.FlowInstanceId}
44-
return s.insert(UpdateFlowState, inputArgs)
72+
var flowOutputs []byte
73+
//if flowState.FlowInputs != nil {
74+
// flowInputs, _ = json.Marshal(flowState.FlowInputs)
75+
//}
76+
if flowState.FlowOutputs != nil {
77+
flowOutputs, _ = json.Marshal(flowState.FlowOutputs)
78+
}
79+
80+
if s.dbDetails.SmVersion == "1.0" {
81+
inputArgs := []interface{}{flowState.EndTime, flowState.FlowStats, flowState.FlowInstanceId}
82+
return s.insert(UpdateFlowState_v1, inputArgs)
83+
}
84+
inputArgs := []interface{}{flowState.EndTime, flowState.FlowStats, flowOutputs, flowState.FlowInstanceId}
85+
return s.insert(UpdateFlowState_v2, inputArgs)
86+
4587
}
4688

4789
func (s *StatefulDB) InsertSteps(step *state.Step) (results *ResultSet, err error) {
@@ -173,7 +215,7 @@ func (s *StatefulDB) delete(deleteSql string, inputArgs []interface{}) (results
173215
return UnmarshalRows(rows)
174216
}
175217

176-
//GetStatement
218+
// GetStatement
177219
func (s *StatefulDB) getStepStatement(prepared string) (stmt *sql.Stmt, err error) {
178220
preparedQueryCacheMutex.Lock()
179221
defer preparedQueryCacheMutex.Unlock()

0 commit comments

Comments
 (0)