1111828e8a
- persisting a session's protocol state after a request now rewrites an existing record only (Redis SET XX, the memory store likewise) and closes the local copy when the record is gone - before, a request that finished after DELETE, a revoked or deleted subject event or the idle expiry put the record back, and the next request rehydrated the ended session - found by plan 19.7-05's TestFanOut (revocation and deletion cases, about one run in a hundred)
117 lines
2.5 KiB
Go
117 lines
2.5 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
|
|
}
|
|
|
|
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
|
|
}
|