NeonXP/rutina
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

WIP: Added lifecycle mixin

Browse source 
Removed hooks
This commit is contained in:
Alexander Kiryukhin 2019-06-15 10:20:37 +03:00
parent c691d42239
commit 741cf397a9
3 changed files with 38 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
README.md
View file

@ -96,9 +96,9 @@ err := <- r.Errors()
Disabled by default. Use `r.With(rutina.WithErrChan())` to turn on.
## Events and hooks
## Lifecycle events
Rutina has own simple lifecycle events system. You can subscribe your hooks on any of this events:
Rutina has own simple lifecycle events:
* `EventRoutineStart` - Fires when starts new routine
* `EventRoutineStop` - Fires when routine stopped with any result
@ -108,15 +108,6 @@ Rutina has own simple lifecycle events system. You can subscribe your hooks on a
* `EventAppComplete` - Fires when all routines stopped with no errors
* `EventAppFail` - Fires when all routines stopped with error
Example:
```go
r.RegisterHook(rutina.EventRoutineStart, func(ev rutina.Event, rid int) error {
log.Println("Started routine with ID", rid)
return nil
})
```
## Mixins
### Usage

14
mixins.go
View file

@ -58,3 +58,17 @@ func WithErrChan() *MixinErrChan {
func (o MixinErrChan) apply(r *Rutina) {
r.errCh = make(chan error, 1)
}
type LifecycleListener func(event Event, routineID int)
type LifecycleMixin struct {
Listener LifecycleListener
}
func (l LifecycleMixin) apply(r *Rutina) {
r.lifecycleListener = l.Listener
}
func WithLifecycleListener(listener LifecycleListener) *LifecycleMixin {
return &LifecycleMixin{Listener: listener}
}

54
rutina.go
View file

@ -12,23 +12,23 @@ import (
//Rutina is routine manager
type Rutina struct {
ctx context.Context // State of application (started/stopped)
Cancel func() // Cancel func that stops all routines
wg sync.WaitGroup // WaitGroup that wait all routines to complete
onceErr sync.Once // Flag that prevents overwrite first error that shutdowns all routines
onceWait sync.Once // Flag that prevents wait already waited rutina
err error // First error that shutdowns all routines
logger *log.Logger // Optional logger
counter *uint64 // Optional counter that names routines with increment ids for debug purposes at logger
errCh chan error // Optional channel for errors when RestartIfFail and DoNothingIfFail
hooks map[Event][]Hook // Lifecycle hooks
ctx context.Context // State of application (started/stopped)
Cancel func() // Cancel func that stops all routines
wg sync.WaitGroup // WaitGroup that wait all routines to complete
onceErr sync.Once // Flag that prevents overwrite first error that shutdowns all routines
onceWait sync.Once // Flag that prevents wait already waited rutina
err error // First error that shutdowns all routines
logger *log.Logger // Optional logger
counter *uint64 // Optional counter that names routines with increment ids for debug purposes at logger
errCh chan error // Optional channel for errors when RestartIfFail and DoNothingIfFail
lifecycleListener LifecycleListener // Optional listener for events
}
// New instance with builtin context
func New(mixins ...Mixin) *Rutina {
ctx, cancel := context.WithCancel(context.Background())
var counter uint64
r := &Rutina{ctx: ctx, Cancel: cancel, counter: &counter, errCh: nil, hooks: map[Event][]Hook{}}
r := &Rutina{ctx: ctx, Cancel: cancel, counter: &counter, errCh: nil}
return r.With(mixins...)
}
@ -40,10 +40,6 @@ func (r *Rutina) With(mixins ...Mixin) *Rutina {
return r
}
func (r *Rutina) RegisterHook(ev Event, hook Hook) {
r.hooks[ev] = append(r.hooks[ev], hook)
}
// Go routine
func (r *Rutina) Go(doer func(ctx context.Context) error, opts ...Options) {
// Check that context is not canceled yet
@ -77,10 +73,10 @@ func (r *Rutina) Go(doer func(ctx context.Context) error, opts ...Options) {
go func() {
defer r.wg.Done()
id := atomic.AddUint64(r.counter, 1)
r.fire(EventRoutineStart, int(id))
r.lifecycleEvent(EventRoutineStart, int(id))
if err := doer(r.ctx); err != nil {
r.fire(EventRoutineFail, int(id))
r.fire(EventRoutineStop, int(id))
r.lifecycleEvent(EventRoutineFail, int(id))
r.lifecycleEvent(EventRoutineStop, int(id))
// errors history
if r.errCh != nil {
r.errCh <- err
@ -98,8 +94,8 @@ func (r *Rutina) Go(doer func(ctx context.Context) error, opts ...Options) {
}
// endregion
} else {
r.fire(EventRoutineComplete, int(id))
r.fire(EventRoutineStop, int(id))
r.lifecycleEvent(EventRoutineComplete, int(id))
r.lifecycleEvent(EventRoutineStop, int(id))
// region routine successfully done
switch onDone {
case ShutdownIfDone:
@ -140,11 +136,11 @@ func (r *Rutina) ListenOsSignals(signals ...os.Signal) {
func (r *Rutina) Wait() error {
r.onceWait.Do(func() {
r.wg.Wait()
r.fire(EventAppStop, 0)
r.lifecycleEvent(EventAppStop, 0)
if r.err == nil {
r.fire(EventAppComplete, 0)
r.lifecycleEvent(EventAppComplete, 0)
} else {
r.fire(EventAppFail, 0)
r.lifecycleEvent(EventAppFail, 0)
}
if r.errCh != nil {
close(r.errCh)
@ -153,16 +149,10 @@ func (r *Rutina) Wait() error {
return r.err
}
func (r *Rutina) fire(ev Event, rid int) {
func (r *Rutina) lifecycleEvent(ev Event, rid int) {
r.log("Event = %s Routine ID = %d", ev.String(), rid)
if hooks, ok := r.hooks[ev]; ok == true {
for _, h := range hooks {
if err := h(ev, r, rid); err != nil {
if r.errCh != nil {
r.errCh <- err
}
}
}
if r.lifecycleListener != nil {
r.lifecycleListener(ev, rid)
}
}