@@ -71,22 +71,12 @@ func (n *CronListenerManager) PermanentLookupListenerSupported() bool {
7171// AddPermanentLookupListener adds a permanent lookup listener for the given key.
7272// It schedules a recurring job that checks for state changes.
7373func (n * CronListenerManager ) AddPermanentLookupListener (namespace string , key string , listener Listener ) error {
74- logger .Debugf ("AddPermanentLookupListener [%s:%s]" , namespace , key )
74+ logger .Infof ("AddPermanentLookupListener [%s:%s]" , namespace , key )
7575
76- var lastValue []byte
7776 j , err := n .scheduler .NewJob (
7877 gocron .DurationJob (n .config .PermanentInterval ()),
79- gocron .NewTask (func () {
80- logger .Debugf ("[PermanentKeyCheck] check for key [%s:%s]" , namespace , key )
81- v , err := n .queryService .GetState (namespace , key )
82- if err == nil && v != nil && len (v .Raw ) != 0 {
83- if ! bytes .Equal (lastValue , v .Raw ) {
84- logger .Debugf ("[PermanentKeyCheck] key [%s:%s] found with new value, notify listener" , namespace , key )
85- listener .OnStatus (context .Background (), key , v .Raw )
86- lastValue = v .Raw
87- }
88- }
89- }),
78+ gocron .NewTask (NewPermanentJob (namespace , key , listener , n .queryService )),
79+ gocron .WithSingletonMode (gocron .LimitModeReschedule ),
9080 )
9181 if err != nil {
9282 return errors .Wrapf (err , "failed scheduling permanent lookup job" )
@@ -103,7 +93,7 @@ func (n *CronListenerManager) AddPermanentLookupListener(namespace string, key s
10393// It schedules a recurring job that checks for the key until it is found or the deadline is reached.
10494// The job is removed once it is invoked.
10595func (n * CronListenerManager ) AddLookupListener (namespace string , key string , listener lookup.Listener ) error {
106- logger .Debugf ("AddLookupListener [%s:%s]" , namespace , key )
96+ logger .Infof ("AddLookupListener [%s:%s]" , namespace , key )
10797
10898 deadline := time .Now ().Add (n .config .OnceDeadline ())
10999
@@ -112,10 +102,10 @@ func (n *CronListenerManager) AddLookupListener(namespace string, key string, li
112102 var err error
113103
114104 task := gocron .NewTask (func () {
115- logger .Debugf ("[KeyCheck] check for key [%s:%s]" , namespace , key )
105+ logger .Infof ("[KeyCheck] check for key [%s:%s]" , namespace , key )
116106 v , err := n .queryService .GetState (namespace , key )
117107 if err == nil && v != nil && len (v .Raw ) != 0 {
118- logger .Debugf ("[KeyCheck] key [%s:%s] found, notify listener" , namespace , key )
108+ logger .Infof ("[KeyCheck] key [%s:%s] found, notify listener" , namespace , key )
119109 listener .OnStatus (context .Background (), key , v .Raw )
120110
121111 // Stop the job, no error is expected here
@@ -125,7 +115,7 @@ func (n *CronListenerManager) AddLookupListener(namespace string, key string, li
125115 }
126116
127117 if time .Now ().After (deadline ) {
128- logger .Debugf ("[KeyCheck] key [%s:%s] not found, deadline reached" , namespace , key )
118+ logger .Infof ("[KeyCheck] key [%s:%s] not found, deadline reached" , namespace , key )
129119 listener .OnError (context .Background (), key , errors .Errorf ("key [%s:%s] not found" , namespace , key ))
130120
131121 // Stop the job, no error is expected here
@@ -138,6 +128,7 @@ func (n *CronListenerManager) AddLookupListener(namespace string, key string, li
138128 j , err = n .scheduler .NewJob (
139129 gocron .DurationJob (n .config .OnceInterval ()),
140130 task ,
131+ gocron .WithSingletonMode (gocron .LimitModeReschedule ),
141132 )
142133 if err != nil {
143134 return errors .Wrapf (err , "failed scheduling lookup job" )
@@ -152,7 +143,7 @@ func (n *CronListenerManager) AddLookupListener(namespace string, key string, li
152143
153144// RemoveLookupListener removes a lookup listener for the given key and stops its associated job.
154145func (n * CronListenerManager ) RemoveLookupListener (id string , listener Listener ) error {
155- logger .Debugf ("RemoveLookupListener [%s]" , id )
146+ logger .Infof ("RemoveLookupListener [%s]" , id )
156147 n .mu .Lock ()
157148 defer n .mu .Unlock ()
158149
@@ -195,3 +186,28 @@ func (n *CronNSListenerManagerProvider) NewManager(network, channel string) (loo
195186
196187 return NewCronListenerManager (qs , n .config )
197188}
189+
190+ type PermanentJob struct {
191+ namespace string
192+ key string
193+ listener Listener
194+
195+ queryService QueryService
196+ lastValue []byte
197+ }
198+
199+ func NewPermanentJob (namespace string , key string , listener Listener , queryService QueryService ) * PermanentJob {
200+ return & PermanentJob {namespace : namespace , key : key , listener : listener , queryService : queryService }
201+ }
202+
203+ func (j * PermanentJob ) Run () {
204+ logger .Infof ("[PermanentKeyCheck] check for key [%s:%s]" , j .namespace , j .key )
205+ v , err := j .queryService .GetState (j .namespace , j .key )
206+ if err == nil && v != nil && len (v .Raw ) != 0 {
207+ if ! bytes .Equal (j .lastValue , v .Raw ) {
208+ logger .Infof ("[PermanentKeyCheck] key [%s:%s] found with new value, notify listener" , j .namespace , j .key )
209+ j .listener .OnStatus (context .Background (), j .key , v .Raw )
210+ j .lastValue = v .Raw
211+ }
212+ }
213+ }
0 commit comments