1
0
Fork 0
mirror of https://github.com/binwiederhier/ntfy.git synced 2024-10-31 17:11:13 +01:00
ntfy/server/cache_mem.go

137 lines
3 KiB
Go
Raw Normal View History

2021-11-03 02:09:49 +01:00
package server
import (
"sort"
2021-11-03 02:09:49 +01:00
"sync"
"time"
)
type memCache struct {
messages map[string][]*message
scheduled map[string]*message // Message ID -> message
nop bool
mu sync.Mutex
2021-11-03 02:09:49 +01:00
}
var _ cache = (*memCache)(nil)
2021-12-09 16:23:17 +01:00
// newMemCache creates an in-memory cache
2021-11-03 02:09:49 +01:00
func newMemCache() *memCache {
return &memCache{
messages: make(map[string][]*message),
scheduled: make(map[string]*message),
nop: false,
2021-12-09 16:23:17 +01:00
}
}
// newNopCache creates an in-memory cache that discards all messages;
// it is always empty and can be used if caching is entirely disabled
func newNopCache() *memCache {
return &memCache{
messages: make(map[string][]*message),
scheduled: make(map[string]*message),
nop: true,
2021-11-03 02:09:49 +01:00
}
}
func (c *memCache) AddMessage(m *message) error {
c.mu.Lock()
defer c.mu.Unlock()
if c.nop {
2021-12-09 16:23:17 +01:00
return nil
}
2021-12-07 17:45:15 +01:00
if m.Event != messageEvent {
return errUnexpectedMessageType
}
if _, ok := c.messages[m.Topic]; !ok {
c.messages[m.Topic] = make([]*message, 0)
2021-11-03 02:09:49 +01:00
}
delayed := m.Time > time.Now().Unix()
if delayed {
c.scheduled[m.ID] = m
}
c.messages[m.Topic] = append(c.messages[m.Topic], m)
2021-11-03 02:09:49 +01:00
return nil
}
func (c *memCache) Messages(topic string, since sinceTime, scheduled bool) ([]*message, error) {
c.mu.Lock()
defer c.mu.Unlock()
if _, ok := c.messages[topic]; !ok || since.IsNone() {
2021-11-03 02:09:49 +01:00
return make([]*message, 0), nil
}
messages := make([]*message, 0)
for _, m := range c.messages[topic] {
_, messageScheduled := c.scheduled[m.ID]
include := m.Time >= since.Time().Unix() && (!messageScheduled || scheduled)
if include {
messages = append(messages, m)
}
}
sort.Slice(messages, func(i, j int) bool {
return messages[i].Time < messages[j].Time
})
return messages, nil
}
func (c *memCache) MessagesDue() ([]*message, error) {
c.mu.Lock()
defer c.mu.Unlock()
messages := make([]*message, 0)
for _, m := range c.scheduled {
due := time.Now().Unix() >= m.Time
if due {
2021-11-03 02:09:49 +01:00
messages = append(messages, m)
}
}
sort.Slice(messages, func(i, j int) bool {
return messages[i].Time < messages[j].Time
})
2021-11-03 02:09:49 +01:00
return messages, nil
}
func (c *memCache) MarkPublished(m *message) error {
c.mu.Lock()
delete(c.scheduled, m.ID)
c.mu.Unlock()
return nil
}
func (c *memCache) MessageCount(topic string) (int, error) {
c.mu.Lock()
defer c.mu.Unlock()
if _, ok := c.messages[topic]; !ok {
2021-11-03 02:09:49 +01:00
return 0, nil
}
return len(c.messages[topic]), nil
2021-11-03 02:09:49 +01:00
}
func (c *memCache) Topics() (map[string]*topic, error) {
c.mu.Lock()
defer c.mu.Unlock()
2021-12-09 04:57:31 +01:00
topics := make(map[string]*topic)
for topic := range c.messages {
2021-12-09 04:57:31 +01:00
topics[topic] = newTopic(topic)
}
return topics, nil
2021-11-03 02:09:49 +01:00
}
func (c *memCache) Prune(olderThan time.Time) error {
c.mu.Lock()
defer c.mu.Unlock()
for topic := range c.messages {
c.pruneTopic(topic, olderThan)
2021-11-03 02:09:49 +01:00
}
return nil
}
func (c *memCache) pruneTopic(topic string, olderThan time.Time) {
2021-12-09 04:57:31 +01:00
messages := make([]*message, 0)
for _, m := range c.messages[topic] {
2021-12-09 04:57:31 +01:00
if m.Time >= olderThan.Unix() {
messages = append(messages, m)
2021-11-03 02:09:49 +01:00
}
}
c.messages[topic] = messages
2021-11-03 02:09:49 +01:00
}