Ilia Denisov
49 lines
1.5 KiB
Go
49 lines
1.5 KiB
Go
package push
|
|
|
|
import (
|
|
"fmt"
|
|
"strconv"
|
|
"sync/atomic"
|
|
)
|
|
|
|
// cursorWidth is the zero-padded decimal width applied to every cursor.
|
|
// 20 digits accommodate the full uint64 range so lexicographic order
|
|
// matches numeric order across the entire process lifetime.
|
|
const cursorWidth = 20
|
|
|
|
// cursorGenerator hands out monotonically increasing uint64 sequence
|
|
// numbers. Cursors restart from 0 on process boot; the ring buffer's
|
|
// freshness-window TTL bounds how long a cursor remains valid, so a
|
|
// fresh process intentionally invalidates every previously-issued
|
|
// cursor.
|
|
type cursorGenerator struct {
|
|
seq atomic.Uint64
|
|
}
|
|
|
|
// next returns the next sequence number. The first call returns 1.
|
|
func (g *cursorGenerator) next() uint64 {
|
|
return g.seq.Add(1)
|
|
}
|
|
|
|
// formatCursor renders n in the canonical zero-padded form so cursor
|
|
// strings sort identically to their numeric counterparts.
|
|
func formatCursor(n uint64) string {
|
|
return fmt.Sprintf("%0*d", cursorWidth, n)
|
|
}
|
|
|
|
// parseCursor decodes a cursor string back to its numeric value. An
|
|
// empty string maps to 0 ("subscribe from now"); malformed input also
|
|
// maps to 0 with ok=false so callers can log without rejecting the
|
|
// subscription — gateway is trusted but reconnects can race against a
|
|
// process restart that scrambled the in-memory sequence.
|
|
func parseCursor(s string) (uint64, bool) {
|
|
if s == "" {
|
|
return 0, true
|
|
}
|
|
n, err := strconv.ParseUint(s, 10, 64)
|
|
if err != nil {
|
|
return 0, false
|
|
}
|
|
return n, true
|
|
}
|