105 lines
3.2 KiB
Go
105 lines
3.2 KiB
Go
/*
|
|
* Copyright (c) 2021 IBM Corp and others.
|
|
*
|
|
* All rights reserved. This program and the accompanying materials
|
|
* are made available under the terms of the Eclipse Public License v2.0
|
|
* and Eclipse Distribution License v1.0 which accompany this distribution.
|
|
*
|
|
* The Eclipse Public License is available at
|
|
* https://www.eclipse.org/legal/epl-2.0/
|
|
* and the Eclipse Distribution License is available at
|
|
* http://www.eclipse.org/org/documents/edl-v10.php.
|
|
*
|
|
* Contributors:
|
|
* Matt Brittan
|
|
* Daichi Tomaru
|
|
*/
|
|
|
|
package mqtt
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Controller for sleep with backoff when the client attempts reconnection
|
|
// It has statuses for each situations cause reconnection.
|
|
type backoffController struct {
|
|
sync.RWMutex
|
|
statusMap map[string]*backoffStatus
|
|
}
|
|
|
|
type backoffStatus struct {
|
|
lastSleepPeriod time.Duration
|
|
lastErrorTime time.Time
|
|
}
|
|
|
|
func newBackoffController() *backoffController {
|
|
return &backoffController{
|
|
statusMap: map[string]*backoffStatus{},
|
|
}
|
|
}
|
|
|
|
// Calculate next sleep period from the specified parameters.
|
|
// Returned values are next sleep period and whether the error situation is continual.
|
|
// If connection errors continuouslly occurs, its sleep period is exponentially increased.
|
|
// Also if there is a lot of time between last and this error, sleep period is initialized.
|
|
func (b *backoffController) getBackoffSleepTime(
|
|
situation string, initSleepPeriod time.Duration, maxSleepPeriod time.Duration, processTime time.Duration, skipFirst bool,
|
|
) (time.Duration, bool) {
|
|
// Decide first sleep time if the situation is not continual.
|
|
var firstProcess = func(status *backoffStatus, init time.Duration, skip bool) (time.Duration, bool) {
|
|
if skip {
|
|
status.lastSleepPeriod = 0
|
|
return 0, false
|
|
}
|
|
status.lastSleepPeriod = init
|
|
return init, false
|
|
}
|
|
|
|
// Prioritize maxSleep.
|
|
if initSleepPeriod > maxSleepPeriod {
|
|
initSleepPeriod = maxSleepPeriod
|
|
}
|
|
b.Lock()
|
|
defer b.Unlock()
|
|
|
|
status, exist := b.statusMap[situation]
|
|
if !exist {
|
|
b.statusMap[situation] = &backoffStatus{initSleepPeriod, time.Now()}
|
|
return firstProcess(b.statusMap[situation], initSleepPeriod, skipFirst)
|
|
}
|
|
|
|
oldTime := status.lastErrorTime
|
|
status.lastErrorTime = time.Now()
|
|
|
|
// When there is a lot of time between last and this error, sleep period is initialized.
|
|
if status.lastErrorTime.Sub(oldTime) > (processTime * 2 + status.lastSleepPeriod) {
|
|
return firstProcess(status, initSleepPeriod, skipFirst)
|
|
}
|
|
|
|
if status.lastSleepPeriod == 0 {
|
|
status.lastSleepPeriod = initSleepPeriod
|
|
return initSleepPeriod, true
|
|
}
|
|
|
|
if nextSleepPeriod := status.lastSleepPeriod * 2; nextSleepPeriod <= maxSleepPeriod {
|
|
status.lastSleepPeriod = nextSleepPeriod
|
|
} else {
|
|
status.lastSleepPeriod = maxSleepPeriod
|
|
}
|
|
|
|
return status.lastSleepPeriod, true
|
|
}
|
|
|
|
// Execute sleep the time returned from getBackoffSleepTime.
|
|
func (b *backoffController) sleepWithBackoff(
|
|
situation string, initSleepPeriod time.Duration, maxSleepPeriod time.Duration, processTime time.Duration, skipFirst bool,
|
|
) (time.Duration, bool) {
|
|
sleep, isFirst := b.getBackoffSleepTime(situation, initSleepPeriod, maxSleepPeriod, processTime, skipFirst)
|
|
if sleep != 0 {
|
|
time.Sleep(sleep)
|
|
}
|
|
return sleep, isFirst
|
|
}
|