From 3563c59b12b0b8b5fd15847bf97d71dfd8416207 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Fri, 17 Apr 2015 16:45:44 +0300 Subject: rpc, whisper, xeth: polish whisper RPC interface --- xeth/xeth.go | 27 ++++++--------------------- 1 file changed, 6 insertions(+), 21 deletions(-) (limited to 'xeth/xeth.go') diff --git a/xeth/xeth.go b/xeth/xeth.go index 693acb910..e7e553036 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -452,14 +452,15 @@ func (self *XEth) AllLogs(earliest, latest int64, skip, max int, address []strin return filter.Find() } -func (p *XEth) NewWhisperFilter(opts *Options) int { +func (p *XEth) NewWhisperFilter(to, from string, topics []string) int { var id int - opts.Fn = func(msg WhisperMessage) { + callback := func(msg WhisperMessage) { p.messagesMut.Lock() defer p.messagesMut.Unlock() - p.messages[id].add(msg) // = append(p.messages[id], msg) + + p.messages[id].insert(msg) } - id = p.Whisper().Watch(opts) + id = p.Whisper().Watch(to, from, topics, callback) p.messages[id] = &whisperFilter{timeout: time.Now()} return id } @@ -478,7 +479,7 @@ func (self *XEth) MessagesChanged(id int) []WhisperMessage { defer self.messagesMut.Unlock() if self.messages[id] != nil { - return self.messages[id].get() + return self.messages[id].retrieve() } return nil @@ -731,22 +732,6 @@ func (m callmsg) Gas() *big.Int { return m.gas } func (m callmsg) Value() *big.Int { return m.value } func (m callmsg) Data() []byte { return m.data } -type whisperFilter struct { - messages []WhisperMessage - timeout time.Time - id int -} - -func (w *whisperFilter) add(msgs ...WhisperMessage) { - w.messages = append(w.messages, msgs...) -} -func (w *whisperFilter) get() []WhisperMessage { - w.timeout = time.Now() - tmp := w.messages - w.messages = nil - return tmp -} - type logFilter struct { logs state.Logs timeout time.Time -- cgit From 7948cc0029db76557d6540341bdfeb818ce32c65 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Mon, 20 Apr 2015 14:56:38 +0300 Subject: rpc, whisper, xeth: fix RPC message retrieval data race --- xeth/xeth.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) (limited to 'xeth/xeth.go') diff --git a/xeth/xeth.go b/xeth/xeth.go index e7e553036..8cc32c958 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -97,7 +97,7 @@ done: } for id, filter := range self.messages { - if time.Since(filter.timeout) > filterTickerTime { + if time.Since(filter.activity()) > filterTickerTime { self.Whisper().Unwatch(id) delete(self.messages, id) } @@ -461,7 +461,7 @@ func (p *XEth) NewWhisperFilter(to, from string, topics []string) int { p.messages[id].insert(msg) } id = p.Whisper().Watch(to, from, topics, callback) - p.messages[id] = &whisperFilter{timeout: time.Now()} + p.messages[id] = newWhisperFilter(id, p.Whisper()) return id } @@ -481,7 +481,16 @@ func (self *XEth) MessagesChanged(id int) []WhisperMessage { if self.messages[id] != nil { return self.messages[id].retrieve() } + return nil +} + +func (self *XEth) Messages(id int) []WhisperMessage { + self.messagesMut.Lock() + defer self.messagesMut.Unlock() + if self.messages[id] != nil { + return self.messages[id].messages() + } return nil } -- cgit From ae4bfc3cfb3f1debad9dd0211950ce09038ffa90 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Tue, 21 Apr 2015 18:31:08 +0300 Subject: rpc, ui/qt/qwhisper, whisper, xeth: introduce complex topic filters --- xeth/xeth.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'xeth/xeth.go') diff --git a/xeth/xeth.go b/xeth/xeth.go index 8cc32c958..ea6ae9950 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -452,7 +452,7 @@ func (self *XEth) AllLogs(earliest, latest int64, skip, max int, address []strin return filter.Find() } -func (p *XEth) NewWhisperFilter(to, from string, topics []string) int { +func (p *XEth) NewWhisperFilter(to, from string, topics [][]string) int { var id int callback := func(msg WhisperMessage) { p.messagesMut.Lock() -- cgit From 978ffd3097242a5faeb7b23b9c72590170004dc6 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Wed, 22 Apr 2015 18:35:50 +0300 Subject: rpc, xeth: finish cleaning up xeth --- xeth/xeth.go | 40 ++++++++++++++++++++++++++++------------ 1 file changed, 28 insertions(+), 12 deletions(-) (limited to 'xeth/xeth.go') diff --git a/xeth/xeth.go b/xeth/xeth.go index ea6ae9950..e4040d9d8 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -452,44 +452,60 @@ func (self *XEth) AllLogs(earliest, latest int64, skip, max int, address []strin return filter.Find() } +// NewWhisperFilter creates and registers a new message filter to watch for +// inbound whisper messages. All parameters at this point are assumed to be +// HEX encoded. func (p *XEth) NewWhisperFilter(to, from string, topics [][]string) int { + // Pre-define the id to be filled later var id int - callback := func(msg WhisperMessage) { - p.messagesMut.Lock() - defer p.messagesMut.Unlock() + // Callback to delegate core whisper messages to this xeth filter + callback := func(msg WhisperMessage) { + p.messagesMut.RLock() // Only read lock to the filter pool + defer p.messagesMut.RUnlock() p.messages[id].insert(msg) } + // Initialize the core whisper filter and wrap into xeth id = p.Whisper().Watch(to, from, topics, callback) + + p.messagesMut.Lock() p.messages[id] = newWhisperFilter(id, p.Whisper()) + p.messagesMut.Unlock() + return id } +// UninstallWhisperFilter disables and removes an existing filter. func (p *XEth) UninstallWhisperFilter(id int) bool { + p.messagesMut.Lock() + defer p.messagesMut.Unlock() + if _, ok := p.messages[id]; ok { delete(p.messages, id) return true } - return false } -func (self *XEth) MessagesChanged(id int) []WhisperMessage { - self.messagesMut.Lock() - defer self.messagesMut.Unlock() +// WhisperMessages retrieves all the known messages that match a specific filter. +func (self *XEth) WhisperMessages(id int) []WhisperMessage { + self.messagesMut.RLock() + defer self.messagesMut.RUnlock() if self.messages[id] != nil { - return self.messages[id].retrieve() + return self.messages[id].messages() } return nil } -func (self *XEth) Messages(id int) []WhisperMessage { - self.messagesMut.Lock() - defer self.messagesMut.Unlock() +// WhisperMessagesChanged retrieves all the new messages matched by a filter +// since the last retrieval +func (self *XEth) WhisperMessagesChanged(id int) []WhisperMessage { + self.messagesMut.RLock() + defer self.messagesMut.RUnlock() if self.messages[id] != nil { - return self.messages[id].messages() + return self.messages[id].retrieve() } return nil } -- cgit