[442] | 1 | package soju
|
---|
| 2 |
|
---|
| 3 | import (
|
---|
| 4 | "fmt"
|
---|
| 5 | "time"
|
---|
| 6 |
|
---|
[488] | 7 | "git.sr.ht/~sircmpwn/go-bare"
|
---|
[442] | 8 | "gopkg.in/irc.v3"
|
---|
| 9 | )
|
---|
| 10 |
|
---|
| 11 | const messageRingBufferCap = 4096
|
---|
| 12 |
|
---|
[488] | 13 | type memoryMsgID struct {
|
---|
| 14 | Seq bare.Uint
|
---|
| 15 | }
|
---|
| 16 |
|
---|
| 17 | func (memoryMsgID) msgIDType() msgIDType {
|
---|
| 18 | return msgIDMemory
|
---|
| 19 | }
|
---|
| 20 |
|
---|
[442] | 21 | func parseMemoryMsgID(s string) (netID int64, entity string, seq uint64, err error) {
|
---|
[488] | 22 | var id memoryMsgID
|
---|
| 23 | netID, entity, err = parseMsgID(s, &id)
|
---|
[442] | 24 | if err != nil {
|
---|
| 25 | return 0, "", 0, err
|
---|
| 26 | }
|
---|
[488] | 27 | return netID, entity, uint64(id.Seq), nil
|
---|
[442] | 28 | }
|
---|
| 29 |
|
---|
| 30 | func formatMemoryMsgID(netID int64, entity string, seq uint64) string {
|
---|
[488] | 31 | id := memoryMsgID{bare.Uint(seq)}
|
---|
| 32 | return formatMsgID(netID, entity, &id)
|
---|
[442] | 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
|
---|
[444] | 103 | cur uint64
|
---|
[442] | 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 | }
|
---|