Update events handler, add mutex to push

5 files changed, 44 insertions, 11 deletions

diff --git a/pkg/events/events.go b/pkg/events/events.go
new file mode 100755
index 00000000..92dc103a
--- /dev/null
+++ b/pkg/events/events.go
@@ -0,0 +1,38 @@
+package events
+
+import (
+	"sync"
+
+	"github.com/spiral/roadrunner/v2/interfaces/events"
+)
+
+// EventHandler helps to broadcast events to multiple listeners.
+type EventHandler struct {
+	listeners []events.EventListener
+	sync.RWMutex
+}
+
+func NewEventsHandler() events.Handler {
+	return &EventHandler{listeners: make([]events.EventListener, 0, 2)}
+}
+
+// NumListeners returns number of event listeners.
+func (eb *EventHandler) NumListeners() int {
+	return len(eb.listeners)
+}
+
+// AddListener registers new event listener.
+func (eb *EventHandler) AddListener(listener events.EventListener) {
+	eb.Lock()
+	defer eb.Unlock()
+	eb.listeners = append(eb.listeners, listener)
+}
+
+// Push broadcast events across all event listeners.
+func (eb *EventHandler) Push(e interface{}) {
+	eb.Lock()
+	defer eb.Unlock()
+	for k := range eb.listeners {
+		eb.listeners[k](e)
+	}
+}
diff --git a/pkg/pool/static_pool.go b/pkg/pool/static_pool.go
index 691290b2..6cc42143 100755
--- a/pkg/pool/static_pool.go
+++ b/pkg/pool/static_pool.go
@@ -9,9 +9,9 @@ import (
 	"github.com/spiral/roadrunner/v2/interfaces/pool"
 	"github.com/spiral/roadrunner/v2/interfaces/worker"
 	"github.com/spiral/roadrunner/v2/internal"
+	events2 "github.com/spiral/roadrunner/v2/pkg/events"
 	syncWorker "github.com/spiral/roadrunner/v2/pkg/worker"
 	workerWatcher "github.com/spiral/roadrunner/v2/pkg/worker_watcher"
-	"github.com/spiral/roadrunner/v2/util"
 )
 
 // StopRequest can be sent by worker to indicate that restart is required.
@@ -71,7 +71,7 @@ func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory worker.Factory,
 		cfg:     cfg,
 		cmd:     cmd,
 		factory: factory,
-		events:  util.NewEventsHandler(),
+		events:  events2.NewEventsHandler(),
 		after:   make([]After, 0, 0),
 		before:  make([]Before, 0, 0),
 	}
diff --git a/pkg/pool/static_pool_test.go b/pkg/pool/static_pool_test.go
index 0794b8e6..dd33a1a6 100755
--- a/pkg/pool/static_pool_test.go
+++ b/pkg/pool/static_pool_test.go
@@ -175,7 +175,7 @@ func Test_StaticPool_Broken_Replace(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, p)
 
-	block := make(chan struct{})
+	block := make(chan struct{}, 1)
 
 	p.AddListener(func(event interface{}) {
 		if wev, ok := event.(events.WorkerEvent); ok {
diff --git a/pkg/worker/worker.go b/pkg/worker/worker.go
index 998ed592..35d3264e 100755
--- a/pkg/worker/worker.go
+++ b/pkg/worker/worker.go
@@ -13,12 +13,11 @@ import (
 	"time"
 
 	"github.com/spiral/errors"
+	"github.com/spiral/goridge/v3"
 	"github.com/spiral/roadrunner/v2/interfaces/events"
 	"github.com/spiral/roadrunner/v2/interfaces/worker"
 	"github.com/spiral/roadrunner/v2/internal"
-	"github.com/spiral/roadrunner/v2/util"
-
-	"github.com/spiral/goridge/v3"
+	events2 "github.com/spiral/roadrunner/v2/pkg/events"
 	"go.uber.org/multierr"
 )
 
@@ -89,7 +88,7 @@ func InitBaseWorker(cmd *exec.Cmd) (worker.BaseProcess, error) {
 	}
 	w := &Process{
 		created: time.Now(),
-		events:  util.NewEventsHandler(),
+		events:  events2.NewEventsHandler(),
 		cmd:     cmd,
 		state:   internal.NewWorkerState(internal.StateInactive),
 		stderr:  new(bytes.Buffer),
diff --git a/pkg/worker_watcher/worker_watcher.go b/pkg/worker_watcher/worker_watcher.go
index 170a6ac8..8788e509 100755
--- a/pkg/worker_watcher/worker_watcher.go
+++ b/pkg/worker_watcher/worker_watcher.go
@@ -279,13 +279,11 @@ func (ww *workerWatcher) wait(w worker.BaseProcess) {
 	const op = errors.Op("process wait")
 	err := w.Wait()
 	if err != nil {
-		ww.mutex.Lock()
 		ww.events.Push(events.WorkerEvent{
 			Event:   events.EventWorkerError,
 			Worker:  w,
 			Payload: errors.E(op, err),
 		})
-		ww.mutex.Unlock()
 	}
 
 	if w.State().Value() == internal.StateDestroyed {
@@ -296,12 +294,10 @@ func (ww *workerWatcher) wait(w worker.BaseProcess) {
 	_ = ww.stack.FindAndRemoveByPid(w.Pid())
 	err = ww.AllocateNew()
 	if err != nil {
-		ww.mutex.Lock()
 		ww.events.Push(events.PoolEvent{
 			Event:   events.EventPoolError,
 			Payload: errors.E(op, err),
 		})
-		ww.mutex.Unlock()
 	}
 }