Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2cb0bf98ad | ||
|
|
a43935308e |
+15
-10
@@ -60,17 +60,22 @@ func New() *Tracker {
|
||||
}
|
||||
}
|
||||
|
||||
// SetConnected updates the connection status
|
||||
// SetConnected updates the connection status. Every connection after the
|
||||
// first counts one reconnect.
|
||||
func (t *Tracker) SetConnected(connected bool) {
|
||||
wasConnected := t.isConnected.Swap(connected)
|
||||
if connected {
|
||||
t.mu.Lock()
|
||||
t.connectedSince = time.Now()
|
||||
t.mu.Unlock()
|
||||
// Increment reconnect count (but not for the initial connection)
|
||||
if wasConnected || t.reconnectCount.Load() > 0 {
|
||||
t.reconnectCount.Add(1)
|
||||
}
|
||||
t.isConnected.Store(connected)
|
||||
if !connected {
|
||||
return
|
||||
}
|
||||
|
||||
t.mu.Lock()
|
||||
// connectedSince is zero until the first connection.
|
||||
connectedBefore := !t.connectedSince.IsZero()
|
||||
t.connectedSince = time.Now()
|
||||
t.mu.Unlock()
|
||||
|
||||
if connectedBefore {
|
||||
t.reconnectCount.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
package metrics
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestReconnectCount connects three times with a disconnect between each, the
|
||||
// way the streamer does, and expects every connection after the first to count
|
||||
// one reconnect.
|
||||
func TestReconnectCount(t *testing.T) {
|
||||
tracker := New()
|
||||
|
||||
tracker.SetConnected(true)
|
||||
tracker.SetConnected(false)
|
||||
tracker.SetConnected(true)
|
||||
tracker.SetConnected(false)
|
||||
tracker.SetConnected(true)
|
||||
|
||||
const wantReconnects = 2
|
||||
if got := tracker.GetReconnectCount(); got != wantReconnects {
|
||||
t.Fatalf("reconnect count = %d, want %d", got, wantReconnects)
|
||||
}
|
||||
}
|
||||
@@ -40,6 +40,14 @@ func TestNewStreamer(t *testing.T) {
|
||||
if s.metrics != metricsTracker {
|
||||
t.Error("metrics tracker not set correctly")
|
||||
}
|
||||
|
||||
if s.idleTimeout != streamIdleTimeout {
|
||||
t.Errorf("idleTimeout = %s, want %s", s.idleTimeout, streamIdleTimeout)
|
||||
}
|
||||
|
||||
if s.minBackoff != minBackoffDelay {
|
||||
t.Errorf("minBackoff = %s, want %s", s.minBackoff, minBackoffDelay)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamDoesNotLeakTickersAcrossReconnects drives many short-lived
|
||||
@@ -150,6 +158,17 @@ func TestReconnectWhenStreamGoesIdle(t *testing.T) {
|
||||
checkReconnect(t, holdResponse, idleTimeout)
|
||||
}
|
||||
|
||||
// TestReconnectWhenServerNeverAnswers has the server accept the first
|
||||
// request and never answer it, not even with headers. The streamer must give
|
||||
// up on it once the idle timeout passes and connect again.
|
||||
func TestReconnectWhenServerNeverAnswers(t *testing.T) {
|
||||
const idleTimeout = 50 * time.Millisecond
|
||||
neverAnswer := func(_ http.ResponseWriter, r *http.Request) {
|
||||
<-r.Context().Done()
|
||||
}
|
||||
checkReconnect(t, neverAnswer, idleTimeout)
|
||||
}
|
||||
|
||||
// TestStreamNotDroppedWhileLinesArrive has the server send lines well within
|
||||
// the idle timeout for several times the timeout, then end the response. Each
|
||||
// line must start the idle timeout over, so the stream lasts until the server
|
||||
|
||||
Reference in New Issue
Block a user