From 3e9c29b57b3940a781938f3cebb8589c8870cff9 Mon Sep 17 00:00:00 2001
From: Pavel Tatarskiy <pavel.tatarsky@gmail.com>
Date: Wed, 9 Oct 2024 20:55:52 +0300
Subject: [PATCH] another try to remove job observer leaks

---
 services/job/job.go | 52 ++++++++++++++++++++++-----------------------
 1 file changed, 26 insertions(+), 26 deletions(-)

diff --git a/services/job/job.go b/services/job/job.go
index 9e4cdb5..78c09a9 100644
--- a/services/job/job.go
+++ b/services/job/job.go
@@ -43,19 +43,19 @@ func NewObserver() *Observer {
 }
 
 type Job struct {
-	ID           string
-	Queue        string
-	l            []LogItem
-	runnable     Runnable
-	observers    map[string]*Observer
-	observersMux sync.Mutex
-	closed       bool
-	mux          sync.Mutex
-	cur          string
-	Context      context.Context
-	storage      Storage
-	main         bool
-	purge        bool
+	ID        string
+	Queue     string
+	l         []LogItem
+	runnable  Runnable
+	observers map[string]*Observer
+	//observersMux sync.Mutex
+	closed  bool
+	mux     sync.Mutex
+	cur     string
+	Context context.Context
+	storage Storage
+	main    bool
+	purge   bool
 }
 
 type LogItemLevel string
@@ -157,23 +157,23 @@ func (s *Job) Run(ctx context.Context) error {
 }
 
 func (s *Job) ObserveLog() *Observer {
-	s.observersMux.Lock()
-	defer s.observersMux.Unlock()
+	//s.observersMux.Lock()
+	//defer s.observersMux.Unlock()
 	o := NewObserver()
 	s.observers[o.ID] = o
 	return o
 }
 
-func (s *Job) RemoveObserver(o *Observer) {
-	s.observersMux.Lock()
-	defer s.observersMux.Unlock()
-	delete(s.observers, o.ID)
-	o.Close()
-}
+//func (s *Job) RemoveObserver(o *Observer) {
+//	s.observersMux.Lock()
+//	defer s.observersMux.Unlock()
+//	delete(s.observers, o.ID)
+//	o.Close()
+//}
 
 func (s *Job) pushToObservers(l LogItem) {
-	s.observersMux.Lock()
-	defer s.observersMux.Unlock()
+	//s.observersMux.Lock()
+	//defer s.observersMux.Unlock()
 	for _, o := range s.observers {
 		o.Push(l)
 	}
@@ -333,8 +333,8 @@ func (s *Job) close() {
 	_ = s.log(LogItem{
 		Level: Close,
 	})
-	s.observersMux.Lock()
-	defer s.observersMux.Unlock()
+	//s.observersMux.Lock()
+	//defer s.observersMux.Unlock()
 	for _, o := range s.observers {
 		o.Close()
 	}
@@ -411,7 +411,7 @@ func (s *Jobs) Log(ctx context.Context, id string) (c chan LogItem, err error) {
 				select {
 				case <-ctx.Done():
 					close(c)
-					j.RemoveObserver(o)
+					//j.RemoveObserver(o)
 					return
 				case i, okk := <-o.C:
 					if !okk {