Add GetLastMessages, log messages to nested buckets
This commit is contained in:
parent
a6d7403eba
commit
6a25c1d6a5
@ -12,6 +12,7 @@ var messages = Immutable.Map();
|
|||||||
var empty = Immutable.List();
|
var empty = Immutable.List();
|
||||||
|
|
||||||
var Message = Immutable.Record({
|
var Message = Immutable.Record({
|
||||||
|
id: null,
|
||||||
server: null,
|
server: null,
|
||||||
from: null,
|
from: null,
|
||||||
to: null,
|
to: null,
|
||||||
|
File diff suppressed because one or more lines are too long
@ -6,6 +6,7 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"path"
|
"path"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/khlieng/name_pending/Godeps/_workspace/src/github.com/blevesearch/bleve"
|
"github.com/khlieng/name_pending/Godeps/_workspace/src/github.com/blevesearch/bleve"
|
||||||
@ -189,12 +190,13 @@ func (u *User) RemoveChannel(server, channel string) {
|
|||||||
|
|
||||||
func (u *User) LogMessage(server, from, to, content string) {
|
func (u *User) LogMessage(server, from, to, content string) {
|
||||||
go u.messageLog.Update(func(tx *bolt.Tx) error {
|
go u.messageLog.Update(func(tx *bolt.Tx) error {
|
||||||
b := tx.Bucket(bucketMessages)
|
bucketKey := server + ":" + to
|
||||||
messageID, _ := b.NextSequence()
|
b, _ := tx.Bucket(bucketMessages).CreateBucketIfNotExists([]byte(bucketKey))
|
||||||
id := server + ":" + to + ":" + strconv.FormatUint(messageID, 10)
|
id, _ := b.NextSequence()
|
||||||
|
idStr := strconv.FormatUint(id, 10)
|
||||||
|
|
||||||
message := Message{
|
message := Message{
|
||||||
ID: messageID,
|
ID: id,
|
||||||
Content: content,
|
Content: content,
|
||||||
Server: server,
|
Server: server,
|
||||||
From: from,
|
From: from,
|
||||||
@ -203,34 +205,65 @@ func (u *User) LogMessage(server, from, to, content string) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
data, _ := json.Marshal(message)
|
data, _ := json.Marshal(message)
|
||||||
b.Put([]byte(id), data)
|
b.Put([]byte(idStr), data)
|
||||||
|
|
||||||
go u.messageIndex.Index(id, message)
|
go u.messageIndex.Index(bucketKey+":"+idStr, message)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (u *User) GetLastMessages(server, channel string, count int) ([]Message, error) {
|
||||||
|
messages := make([]Message, count)
|
||||||
|
|
||||||
|
u.messageLog.View(func(tx *bolt.Tx) error {
|
||||||
|
b := tx.Bucket(bucketMessages).Bucket([]byte(server + ":" + channel))
|
||||||
|
if b == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
c := b.Cursor()
|
||||||
|
|
||||||
|
for k, v := c.Last(); count > 0 && k != nil; k, v = c.Prev() {
|
||||||
|
count--
|
||||||
|
json.Unmarshal(v, &messages[count])
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
|
if count < len(messages) {
|
||||||
|
return messages[count:], nil
|
||||||
|
} else {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (u *User) GetMessages(server, channel string, count int, fromID uint64) ([]Message, error) {
|
func (u *User) GetMessages(server, channel string, count int, fromID uint64) ([]Message, error) {
|
||||||
messages := make([]Message, count)
|
messages := make([]Message, count)
|
||||||
i := count - 1
|
|
||||||
prefix := []byte(server + ":" + channel + ":" + strconv.FormatUint(fromID, 10))
|
|
||||||
|
|
||||||
u.messageLog.View(func(tx *bolt.Tx) error {
|
u.messageLog.View(func(tx *bolt.Tx) error {
|
||||||
c := tx.Bucket(bucketMessages).Cursor()
|
b := tx.Bucket(bucketMessages).Bucket([]byte(server + ":" + channel))
|
||||||
|
if b == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
for k, v := c.Seek(prefix); i > 0 && bytes.HasPrefix(k, prefix); k, v = c.Prev() {
|
c := b.Cursor()
|
||||||
var message Message
|
c.Seek([]byte(strconv.FormatUint(fromID, 10)))
|
||||||
|
|
||||||
json.Unmarshal(v, &message)
|
for k, v := c.Prev(); count > 0 && k != nil; k, v = c.Prev() {
|
||||||
messages[i] = message
|
count--
|
||||||
i--
|
json.Unmarshal(v, &messages[count])
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|
||||||
return messages[i:], nil
|
if count < len(messages) {
|
||||||
|
return messages[count:], nil
|
||||||
|
} else {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (u *User) SearchMessages(server, channel, phrase string) ([]Message, error) {
|
func (u *User) SearchMessages(server, channel, phrase string) ([]Message, error) {
|
||||||
@ -249,16 +282,16 @@ func (u *User) SearchMessages(server, channel, phrase string) ([]Message, error)
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
log.Printf("%.3fms\n", searchResults.Took.Seconds()*1000)
|
|
||||||
|
|
||||||
messages := []Message{}
|
messages := []Message{}
|
||||||
u.messageLog.View(func(tx *bolt.Tx) error {
|
u.messageLog.View(func(tx *bolt.Tx) error {
|
||||||
b := tx.Bucket(bucketMessages)
|
b := tx.Bucket(bucketMessages)
|
||||||
|
|
||||||
for _, hit := range searchResults.Hits {
|
for _, hit := range searchResults.Hits {
|
||||||
|
idx := strings.LastIndex(hit.ID, ":")
|
||||||
|
bc := b.Bucket([]byte(hit.ID[:idx]))
|
||||||
var message Message
|
var message Message
|
||||||
|
|
||||||
json.Unmarshal(b.Get([]byte(hit.ID)), &message)
|
json.Unmarshal(bc.Get([]byte(hit.ID[idx+1:])), &message)
|
||||||
messages = append(messages, message)
|
messages = append(messages, message)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user