c1c0c77ddc
- an unexported reserver (Reserve, Release) next to updater: the memory store reserves under its mutex, the Redis store in one Lua script (prune, SCARD, SET PX plus SADD); serveNew reserves instead of Count then Put and releases on a failed connect or initialize (A-WR-03) - serveNew, lookup's rehydration and the stateless getServer build the SDK server with no router-wide lock held and insert or adopt under the lock afterwards, so a slow guard or database check no longer stalls the instance (A-WR-07)
143 lines
3.1 KiB
Go
143 lines
3.1 KiB
Go
package pier
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// memoryStore keeps records in the process with lazy expiry and fans events
|
|
// out to in-process subscribers; the default store, for one instance.
|
|
type memoryStore struct {
|
|
mu sync.Mutex
|
|
records map[string]memoryRecord
|
|
subs []func(Event)
|
|
now func() time.Time
|
|
}
|
|
|
|
type memoryRecord struct {
|
|
rec Record
|
|
expires time.Time
|
|
}
|
|
|
|
func newMemoryStore() *memoryStore {
|
|
return &memoryStore{records: map[string]memoryRecord{}, now: time.Now}
|
|
}
|
|
|
|
func (m *memoryStore) sweepLocked() {
|
|
now := m.now()
|
|
for id, r := range m.records {
|
|
if !r.expires.After(now) {
|
|
delete(m.records, id)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *memoryStore) Put(_ context.Context, id string, rec Record, ttl time.Duration) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.sweepLocked()
|
|
m.records[id] = memoryRecord{rec: rec, expires: m.now().Add(ttl)}
|
|
return nil
|
|
}
|
|
|
|
func (m *memoryStore) Get(_ context.Context, id string) (Record, bool, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
r, ok := m.records[id]
|
|
if !ok || !r.expires.After(m.now()) {
|
|
delete(m.records, id)
|
|
return Record{}, false, nil
|
|
}
|
|
return r.rec, true, nil
|
|
}
|
|
|
|
// Update rewrites an existing record; a missing or expired one stays gone.
|
|
func (m *memoryStore) Update(_ context.Context, id string, rec Record, ttl time.Duration) (bool, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
r, ok := m.records[id]
|
|
if !ok || !r.expires.After(m.now()) {
|
|
delete(m.records, id)
|
|
return false, nil
|
|
}
|
|
m.records[id] = memoryRecord{rec: rec, expires: m.now().Add(ttl)}
|
|
return true, nil
|
|
}
|
|
|
|
func (m *memoryStore) Touch(_ context.Context, id string, ttl time.Duration) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
r, ok := m.records[id]
|
|
if !ok || !r.expires.After(m.now()) {
|
|
delete(m.records, id)
|
|
return nil
|
|
}
|
|
r.expires = m.now().Add(ttl)
|
|
m.records[id] = r
|
|
return nil
|
|
}
|
|
|
|
func (m *memoryStore) Delete(_ context.Context, id string) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
delete(m.records, id)
|
|
return nil
|
|
}
|
|
|
|
func (m *memoryStore) Count(_ context.Context, binding string) (int, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.sweepLocked()
|
|
n := 0
|
|
for _, r := range m.records {
|
|
if r.rec.Binding == binding {
|
|
n++
|
|
}
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// Reserve counts the binding's live records and writes rec under one lock.
|
|
func (m *memoryStore) Reserve(_ context.Context, binding, id string, rec Record, max int, ttl time.Duration) (bool, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.sweepLocked()
|
|
n := 0
|
|
for _, r := range m.records {
|
|
if r.rec.Binding == binding {
|
|
n++
|
|
}
|
|
}
|
|
if n >= max {
|
|
return false, nil
|
|
}
|
|
m.records[id] = memoryRecord{rec: rec, expires: m.now().Add(ttl)}
|
|
return true, nil
|
|
}
|
|
|
|
// Release drops a reservation.
|
|
func (m *memoryStore) Release(_ context.Context, _ string, id string) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
delete(m.records, id)
|
|
return nil
|
|
}
|
|
|
|
func (m *memoryStore) Publish(_ context.Context, ev Event) error {
|
|
m.mu.Lock()
|
|
subs := append([]func(Event){}, m.subs...)
|
|
m.mu.Unlock()
|
|
for _, fn := range subs {
|
|
go fn(ev)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (m *memoryStore) Subscribe(_ context.Context, fn func(Event)) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.subs = append(m.subs, fn)
|
|
return nil
|
|
}
|