status-im
/
go-waku
mirror of https://github.com/status-im/go-waku.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Enabling pubsub topic filter in history queries 

Fixes #19
This commit is contained in:
Richard Ramos 2021-04-28 11:11:32 -04:00
parent 4a4d0c97ec
commit aee86211d1
No known key found for this signature in database
GPG Key ID: 80D4B01265FDFE8F
6 changed files with 171 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
waku/persistence/store.go
View File

@ -4,6 +4,8 @@ import (
"database/sql" "database/sql"
"log" "log"
gcrypto "github.com/ethereum/go-ethereum/crypto"
"github.com/status-im/go-waku/waku/v2/protocol"
"github.com/status-im/go-waku/waku/v2/protocol/pb" "github.com/status-im/go-waku/waku/v2/protocol/pb"
"github.com/status-im/go-waku/waku/v2/protocol/store" "github.com/status-im/go-waku/waku/v2/protocol/store"
) )
@ -55,10 +57,11 @@ func NewDBStore(opt DBOption) (*DBStore, error) {
} }
func (d *DBStore) createTable() error { func (d *DBStore) createTable() error {
sqlStmt := `CREATE TABLE IF NOT EXISTS messages ( sqlStmt := `CREATE TABLE IF NOT EXISTS message (
id BLOB PRIMARY KEY, id BLOB PRIMARY KEY,
timestamp INTEGER NOT NULL, timestamp INTEGER NOT NULL,
contentTopic BLOB NOT NULL, contentTopic BLOB NOT NULL,
pubsubTopic BLOB NOT NULL,
payload BLOB, payload BLOB,
version INTEGER NOT NULL DEFAULT 0 version INTEGER NOT NULL DEFAULT 0
) WITHOUT ROWID;` ) WITHOUT ROWID;`
@ -75,12 +78,12 @@ func (d *DBStore) Stop() {
} }
// Inserts a WakuMessage into the DB // Inserts a WakuMessage into the DB
func (d *DBStore) Put(cursor *pb.Index, message *pb.WakuMessage) error { func (d *DBStore) Put(cursor *pb.Index, pubsubTopic string, message *pb.WakuMessage) error {
stmt, err := d.db.Prepare("INSERT INTO messages (id, timestamp, contentTopic, payload, version) VALUES (?, ?, ?, ?, ?)") stmt, err := d.db.Prepare("INSERT INTO message (id, timestamp, contentTopic, pubsubTopic, payload, version) VALUES (?, ?, ?, ?, ?, ?)")
if err != nil { if err != nil {
return err return err
} }
_, err = stmt.Exec(cursor.Digest, uint64(message.Timestamp), message.ContentTopic, message.Payload, message.Version) _, err = stmt.Exec(cursor.Digest, uint64(message.Timestamp), message.ContentTopic, pubsubTopic, message.Payload, message.Version)
if err != nil { if err != nil {
return err return err
} }
@ -89,13 +92,13 @@ func (d *DBStore) Put(cursor *pb.Index, message *pb.WakuMessage) error {
} }
// Returns all the stored WakuMessages // Returns all the stored WakuMessages
func (d *DBStore) GetAll() ([]*pb.WakuMessage, error) { func (d *DBStore) GetAll() ([]*protocol.Envelope, error) {
rows, err := d.db.Query("SELECT timestamp, contentTopic, payload, version FROM messages ORDER BY timestamp ASC") rows, err := d.db.Query("SELECT timestamp, contentTopic, pubsubTopic, payload, version FROM message ORDER BY timestamp ASC")
if err != nil { if err != nil {
return nil, err return nil, err
} }
var result []*pb.WakuMessage var result []*protocol.Envelope
defer rows.Close() defer rows.Close()
@ -104,8 +107,9 @@ func (d *DBStore) GetAll() ([]*pb.WakuMessage, error) {
var contentTopic string var contentTopic string
var payload []byte var payload []byte
var version uint32 var version uint32
var pubsubTopic string
err = rows.Scan(&timestamp, &contentTopic, &payload, &version) err = rows.Scan(&timestamp, &contentTopic, &pubsubTopic, &payload, &version)
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
@ -116,7 +120,9 @@ func (d *DBStore) GetAll() ([]*pb.WakuMessage, error) {
msg.Timestamp = float64(timestamp) msg.Timestamp = float64(timestamp)
msg.Version = version msg.Version = version
result = append(result, msg) data, _ := msg.Marshal()
envelope := protocol.NewEnvelope(msg, pubsubTopic, len(data), gcrypto.Keccak256(data))
result = append(result, envelope)
} }
err = rows.Err() err = rows.Err()

6
waku/v2/node/wakunode2.go
View File

@ -250,7 +250,7 @@ func (node *WakuNode) Subscribe(topic *Topic) (*Subscription, error) {
node.bcaster.Register(subscription.C) node.bcaster.Register(subscription.C)
go func() { go func(t Topic) {
nextMsgTicker := time.NewTicker(time.Millisecond * 10) nextMsgTicker := time.NewTicker(time.Millisecond * 10)
defer nextMsgTicker.Stop() defer nextMsgTicker.Stop()
@ -280,12 +280,12 @@ func (node *WakuNode) Subscribe(topic *Topic) (*Subscription, error) {
return return
} }
envelope := protocol.NewEnvelope(wakuMessage, len(msg.Data), gcrypto.Keccak256(msg.Data)) envelope := protocol.NewEnvelope(wakuMessage, string(t), len(msg.Data), gcrypto.Keccak256(msg.Data))
node.bcaster.Submit(envelope) node.bcaster.Submit(envelope)
} }
} }
}() }(t)
return subscription, nil return subscription, nil
} }

