unistack-org/micro
4
0
Fork 2
Code Issues 18 Pull Requests Actions Packages Projects Releases 324 Wiki Activity

fixup tests
All checks were successful
lint / lint (pull_request) Successful in 1m10s
Details
pr / test (pull_request) Successful in 1m11s
Details

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2023-10-16 19:07:36 +03:00
parent ebd7eab627
commit 357d6c7b99
4 changed files with 14 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
flow/default.go
View File

@@ -190,7 +190,7 @@ func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...optio
steps, err := w.getSteps(options.Start, options.Reverse)
if err != nil {
if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusPending.String())}); werr != nil {
w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
w.opts.Logger.Error(w.opts.Context, "store write error", "error", werr.Error())
}
return "", err
}
@@ -214,7 +214,7 @@ func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...optio
done := make(chan struct{})
if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusRunning.String())}); werr != nil {
w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
w.opts.Logger.Error(w.opts.Context, "store write error", "error", werr.Error())
return eid, werr
}
for idx := range steps {
@@ -239,7 +239,7 @@ func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...optio
return
}
if w.opts.Logger.V(logger.TraceLevel) {
w.opts.Logger.Tracef(nctx, "will be executed %v", steps[idx][nidx])
w.opts.Logger.Trace(nctx, fmt.Sprintf("step will be executed %v", steps[idx][nidx]))
}
cstep := steps[idx][nidx]
// nolint: nestif
@@ -259,21 +259,21 @@ func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...optio
if serr != nil {
step.SetStatus(StatusFailure)
if werr := stepStore.Write(ctx, step.ID()+w.opts.Store.Options().Separator+"rsp", serr); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
w.opts.Logger.Error(ctx, "store write error", "error", werr.Error())
}
if werr := stepStore.Write(ctx, step.ID()+w.opts.Store.Options().Separator+"status", &codec.Frame{Data: []byte(StatusFailure.String())}); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
w.opts.Logger.Error(ctx, "store write error", "error", werr.Error())
}
cherr <- serr
return
}
if werr := stepStore.Write(ctx, step.ID()+w.opts.Store.Options().Separator+"rsp", rsp); werr != nil {
w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
w.opts.Logger.Error(ctx, "store write error", "error", werr.Error())
cherr <- werr
return
}
if werr := stepStore.Write(ctx, step.ID()+w.opts.Store.Options().Separator+"status", &codec.Frame{Data: []byte(StatusSuccess.String())}); werr != nil {
w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
w.opts.Logger.Error(ctx, "store write error", "error", werr.Error())
cherr <- werr
return
}
@@ -292,16 +292,16 @@ func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...optio
if serr != nil {
cstep.SetStatus(StatusFailure)
if werr := stepStore.Write(ctx, cstep.ID()+w.opts.Store.Options().Separator+"rsp", serr); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
w.opts.Logger.Error(ctx, "store write error", "error", werr.Error())
}
if werr := stepStore.Write(ctx, cstep.ID()+w.opts.Store.Options().Separator+"status", &codec.Frame{Data: []byte(StatusFailure.String())}); werr != nil && w.opts.Logger.V(logger.ErrorLevel) {
w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
w.opts.Logger.Error(ctx, "store write error", "error", werr.Error())
}
cherr <- serr
return
}
if werr := stepStore.Write(ctx, cstep.ID()+w.opts.Store.Options().Separator+"rsp", rsp); werr != nil {
w.opts.Logger.Errorf(ctx, "store write error: %v", werr)
w.opts.Logger.Error(ctx, "store write error", "error", werr.Error())
cherr <- werr
return
}
@@ -319,7 +319,7 @@ func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...optio
return eid, nil
}
logger.Tracef(ctx, "wait for finish or error")
w.opts.Logger.Trace(ctx, "wait for finish or error")
select {
case <-nctx.Done():
err = nctx.Err()
@@ -335,15 +335,15 @@ func (w *microWorkflow) Execute(ctx context.Context, req *Message, opts ...optio
switch {
case nctx.Err() != nil:
if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusAborted.String())}); werr != nil {
w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
w.opts.Logger.Error(w.opts.Context, "store write error", "error", werr.Error())
}
case err == nil:
if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusSuccess.String())}); werr != nil {
w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
w.opts.Logger.Error(w.opts.Context, "store write error", "error", werr.Error())
}
case err != nil:
if werr := workflowStore.Write(w.opts.Context, "status", &codec.Frame{Data: []byte(StatusFailure.String())}); werr != nil {
w.opts.Logger.Errorf(w.opts.Context, "store error: %v", werr)
w.opts.Logger.Error(w.opts.Context, "store write error", "error", werr.Error())
}
}