diff options
author | Valery Piashchynski <[email protected]> | 2020-12-17 12:13:55 +0300 |
---|---|---|
committer | GitHub <[email protected]> | 2020-12-17 12:13:55 +0300 |
commit | ee0cb478c74c393a35155c2bf51e1ef260e0e5e2 (patch) | |
tree | 2c99d4c6e2b2e9e3fa155d5d68a9d471c9aeeb9b /interfaces/events | |
parent | a1dc59cabb6e63eab232922f4eb5a19dbd168f44 (diff) | |
parent | edf924b37bcdad14eb31014c571ab58720aa178f (diff) |
Merge pull request #452 from spiral/refactor/splitv2.0.0-alpha23
Refactor/split
Diffstat (limited to 'interfaces/events')
-rw-r--r-- | interfaces/events/handler.go | 10 | ||||
-rw-r--r-- | interfaces/events/pool_events.go | 65 | ||||
-rw-r--r-- | interfaces/events/worker_events.go | 34 |
3 files changed, 109 insertions, 0 deletions
diff --git a/interfaces/events/handler.go b/interfaces/events/handler.go new file mode 100644 index 00000000..01f64d73 --- /dev/null +++ b/interfaces/events/handler.go @@ -0,0 +1,10 @@ +package events + +type Handler interface { + NumListeners() int + AddListener(listener EventListener) + Push(e interface{}) +} + +// Event listener listens for the events produced by worker, worker pool or other service. +type EventListener func(event interface{}) diff --git a/interfaces/events/pool_events.go b/interfaces/events/pool_events.go new file mode 100644 index 00000000..cc32f6b2 --- /dev/null +++ b/interfaces/events/pool_events.go @@ -0,0 +1,65 @@ +package events + +const ( + // EventWorkerConstruct thrown when new worker is spawned. + EventWorkerConstruct P = iota + 7800 + + // EventWorkerDestruct thrown after worker destruction. + EventWorkerDestruct + + // EventPoolError caused on pool wide errors. + EventPoolError + + // EventSupervisorError triggered when supervisor can not complete work. + EventSupervisorError + + // EventNoFreeWorkers triggered when there are no free workers in the stack and timeout for worker allocate elapsed + EventNoFreeWorkers + + // todo: EventMaxMemory caused when worker consumes more memory than allowed. + EventMaxMemory + + // todo: EventTTL thrown when worker is removed due TTL being reached. Context is rr.WorkerError + EventTTL + + // todo: EventIdleTTL triggered when worker spends too much time at rest. + EventIdleTTL + + // todo: EventExecTTL triggered when worker spends too much time doing the task (max_execution_time). + EventExecTTL +) + +type P int64 + +func (ev P) String() string { + switch ev { + case EventWorkerConstruct: + return "EventWorkerConstruct" + case EventWorkerDestruct: + return "EventWorkerDestruct" + case EventPoolError: + return "EventPoolError" + case EventSupervisorError: + return "EventSupervisorError" + case EventNoFreeWorkers: + return "EventNoFreeWorkers" + case EventMaxMemory: + return "EventMaxMemory" + case EventTTL: + return "EventTTL" + case EventIdleTTL: + return "EventIdleTTL" + case EventExecTTL: + return "EventExecTTL" + } + return "Unknown event type" +} + +// PoolEvent triggered by pool on different events. Pool as also trigger WorkerEvent in case of log. +type PoolEvent struct { + // Event type, see below. + Event P + + // Payload depends on event type, typically it's worker or error. + Payload interface{} +} diff --git a/interfaces/events/worker_events.go b/interfaces/events/worker_events.go new file mode 100644 index 00000000..497f0a06 --- /dev/null +++ b/interfaces/events/worker_events.go @@ -0,0 +1,34 @@ +package events + +// EventWorkerKill thrown after WorkerProcess is being forcefully killed. +const ( + // EventWorkerError triggered after WorkerProcess. Except payload to be error. + EventWorkerError E = iota + 200 + + // EventWorkerLog triggered on every write to WorkerProcess StdErr pipe (batched). Except payload to be []byte string. + EventWorkerLog +) + +type E int64 + +func (ev E) String() string { + switch ev { + case EventWorkerError: + return "EventWorkerError" + case EventWorkerLog: + return "EventWorkerLog" + } + return "Unknown event type" +} + +// WorkerEvent wraps worker events. +type WorkerEvent struct { + // Event id, see below. + Event E + + // Worker triggered the event. + Worker interface{} + + // Event specific payload. + Payload interface{} +} |