8
waku/v2/protocol/envelope.go
View File

@ -4,13 +4,15 @@ import "github.com/status-im/go-waku/waku/v2/protocol/pb"
type Envelope struct { type Envelope struct {
msg *pb.WakuMessage msg *pb.WakuMessage
pubsubTopic string
size int size int
hash []byte hash []byte
} }
func NewEnvelope(msg *pb.WakuMessage, size int, hash []byte) *Envelope { func NewEnvelope(msg *pb.WakuMessage, pubSubTopic string, size int, hash []byte) *Envelope {
return &Envelope{ return &Envelope{
msg: msg, msg: msg,
pubsubTopic: pubSubTopic,
size: size, size: size,
hash: hash, hash: hash,
} }
@ -20,6 +22,10 @@ func (e *Envelope) Message() *pb.WakuMessage {
return e.msg return e.msg
} }
func (e *Envelope) PubsubTopic() string {
return e.pubsubTopic
}
func (e *Envelope) Hash() []byte { func (e *Envelope) Hash() []byte {
return e.hash return e.hash
} }

150
waku/v2/protocol/pb/waku_store.pb.go
View File

@ -106,7 +106,7 @@ func (m *Index) GetReceivedTime() float64 {
type PagingInfo struct { type PagingInfo struct {
PageSize uint64 `protobuf:"varint,1,opt,name=pageSize,proto3" json:"pageSize,omitempty"` PageSize uint64 `protobuf:"varint,1,opt,name=pageSize,proto3" json:"pageSize,omitempty"`
Cursor *Index `protobuf:"bytes,2,opt,name=cursor,proto3" json:"cursor,omitempty"` Cursor *Index `protobuf:"bytes,2,opt,name=cursor,proto3" json:"cursor,omitempty"`
Direction PagingInfo_Direction `protobuf:"varint,3,opt,name=direction,proto3,enum=protocol.PagingInfo_Direction" json:"direction,omitempty"` Direction PagingInfo_Direction `protobuf:"varint,3,opt,name=direction,proto3,enum=pb.PagingInfo_Direction" json:"direction,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"` XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"` XXX_sizecache int32 `json:"-"`
@ -214,10 +214,11 @@ func (m *ContentFilter) GetContentTopic() string {
} }
type HistoryQuery struct { type HistoryQuery struct {
ContentFilters []*ContentFilter `protobuf:"bytes,2,rep,name=contentFilters,proto3" json:"contentFilters,omitempty"` PubsubTopic string `protobuf:"bytes,2,opt,name=pubsubTopic,proto3" json:"pubsubTopic,omitempty"`
PagingInfo *PagingInfo `protobuf:"bytes,3,opt,name=pagingInfo,proto3" json:"pagingInfo,omitempty"` ContentFilters []*ContentFilter `protobuf:"bytes,3,rep,name=contentFilters,proto3" json:"contentFilters,omitempty"`
StartTime float64 `protobuf:"fixed64,4,opt,name=startTime,proto3" json:"startTime,omitempty"` PagingInfo *PagingInfo `protobuf:"bytes,4,opt,name=pagingInfo,proto3" json:"pagingInfo,omitempty"`
EndTime float64 `protobuf:"fixed64,5,opt,name=endTime,proto3" json:"endTime,omitempty"` StartTime float64 `protobuf:"fixed64,5,opt,name=startTime,proto3" json:"startTime,omitempty"`
EndTime float64 `protobuf:"fixed64,6,opt,name=endTime,proto3" json:"endTime,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"` XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"` XXX_sizecache int32 `json:"-"`
@ -256,6 +257,13 @@ func (m *HistoryQuery) XXX_DiscardUnknown() {
var xxx_messageInfo_HistoryQuery proto.InternalMessageInfo var xxx_messageInfo_HistoryQuery proto.InternalMessageInfo
func (m *HistoryQuery) GetPubsubTopic() string {
if m != nil {
return m.PubsubTopic
}
return ""
}
func (m *HistoryQuery) GetContentFilters() []*ContentFilter { func (m *HistoryQuery) GetContentFilters() []*ContentFilter {
if m != nil { if m != nil {
return m.ContentFilters return m.ContentFilters
@ -403,48 +411,49 @@ func (m *HistoryRPC) GetResponse() *HistoryResponse {
} }
func init() { func init() {
proto.RegisterEnum("protocol.PagingInfo_Direction", PagingInfo_Direction_name, PagingInfo_Direction_value) proto.RegisterEnum("pb.PagingInfo_Direction", PagingInfo_Direction_name, PagingInfo_Direction_value)
proto.RegisterType((*Index)(nil), "protocol.Index") proto.RegisterType((*Index)(nil), "pb.Index")
proto.RegisterType((*PagingInfo)(nil), "protocol.PagingInfo") proto.RegisterType((*PagingInfo)(nil), "pb.PagingInfo")
proto.RegisterType((*ContentFilter)(nil), "protocol.ContentFilter") proto.RegisterType((*ContentFilter)(nil), "pb.ContentFilter")
proto.RegisterType((*HistoryQuery)(nil), "protocol.HistoryQuery") proto.RegisterType((*HistoryQuery)(nil), "pb.HistoryQuery")
proto.RegisterType((*HistoryResponse)(nil), "protocol.HistoryResponse") proto.RegisterType((*HistoryResponse)(nil), "pb.HistoryResponse")
proto.RegisterType((*HistoryRPC)(nil), "protocol.HistoryRPC") proto.RegisterType((*HistoryRPC)(nil), "pb.HistoryRPC")
} }
func init() { proto.RegisterFile("waku_store.proto", fileDescriptor_ca6891f77a46e680) } func init() { proto.RegisterFile("waku_store.proto", fileDescriptor_ca6891f77a46e680) }
var fileDescriptor_ca6891f77a46e680 = []byte{ var fileDescriptor_ca6891f77a46e680 = []byte{
// 464 bytes of a gzipped FileDescriptorProto // 475 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x52, 0xc1, 0x6e, 0xd3, 0x40, 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x84, 0x93, 0xcf, 0x6e, 0xd3, 0x40,
0x10, 0xed, 0xb6, 0x4d, 0x6a, 0x4f, 0x42, 0x1b, 0x8d, 0xa0, 0x98, 0x0a, 0xac, 0xc8, 0x07, 0xc8, 0x10, 0xc6, 0xbb, 0x49, 0x93, 0xc6, 0xe3, 0x90, 0x86, 0x41, 0x42, 0x56, 0x05, 0x91, 0xf1, 0xa1,
0x01, 0x45, 0x22, 0x85, 0x1b, 0x12, 0x6a, 0x53, 0x55, 0x44, 0x08, 0x51, 0x96, 0x4a, 0x3d, 0x56, 0x8a, 0x84, 0x64, 0xa4, 0x54, 0x42, 0xe2, 0xd8, 0xa6, 0xaa, 0x88, 0x10, 0xa2, 0x2c, 0x95, 0x7a,
0xc6, 0x1e, 0xac, 0x55, 0x5b, 0xaf, 0xbb, 0xbb, 0x06, 0xda, 0xaf, 0xe0, 0xc8, 0x9f, 0x70, 0xe2, 0xac, 0xfc, 0x67, 0xb0, 0x56, 0xa5, 0xb6, 0xbb, 0xbb, 0x06, 0xca, 0x99, 0x87, 0xe0, 0x1d, 0x78,
0xce, 0x91, 0x4f, 0x40, 0xe1, 0x47, 0x50, 0x36, 0x6b, 0x3b, 0x29, 0x1c, 0x38, 0x59, 0xf3, 0xe6, 0x11, 0x8e, 0x48, 0xbc, 0x00, 0x0a, 0x2f, 0x82, 0xbc, 0x76, 0x1c, 0x27, 0x97, 0x1e, 0xe7, 0x37,
0xed, 0xcc, 0x7b, 0x6f, 0x0c, 0xbd, 0x4f, 0xf1, 0x59, 0x79, 0xaa, 0x8d, 0x54, 0x34, 0x2c, 0x94, 0x9f, 0x67, 0xbe, 0x99, 0x59, 0xc3, 0xf8, 0x4b, 0x70, 0x5d, 0x5c, 0x29, 0x9d, 0x49, 0xf2, 0x73,
0x34, 0x12, 0x3d, 0xfb, 0x49, 0xe4, 0xf9, 0x0e, 0xda, 0xde, 0x05, 0x69, 0x1d, 0x67, 0xae, 0x1b, 0x99, 0xe9, 0x0c, 0x3b, 0x79, 0x78, 0x80, 0x86, 0xde, 0x90, 0x52, 0x41, 0x52, 0x73, 0x6f, 0x0e,
0x8d, 0xa1, 0x35, 0xc9, 0x53, 0xfa, 0x8c, 0xdb, 0xd0, 0x4e, 0x45, 0x46, 0xda, 0x04, 0xac, 0xcf, 0xbd, 0x45, 0x1a, 0xd3, 0x57, 0x7c, 0x0c, 0xfd, 0x58, 0x24, 0xa4, 0xb4, 0xc3, 0x5c, 0x36, 0x1d,
0x06, 0x5d, 0xee, 0x2a, 0x8c, 0xa0, 0xab, 0x28, 0x21, 0xf1, 0x91, 0xd2, 0x63, 0x71, 0x41, 0xc1, 0xf2, 0x3a, 0x42, 0x0f, 0x86, 0x92, 0x22, 0x12, 0x9f, 0x29, 0xbe, 0x10, 0x37, 0xe4, 0x74, 0x5c,
0x6a, 0x9f, 0x0d, 0x18, 0x5f, 0xc2, 0xa2, 0x6f, 0x0c, 0xe0, 0x28, 0xce, 0x44, 0x9e, 0x4d, 0xf2, 0x36, 0x65, 0x7c, 0x83, 0x79, 0x3f, 0x19, 0xc0, 0x79, 0x90, 0x88, 0x34, 0x59, 0xa4, 0x1f, 0x33,
0x0f, 0x12, 0x77, 0xc0, 0x2b, 0xe2, 0x8c, 0xde, 0x89, 0x6b, 0xb2, 0xc3, 0xd6, 0x79, 0x5d, 0xe3, 0x3c, 0x80, 0x41, 0x1e, 0x24, 0xf4, 0x41, 0x7c, 0x23, 0x53, 0x6c, 0x97, 0x37, 0x31, 0x3e, 0x83,
0x23, 0x68, 0x27, 0xa5, 0xd2, 0x52, 0xd9, 0x41, 0x9d, 0xd1, 0xd6, 0xb0, 0x92, 0x37, 0xb4, 0x3a, 0x7e, 0x54, 0x48, 0x95, 0x49, 0x53, 0xc8, 0x9e, 0x59, 0x7e, 0x1e, 0xfa, 0xc6, 0x01, 0xaf, 0x13,
0xb8, 0x6b, 0xe3, 0x73, 0xf0, 0x53, 0xa1, 0x28, 0x31, 0x42, 0xe6, 0xc1, 0x5a, 0x9f, 0x0d, 0x36, 0xf8, 0x12, 0xac, 0x58, 0x48, 0x8a, 0xb4, 0xc8, 0x52, 0xa7, 0xeb, 0xb2, 0xe9, 0x68, 0xe6, 0x94,
0x47, 0x61, 0xc3, 0x6d, 0xb6, 0x0d, 0x0f, 0x2a, 0x16, 0x6f, 0x1e, 0x44, 0x0f, 0xc1, 0xaf, 0x71, 0xaa, 0x75, 0x07, 0xff, 0x74, 0x95, 0xe7, 0x6b, 0xa9, 0x77, 0x08, 0x56, 0xc3, 0x71, 0x08, 0x83,
0xec, 0x82, 0xb7, 0xbf, 0x37, 0x7e, 0x75, 0xb2, 0xc7, 0x0f, 0x7a, 0x2b, 0xd8, 0x81, 0x8d, 0xc3, 0x93, 0xe3, 0xf9, 0x9b, 0xcb, 0x63, 0x7e, 0x3a, 0xde, 0x41, 0x1b, 0xf6, 0xce, 0xde, 0x71, 0x13,
0x37, 0xdc, 0x16, 0x2c, 0xda, 0x85, 0x5b, 0x63, 0x99, 0x1b, 0xca, 0xcd, 0xa1, 0x38, 0x37, 0xa4, 0x30, 0xef, 0x08, 0x1e, 0xcc, 0xb3, 0x54, 0x53, 0xaa, 0xcf, 0xc4, 0x27, 0x4d, 0xb2, 0x1c, 0x31,
0x66, 0x76, 0x93, 0x39, 0x70, 0x2c, 0x0b, 0x91, 0x58, 0xfd, 0x3e, 0x5f, 0xc2, 0xa2, 0xef, 0x0c, 0xaa, 0xc0, 0x45, 0x96, 0x8b, 0xc8, 0x78, 0xb6, 0xf8, 0x06, 0xf3, 0xfe, 0x30, 0x18, 0xbe, 0x16,
0xba, 0x2f, 0xc5, 0x2c, 0xe3, 0xab, 0xb7, 0x25, 0xa9, 0x2b, 0x7c, 0x01, 0x9b, 0xc9, 0xe2, 0x14, 0xe5, 0x46, 0xef, 0xde, 0x17, 0x24, 0xef, 0xd0, 0x05, 0x3b, 0x2f, 0x42, 0x55, 0x84, 0xd5, 0x37,
0x1d, 0xac, 0xf6, 0xd7, 0x06, 0x9d, 0xd1, 0xdd, 0x46, 0xf0, 0xd2, 0x16, 0x7e, 0x83, 0x8e, 0x4f, 0x1d, 0xf3, 0x4d, 0x1b, 0xe1, 0x2b, 0x18, 0x45, 0xed, 0x3e, 0xca, 0xe9, 0xba, 0xdd, 0xa9, 0x3d,
0x01, 0x8a, 0xda, 0x91, 0x75, 0xdb, 0x19, 0xdd, 0xfe, 0x97, 0x5b, 0xbe, 0xc0, 0xc3, 0xfb, 0xe0, 0x7b, 0x58, 0x0e, 0xb3, 0xe1, 0x80, 0x6f, 0x09, 0xd1, 0x07, 0xc8, 0x9b, 0x69, 0x9d, 0x5d, 0xb3,
0x6b, 0x13, 0x2b, 0x63, 0xef, 0xb2, 0x6e, 0xef, 0xd2, 0x00, 0x18, 0xc0, 0x06, 0xe5, 0xf3, 0x9b, 0xa9, 0xd1, 0xe6, 0x0e, 0x78, 0x4b, 0x81, 0x4f, 0xc0, 0x52, 0x3a, 0x90, 0xda, 0x5c, 0xa8, 0x67,
0xb5, 0x6c, 0xaf, 0x2a, 0xa3, 0x6b, 0xd8, 0x72, 0xf2, 0x39, 0xe9, 0x42, 0xe6, 0x9a, 0xf0, 0x09, 0x2e, 0xb4, 0x06, 0xe8, 0xc0, 0x1e, 0xa5, 0xd5, 0xf5, 0xfa, 0x26, 0xb7, 0x0a, 0xbd, 0x14, 0xf6,
0x78, 0xee, 0xbf, 0xd0, 0x01, 0xb3, 0xda, 0xef, 0x34, 0xeb, 0x4f, 0xe2, 0xb3, 0xf2, 0xf5, 0xbc, 0xeb, 0xa1, 0x38, 0xa9, 0x3c, 0x4b, 0x15, 0xe1, 0x73, 0x18, 0xd4, 0x2f, 0x44, 0x39, 0xcc, 0xf8,
0xcb, 0x6b, 0xda, 0x0d, 0xcd, 0xab, 0xff, 0xa7, 0x39, 0xfa, 0xc2, 0x00, 0xaa, 0xe5, 0x47, 0x63, 0xdd, 0x2f, 0x1b, 0x5f, 0x06, 0xd7, 0xc5, 0xdb, 0x8a, 0xf3, 0x46, 0xb0, 0xe5, 0xb3, 0x73, 0x9f,
0x7c, 0x00, 0xa0, 0xe8, 0xb2, 0x24, 0x6d, 0x4e, 0x45, 0xea, 0xc2, 0xf6, 0x1d, 0x32, 0x49, 0xf1, 0x4f, 0xef, 0x3b, 0x03, 0x58, 0x35, 0x3c, 0x9f, 0xe3, 0x53, 0x00, 0x49, 0xb7, 0x05, 0x29, 0x7d,
0x31, 0xb4, 0x2e, 0x67, 0x09, 0xbb, 0xf1, 0xdb, 0xcd, 0xf8, 0xc5, 0xfc, 0xf9, 0x9c, 0x84, 0xcf, 0x25, 0xe2, 0x7a, 0xed, 0x56, 0x4d, 0x16, 0x31, 0x1e, 0x42, 0xef, 0xb6, 0xdc, 0x75, 0x5d, 0x78,
0xc0, 0x53, 0xce, 0x90, 0xcb, 0xf0, 0xde, 0x5f, 0x0f, 0x2a, 0xc7, 0xbc, 0xa6, 0xee, 0xf7, 0x7e, 0x5c, 0x16, 0x6e, 0xdf, 0x80, 0x57, 0x69, 0x7c, 0x01, 0x03, 0x59, 0xdb, 0x37, 0xef, 0xc5, 0x9e,
0x4c, 0x43, 0xf6, 0x73, 0x1a, 0xb2, 0x5f, 0xd3, 0x90, 0x7d, 0xfd, 0x1d, 0xae, 0xbc, 0x6f, 0xdb, 0x3d, 0x6a, 0x49, 0x57, 0x93, 0xf1, 0x46, 0x74, 0x32, 0xfe, 0xb5, 0x9c, 0xb0, 0xdf, 0xcb, 0x09,
0x57, 0xbb, 0x7f, 0x02, 0x00, 0x00, 0xff, 0xff, 0xb5, 0xbb, 0x7a, 0xf0, 0x4d, 0x03, 0x00, 0x00, 0xfb, 0xbb, 0x9c, 0xb0, 0x1f, 0xff, 0x26, 0x3b, 0x61, 0xdf, 0xfc, 0x0d, 0x47, 0xff, 0x03, 0x00,
0x00, 0xff, 0xff, 0xe1, 0x36, 0x3e, 0x73, 0x39, 0x03, 0x00, 0x00,
} }
func (m *Index) Marshal() (dAtA []byte, err error) { func (m *Index) Marshal() (dAtA []byte, err error) {
@ -598,13 +607,13 @@ func (m *HistoryQuery) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i -= 8 i -= 8
encoding_binary.LittleEndian.PutUint64(dAtA[i:], uint64(math.Float64bits(float64(m.EndTime)))) encoding_binary.LittleEndian.PutUint64(dAtA[i:], uint64(math.Float64bits(float64(m.EndTime))))
i-- i--
dAtA[i] = 0x29 dAtA[i] = 0x31
} }
if m.StartTime != 0 { if m.StartTime != 0 {
i -= 8 i -= 8
encoding_binary.LittleEndian.PutUint64(dAtA[i:], uint64(math.Float64bits(float64(m.StartTime)))) encoding_binary.LittleEndian.PutUint64(dAtA[i:], uint64(math.Float64bits(float64(m.StartTime))))
i-- i--
dAtA[i] = 0x21 dAtA[i] = 0x29
} }
if m.PagingInfo != nil { if m.PagingInfo != nil {
{ {
@ -616,7 +625,7 @@ func (m *HistoryQuery) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i = encodeVarintWakuStore(dAtA, i, uint64(size)) i = encodeVarintWakuStore(dAtA, i, uint64(size))
} }
i-- i--
dAtA[i] = 0x1a dAtA[i] = 0x22
} }
if len(m.ContentFilters) > 0 { if len(m.ContentFilters) > 0 {
for iNdEx := len(m.ContentFilters) - 1; iNdEx >= 0; iNdEx-- { for iNdEx := len(m.ContentFilters) - 1; iNdEx >= 0; iNdEx-- {
@ -629,9 +638,16 @@ func (m *HistoryQuery) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i = encodeVarintWakuStore(dAtA, i, uint64(size)) i = encodeVarintWakuStore(dAtA, i, uint64(size))
} }
i-- i--
dAtA[i] = 0x12 dAtA[i] = 0x1a
} }
} }
if len(m.PubsubTopic) > 0 {
i -= len(m.PubsubTopic)
copy(dAtA[i:], m.PubsubTopic)
i = encodeVarintWakuStore(dAtA, i, uint64(len(m.PubsubTopic)))
i--
dAtA[i] = 0x12
}
return len(dAtA) - i, nil return len(dAtA) - i, nil
} }
@ -820,6 +836,10 @@ func (m *HistoryQuery) Size() (n int) {
} }
var l int var l int
_ = l _ = l
l = len(m.PubsubTopic)
if l > 0 {
n += 1 + l + sovWakuStore(uint64(l))
}
if len(m.ContentFilters) > 0 { if len(m.ContentFilters) > 0 {
for _, e := range m.ContentFilters { for _, e := range m.ContentFilters {
l = e.Size() l = e.Size()
@ -1228,6 +1248,38 @@ func (m *HistoryQuery) Unmarshal(dAtA []byte) error {
} }
switch fieldNum { switch fieldNum {
case 2: case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field PubsubTopic", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowWakuStore
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthWakuStore
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthWakuStore
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.PubsubTopic = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 3:
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field ContentFilters", wireType) return fmt.Errorf("proto: wrong wireType = %d for field ContentFilters", wireType)
} }
@ -1261,7 +1313,7 @@ func (m *HistoryQuery) Unmarshal(dAtA []byte) error {
return err return err
} }
iNdEx = postIndex iNdEx = postIndex
case 3: case 4:
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field PagingInfo", wireType) return fmt.Errorf("proto: wrong wireType = %d for field PagingInfo", wireType)
} }
@ -1297,7 +1349,7 @@ func (m *HistoryQuery) Unmarshal(dAtA []byte) error {
return err return err
} }
iNdEx = postIndex iNdEx = postIndex
case 4: case 5:
if wireType != 1 { if wireType != 1 {
return fmt.Errorf("proto: wrong wireType = %d for field StartTime", wireType) return fmt.Errorf("proto: wrong wireType = %d for field StartTime", wireType)
} }
@ -1308,7 +1360,7 @@ func (m *HistoryQuery) Unmarshal(dAtA []byte) error {
v = uint64(encoding_binary.LittleEndian.Uint64(dAtA[iNdEx:])) v = uint64(encoding_binary.LittleEndian.Uint64(dAtA[iNdEx:]))
iNdEx += 8 iNdEx += 8
m.StartTime = float64(math.Float64frombits(v)) m.StartTime = float64(math.Float64frombits(v))
case 5: case 6:
if wireType != 1 { if wireType != 1 {
return fmt.Errorf("proto: wrong wireType = %d for field EndTime", wireType) return fmt.Errorf("proto: wrong wireType = %d for field EndTime", wireType)
} }

9
waku/v2/protocol/pb/waku_store.proto
View File

@ -24,10 +24,11 @@ message ContentFilter {
} }
message HistoryQuery { message HistoryQuery {
repeated ContentFilter contentFilters = 2; string pubsubTopic = 2;
PagingInfo pagingInfo = 3; // used for pagination repeated ContentFilter contentFilters = 3;
double startTime = 4; PagingInfo pagingInfo = 4; // used for pagination
double endTime = 5; double startTime = 5;
double endTime = 6;
} }
message HistoryResponse { message HistoryResponse {

40
waku/v2/protocol/store/waku_store.go
View File

@ -28,7 +28,7 @@ import (
var log = logging.Logger("wakustore") var log = logging.Logger("wakustore")
const WakuStoreProtocolId = libp2pProtocol.ID("/vac/waku/store/2.0.0-beta1") const WakuStoreProtocolId = libp2pProtocol.ID("/vac/waku/store/2.0.0-beta3")
const MaxPageSize = 100 // Maximum number of waku messages in each page const MaxPageSize = 100 // Maximum number of waku messages in each page
const ConnectionTimeout = 10 * time.Second const ConnectionTimeout = 10 * time.Second
const DefaultContentTopic = "/waku/2/default-content/proto" const DefaultContentTopic = "/waku/2/default-content/proto"
@ -148,12 +148,31 @@ func (w *WakuStore) FindMessages(query *pb.HistoryQuery) *pb.HistoryResponse {
} }
} }
// filter based on content filters
// an empty list of contentFilters means no content filter is requested
if len(query.ContentFilters) != 0 {
match := false
for _, cf := range query.ContentFilters { for _, cf := range query.ContentFilters {
if cf.ContentTopic == indexedMsg.msg.ContentTopic { if cf.ContentTopic == indexedMsg.msg.ContentTopic {
data = append(data, indexedMsg) match = true
break
}
}
if !match {
continue
} }
} }
// filter based on pubsub topic
// an empty pubsub topic means no pubsub topic filter is requested
if query.PubsubTopic != "" {
if indexedMsg.pubsubTopic != query.PubsubTopic {
continue
}
}
// Some criteria matched
data = append(data, indexedMsg)
} }
result.Messages, result.PagingInfo = paginateWithoutIndex(data, query.PagingInfo) result.Messages, result.PagingInfo = paginateWithoutIndex(data, query.PagingInfo)
@ -161,14 +180,15 @@ func (w *WakuStore) FindMessages(query *pb.HistoryQuery) *pb.HistoryResponse {
} }
type MessageProvider interface { type MessageProvider interface {
GetAll() ([]*pb.WakuMessage, error) GetAll() ([]*protocol.Envelope, error)
Put(cursor *pb.Index, message *pb.WakuMessage) error Put(cursor *pb.Index, pubsubTopic string, message *pb.WakuMessage) error
Stop() Stop()
} }
type IndexedWakuMessage struct { type IndexedWakuMessage struct {
msg *pb.WakuMessage msg *pb.WakuMessage
index *pb.Index index *pb.Index
pubsubTopic string
} }
type WakuStore struct { type WakuStore struct {
@ -211,19 +231,19 @@ func (store *WakuStore) Start(h host.Host) {
return return
} }
messages, err := store.msgProvider.GetAll() envelopes, err := store.msgProvider.GetAll()
if err != nil { if err != nil {
log.Error("could not load DBProvider messages") log.Error("could not load DBProvider messages")
return return
} }
for _, msg := range messages { for _, env := range envelopes {
idx, err := computeIndex(msg) idx, err := computeIndex(env.Message())
if err != nil { if err != nil {
log.Error("could not calculate message index", err) log.Error("could not calculate message index", err)
continue continue
} }
store.messages = append(store.messages, IndexedWakuMessage{msg: msg, index: idx}) store.messages = append(store.messages, IndexedWakuMessage{msg: env.Message(), index: idx, pubsubTopic: env.PubsubTopic()})
} }
log.Info("Store protocol started") log.Info("Store protocol started")
@ -238,14 +258,14 @@ func (store *WakuStore) storeIncomingMessages() {
} }
store.messagesMutex.Lock() store.messagesMutex.Lock()
store.messages = append(store.messages, IndexedWakuMessage{msg: envelope.Message(), index: index}) store.messages = append(store.messages, IndexedWakuMessage{msg: envelope.Message(), index: index, pubsubTopic: envelope.PubsubTopic()})
store.messagesMutex.Unlock() store.messagesMutex.Unlock()
if store.msgProvider == nil { if store.msgProvider == nil {
continue continue
} }
err = store.msgProvider.Put(index, envelope.Message()) // Should the index be stored? err = store.msgProvider.Put(index, envelope.PubsubTopic(), envelope.Message()) // Should the index be stored?
if err != nil { if err != nil {
log.Error("could not store message", err) log.Error("could not store message", err)
continue continue