-
Notifications
You must be signed in to change notification settings - Fork 13
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1204 from mesg-foundation/feature/workflow-ressource
Move workflow to its own database/api
- Loading branch information
Showing
29 changed files
with
1,708 additions
and
819 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| package database | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
|
|
||
| "github.com/mesg-foundation/engine/hash" | ||
| "github.com/mesg-foundation/engine/workflow" | ||
| "github.com/sirupsen/logrus" | ||
| "github.com/syndtr/goleveldb/leveldb" | ||
| ) | ||
|
|
||
| // WorkflowDB describes the API of database package. | ||
| type WorkflowDB interface { | ||
| // Save saves a workflow to database. | ||
| Save(s *workflow.Workflow) error | ||
|
|
||
| // Get gets a workflow from database by its unique hash. | ||
| Get(hash hash.Hash) (*workflow.Workflow, error) | ||
|
|
||
| // Delete deletes a workflow from database by its unique hash. | ||
| Delete(hash hash.Hash) error | ||
|
|
||
| // All returns all workflows from database. | ||
| All() ([]*workflow.Workflow, error) | ||
|
|
||
| // Close closes underlying database connection. | ||
| Close() error | ||
| } | ||
|
|
||
| // LevelDBWorkflowDB is a database for storing workflows definition. | ||
| type LevelDBWorkflowDB struct { | ||
| db *leveldb.DB | ||
| } | ||
|
|
||
| // NewWorkflowDB returns the database which is located under given path. | ||
| func NewWorkflowDB(path string) (*LevelDBWorkflowDB, error) { | ||
| db, err := leveldb.OpenFile(path, nil) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &LevelDBWorkflowDB{db: db}, nil | ||
| } | ||
|
|
||
| // marshal returns the byte slice from workflow. | ||
| func (d *LevelDBWorkflowDB) marshal(s *workflow.Workflow) ([]byte, error) { | ||
| return json.Marshal(s) | ||
| } | ||
|
|
||
| // unmarshal returns the workflow from byte slice. | ||
| func (d *LevelDBWorkflowDB) unmarshal(hash hash.Hash, value []byte) (*workflow.Workflow, error) { | ||
| var s workflow.Workflow | ||
| if err := json.Unmarshal(value, &s); err != nil { | ||
| return nil, fmt.Errorf("database: could not decode workflow %q: %s", hash, err) | ||
| } | ||
| return &s, nil | ||
| } | ||
|
|
||
| // All returns every workflow in database. | ||
| func (d *LevelDBWorkflowDB) All() ([]*workflow.Workflow, error) { | ||
| var ( | ||
| workflows []*workflow.Workflow | ||
| iter = d.db.NewIterator(nil, nil) | ||
| ) | ||
| for iter.Next() { | ||
| hash := hash.Hash(iter.Key()) | ||
| s, err := d.unmarshal(hash, iter.Value()) | ||
| if err != nil { | ||
| // NOTE: Ignore all decode errors (possibly due to a workflow | ||
| // structure change or database corruption) | ||
| logrus.WithField("workflow", hash.String()).Warning(err.Error()) | ||
| continue | ||
| } | ||
| workflows = append(workflows, s) | ||
| } | ||
| iter.Release() | ||
| return workflows, iter.Error() | ||
| } | ||
|
|
||
| // Delete deletes workflow from database. | ||
| func (d *LevelDBWorkflowDB) Delete(hash hash.Hash) error { | ||
| tx, err := d.db.OpenTransaction() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if _, err := tx.Get(hash, nil); err != nil { | ||
| tx.Discard() | ||
| if err == leveldb.ErrNotFound { | ||
| return &ErrNotFound{resource: "workflow", hash: hash} | ||
| } | ||
| return err | ||
| } | ||
| if err := tx.Delete(hash, nil); err != nil { | ||
| tx.Discard() | ||
| return err | ||
| } | ||
| return tx.Commit() | ||
|
|
||
| } | ||
|
|
||
| // Get retrives workflow from database. | ||
| func (d *LevelDBWorkflowDB) Get(hash hash.Hash) (*workflow.Workflow, error) { | ||
| b, err := d.db.Get(hash, nil) | ||
| if err != nil { | ||
| if err == leveldb.ErrNotFound { | ||
| return nil, &ErrNotFound{resource: "workflow", hash: hash} | ||
| } | ||
| return nil, err | ||
| } | ||
| return d.unmarshal(hash, b) | ||
| } | ||
|
|
||
| // Save stores workflow in database. | ||
| // If there is an another workflow that uses the same sid, it'll be deleted. | ||
| func (d *LevelDBWorkflowDB) Save(s *workflow.Workflow) error { | ||
| if s.Hash.IsZero() { | ||
| return errCannotSaveWithoutHash | ||
| } | ||
|
|
||
| b, err := d.marshal(s) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return d.db.Put(s.Hash, b, nil) | ||
| } | ||
|
|
||
| // Close closes database. | ||
| func (d *LevelDBWorkflowDB) Close() error { | ||
| return d.db.Close() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package engine | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/mesg-foundation/engine/execution" | ||
| "github.com/mesg-foundation/engine/hash" | ||
| eventsdk "github.com/mesg-foundation/engine/sdk/event" | ||
| executionsdk "github.com/mesg-foundation/engine/sdk/execution" | ||
| workflowsdk "github.com/mesg-foundation/engine/sdk/workflow" | ||
| "github.com/mesg-foundation/engine/workflow" | ||
| ) | ||
|
|
||
| // Workflow exposes functions of the workflow | ||
| type Workflow struct { | ||
| event *eventsdk.Event | ||
| eventStream *eventsdk.Listener | ||
|
|
||
| execution *executionsdk.Execution | ||
| executionStream *executionsdk.Listener | ||
|
|
||
| workflow *workflowsdk.Workflow | ||
|
|
||
| ErrC chan error | ||
| } | ||
|
|
||
| // New creates a new Workflow instance | ||
| func New(event *eventsdk.Event, execution *executionsdk.Execution, workflow *workflowsdk.Workflow) *Workflow { | ||
| return &Workflow{ | ||
| event: event, | ||
| execution: execution, | ||
| workflow: workflow, | ||
| ErrC: make(chan error), | ||
| } | ||
| } | ||
|
|
||
| // Start the workflow engine | ||
| func (w *Workflow) Start() error { | ||
| if w.eventStream != nil || w.executionStream != nil { | ||
| return fmt.Errorf("workflow engine already running") | ||
| } | ||
| w.eventStream = w.event.GetStream(nil) | ||
| w.executionStream = w.execution.GetStream(&executionsdk.Filter{ | ||
| Statuses: []execution.Status{execution.Completed}, | ||
| }) | ||
| for { | ||
| select { | ||
| case event := <-w.eventStream.C: | ||
| go w.processTrigger(workflow.EVENT, event.InstanceHash, event.Key, event.Data, event.Hash, nil) | ||
| case execution := <-w.executionStream.C: | ||
| go w.processTrigger(workflow.RESULT, execution.InstanceHash, execution.TaskKey, execution.Outputs, nil, execution) | ||
| go w.processExecution(execution) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (w *Workflow) processTrigger(trigger workflow.TriggerType, instanceHash hash.Hash, key string, data map[string]interface{}, eventHash hash.Hash, exec *execution.Execution) { | ||
| workflows, err := w.workflow.List() | ||
| if err != nil { | ||
| w.ErrC <- err | ||
| return | ||
| } | ||
| for _, wf := range workflows { | ||
| if wf.Trigger.Match(trigger, instanceHash, key, data) { | ||
| if err := w.triggerExecution(wf, exec, eventHash, data); err != nil { | ||
| w.ErrC <- err | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (w *Workflow) processExecution(exec *execution.Execution) error { | ||
| if exec.WorkflowHash.IsZero() { | ||
| return nil | ||
| } | ||
| wf, err := w.workflow.Get(exec.WorkflowHash) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return w.triggerExecution(wf, exec, nil, exec.Outputs) | ||
| } | ||
|
|
||
| func (w *Workflow) triggerExecution(wf *workflow.Workflow, prev *execution.Execution, eventHash hash.Hash, data map[string]interface{}) error { | ||
| height, err := w.getHeight(wf, prev) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if len(wf.Tasks) <= height { | ||
| // end of workflow | ||
| return nil | ||
| } | ||
| var parentHash hash.Hash | ||
| if prev != nil { | ||
| parentHash = prev.Hash | ||
| } | ||
| task := wf.Tasks[height] | ||
| if _, err := w.execution.Execute(wf.Hash, task.InstanceHash, eventHash, parentHash, task.TaskKey, data, []string{}); err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (w *Workflow) getHeight(wf *workflow.Workflow, exec *execution.Execution) (int, error) { | ||
| if exec == nil { | ||
| return 0, nil | ||
| } | ||
| // Result from other workflow | ||
| if !exec.WorkflowHash.Equal(wf.Hash) { | ||
| return 0, nil | ||
| } | ||
| // Execution triggered by an event | ||
| if !exec.EventHash.IsZero() { | ||
| return 1, nil | ||
| } | ||
| if exec.ParentHash.IsZero() { | ||
| panic("parent hash should be present if event is not") | ||
| } | ||
| if exec.ParentHash.Equal(exec.Hash) { | ||
| panic("parent hash cannot be equal to execution hash") | ||
| } | ||
| parent, err := w.execution.Get(exec.ParentHash) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| parentHeight, err := w.getHeight(wf, parent) | ||
| return parentHeight + 1, err | ||
| } |
Oops, something went wrong.