1 | package soju
|
---|
2 |
|
---|
3 | import (
|
---|
4 | "fmt"
|
---|
5 | "time"
|
---|
6 |
|
---|
7 | "git.sr.ht/~sircmpwn/go-bare"
|
---|
8 | "gopkg.in/irc.v3"
|
---|
9 | )
|
---|
10 |
|
---|
11 | const messageRingBufferCap = 4096
|
---|
12 |
|
---|
13 | type memoryMsgID struct {
|
---|
14 | Seq bare.Uint
|
---|
15 | }
|
---|
16 |
|
---|
17 | func (memoryMsgID) msgIDType() msgIDType {
|
---|
18 | return msgIDMemory
|
---|
19 | }
|
---|
20 |
|
---|
21 | func parseMemoryMsgID(s string) (netID int64, entity string, seq uint64, err error) {
|
---|
22 | var id memoryMsgID
|
---|
23 | netID, entity, err = parseMsgID(s, &id)
|
---|
24 | if err != nil {
|
---|
25 | return 0, "", 0, err
|
---|
26 | }
|
---|
27 | return netID, entity, uint64(id.Seq), nil
|
---|
28 | }
|
---|
29 |
|
---|
30 | func formatMemoryMsgID(netID int64, entity string, seq uint64) string {
|
---|
31 | id := memoryMsgID{bare.Uint(seq)}
|
---|
32 | return formatMsgID(netID, entity, &id)
|
---|
33 | }
|
---|
34 |
|
---|
35 | type ringBufferKey struct {
|
---|
36 | networkID int64
|
---|
37 | entity string
|
---|
38 | }
|
---|
39 |
|
---|
40 | type memoryMessageStore struct {
|
---|
41 | buffers map[ringBufferKey]*messageRingBuffer
|
---|
42 | }
|
---|
43 |
|
---|
44 | func newMemoryMessageStore() *memoryMessageStore {
|
---|
45 | return &memoryMessageStore{
|
---|
46 | buffers: make(map[ringBufferKey]*messageRingBuffer),
|
---|
47 | }
|
---|
48 | }
|
---|
49 |
|
---|
50 | func (ms *memoryMessageStore) Close() error {
|
---|
51 | ms.buffers = nil
|
---|
52 | return nil
|
---|
53 | }
|
---|
54 |
|
---|
55 | func (ms *memoryMessageStore) get(network *network, entity string) *messageRingBuffer {
|
---|
56 | k := ringBufferKey{networkID: network.ID, entity: entity}
|
---|
57 | if rb, ok := ms.buffers[k]; ok {
|
---|
58 | return rb
|
---|
59 | }
|
---|
60 | rb := newMessageRingBuffer(messageRingBufferCap)
|
---|
61 | ms.buffers[k] = rb
|
---|
62 | return rb
|
---|
63 | }
|
---|
64 |
|
---|
65 | func (ms *memoryMessageStore) LastMsgID(network *network, entity string, t time.Time) (string, error) {
|
---|
66 | var seq uint64
|
---|
67 | k := ringBufferKey{networkID: network.ID, entity: entity}
|
---|
68 | if rb, ok := ms.buffers[k]; ok {
|
---|
69 | seq = rb.cur
|
---|
70 | }
|
---|
71 | return formatMemoryMsgID(network.ID, entity, seq), nil
|
---|
72 | }
|
---|
73 |
|
---|
74 | func (ms *memoryMessageStore) Append(network *network, entity string, msg *irc.Message) (string, error) {
|
---|
75 | k := ringBufferKey{networkID: network.ID, entity: entity}
|
---|
76 | rb, ok := ms.buffers[k]
|
---|
77 | if !ok {
|
---|
78 | rb = newMessageRingBuffer(messageRingBufferCap)
|
---|
79 | ms.buffers[k] = rb
|
---|
80 | }
|
---|
81 |
|
---|
82 | seq := rb.Append(msg)
|
---|
83 | return formatMemoryMsgID(network.ID, entity, seq), nil
|
---|
84 | }
|
---|
85 |
|
---|
86 | func (ms *memoryMessageStore) LoadLatestID(network *network, entity, id string, limit int) ([]*irc.Message, error) {
|
---|
87 | _, _, seq, err := parseMemoryMsgID(id)
|
---|
88 | if err != nil {
|
---|
89 | return nil, err
|
---|
90 | }
|
---|
91 |
|
---|
92 | k := ringBufferKey{networkID: network.ID, entity: entity}
|
---|
93 | rb, ok := ms.buffers[k]
|
---|
94 | if !ok {
|
---|
95 | return nil, nil
|
---|
96 | }
|
---|
97 |
|
---|
98 | return rb.LoadLatestSeq(seq, limit)
|
---|
99 | }
|
---|
100 |
|
---|
101 | type messageRingBuffer struct {
|
---|
102 | buf []*irc.Message
|
---|
103 | cur uint64
|
---|
104 | }
|
---|
105 |
|
---|
106 | func newMessageRingBuffer(capacity int) *messageRingBuffer {
|
---|
107 | return &messageRingBuffer{
|
---|
108 | buf: make([]*irc.Message, capacity),
|
---|
109 | cur: 1,
|
---|
110 | }
|
---|
111 | }
|
---|
112 |
|
---|
113 | func (rb *messageRingBuffer) cap() uint64 {
|
---|
114 | return uint64(len(rb.buf))
|
---|
115 | }
|
---|
116 |
|
---|
117 | func (rb *messageRingBuffer) Append(msg *irc.Message) uint64 {
|
---|
118 | seq := rb.cur
|
---|
119 | i := int(seq % rb.cap())
|
---|
120 | rb.buf[i] = msg
|
---|
121 | rb.cur++
|
---|
122 | return seq
|
---|
123 | }
|
---|
124 |
|
---|
125 | func (rb *messageRingBuffer) LoadLatestSeq(seq uint64, limit int) ([]*irc.Message, error) {
|
---|
126 | if seq > rb.cur {
|
---|
127 | return nil, fmt.Errorf("loading messages from sequence number (%v) greater than current (%v)", seq, rb.cur)
|
---|
128 | } else if seq == rb.cur {
|
---|
129 | return nil, nil
|
---|
130 | }
|
---|
131 |
|
---|
132 | // The query excludes the message with the sequence number seq
|
---|
133 | diff := rb.cur - seq - 1
|
---|
134 | if diff > rb.cap() {
|
---|
135 | // We dropped diff - cap entries
|
---|
136 | diff = rb.cap()
|
---|
137 | }
|
---|
138 | if int(diff) > limit {
|
---|
139 | diff = uint64(limit)
|
---|
140 | }
|
---|
141 |
|
---|
142 | l := make([]*irc.Message, int(diff))
|
---|
143 | for i := 0; i < int(diff); i++ {
|
---|
144 | j := int((rb.cur - diff + uint64(i)) % rb.cap())
|
---|
145 | l[i] = rb.buf[j]
|
---|
146 | }
|
---|
147 |
|
---|
148 | return l, nil
|
---|
149 | }
|
---|