diff options
| -rw-r--r-- | errors.go | 8 | ||||
| -rw-r--r-- | pool.go | 15 | ||||
| -rw-r--r-- | router.go | 14 | ||||
| -rw-r--r-- | static_pool.go | 12 | ||||
| -rw-r--r-- | static_pool_test.go | 2 |
5 files changed, 22 insertions, 29 deletions
@@ -22,11 +22,3 @@ type WorkerError struct { func (e WorkerError) Error() string { return e.Caused.Error() } - -// PoolError is pool wide error -type PoolError string - -// Error converts error context to string -func (e PoolError) Error() string { - return string(e) -} @@ -1,14 +1,17 @@ package roadrunner const ( - // EventCreated thrown when new worker is spawned. - EventCreated = iota + // EventWorkerCreate thrown when new worker is spawned. + EventWorkerCreate = iota - // EventDestruct thrown before worker destruction. - EventDestruct + // EventWorkerDestruct thrown before worker destruction. + EventWorkerDestruct - // EventError thrown any worker related even happen (error passed as context) - EventError + // EventWorkerError thrown any worker related even happen (passed with WorkerError) + EventWorkerError + + // EventPoolError caused on pool wide errors + EventPoolError ) // Pool managed set of inner worker processes. @@ -135,15 +135,13 @@ func (r *Router) throw(event int, ctx interface{}) { } } -// Observe pools +// Observe pool events func (r *Router) poolObserver(event int, ctx interface{}) { - if event == EventError { - if _, ok := ctx.(PoolError); ok { - // pool failure, rebuilding - r.Reset() - } - } - // bypassing to user specified observer r.throw(event, ctx) + + if event == EventPoolError { + // pool failure, rebuilding + r.Reset() + } } diff --git a/static_pool.go b/static_pool.go index 4e75cddc..be5e9b06 100644 --- a/static_pool.go +++ b/static_pool.go @@ -170,11 +170,11 @@ func (p *StaticPool) replaceWorker(w *Worker, caused interface{}) { go p.destroyWorker(w) if nw, err := p.createWorker(); err != nil { - p.throw(EventError, WorkerError{Worker: w, Caused: err}) + p.throw(EventWorkerError, WorkerError{Worker: w, Caused: err}) if len(p.Workers()) == 0 { // possible situation when major error causes all PHP scripts to die (for example dead DB) - p.throw(EventError, PoolError("all workers are dead")) + p.throw(EventPoolError, fmt.Errorf("all workers are dead")) } } else { p.free <- nw @@ -183,7 +183,7 @@ func (p *StaticPool) replaceWorker(w *Worker, caused interface{}) { // destroyWorker destroys workers and removes it from the pool. func (p *StaticPool) destroyWorker(w *Worker) { - p.throw(EventDestruct, w) + p.throw(EventWorkerDestruct, w) // detaching p.muw.Lock() @@ -203,7 +203,7 @@ func (p *StaticPool) destroyWorker(w *Worker) { case <-time.NewTimer(p.cfg.DestroyTimeout).C: // failed to stop process if err := w.Kill(); err != nil { - p.throw(EventError, WorkerError{Worker: w, Caused: err}) + p.throw(EventWorkerError, WorkerError{Worker: w, Caused: err}) } } } @@ -216,11 +216,11 @@ func (p *StaticPool) createWorker() (*Worker, error) { return nil, err } - p.throw(EventCreated, w) + p.throw(EventWorkerCreate, w) go func(w *Worker) { if err := w.Wait(); err != nil { - p.throw(EventError, WorkerError{Worker: w, Caused: err}) + p.throw(EventWorkerError, WorkerError{Worker: w, Caused: err}) } }(w) diff --git a/static_pool_test.go b/static_pool_test.go index a1db7c8c..64635716 100644 --- a/static_pool_test.go +++ b/static_pool_test.go @@ -150,7 +150,7 @@ func Test_StaticPool_Broken_Replace(t *testing.T) { assert.NotNil(t, p) assert.NoError(t, err) - p.Observe(func(e int, w *Worker, ctx interface{}) { + p.Observe(func(e int, ctx interface{}) { if err, ok := ctx.(error); ok { assert.Contains(t, err.Error(), "undefined_function()") } |