this repo has no description
Only ping a consumer if we haven't written _anything_ in a while (#264)
Consumer timeout checking is great but if we're playing back a TON of
events, the ping ends up deep in a backlog and out 5 second timeout on
the ping doesn't make sense so let's keep track of the last time we
successfully wrote to the consumer and only try to ping if it's been 30
seconds since our last write.
+14
+14
bgs/bgs.go
+14
bgs/bgs.go
···
460
460
return fmt.Errorf("upgrading websocket: %w", err)
461
461
}
462
462
463
+
lastWriteLk := sync.Mutex{}
464
+
lastWrite := time.Now()
465
+
463
466
// Start a goroutine to ping the client every 30 seconds to check if it's
464
467
// still alive. If the client doesn't respond to a ping within 5 seconds,
465
468
// we'll close the connection and teardown the consumer.
···
470
473
for {
471
474
select {
472
475
case <-ticker.C:
476
+
lastWriteLk.Lock()
477
+
lw := lastWrite
478
+
lastWriteLk.Unlock()
479
+
480
+
if time.Since(lw) < 30*time.Second {
481
+
continue
482
+
}
483
+
473
484
if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil {
474
485
log.Errorf("failed to ping client: %s", err)
475
486
cancel()
···
549
560
if err := wc.Close(); err != nil {
550
561
return fmt.Errorf("failed to flush-close our event write: %w", err)
551
562
}
563
+
lastWriteLk.Lock()
564
+
lastWrite = time.Now()
565
+
lastWriteLk.Unlock()
552
566
sentCounter.Inc()
553
567
case <-ctx.Done():
554
568
return nil