diff --git a/rpc/api.go b/rpc/api.go
index 614e08764041136d7eb6b9b87f025825e64f096e..a73188f07adc4b7a3ff8dd750e4fe352e0e52974 100644
--- a/rpc/api.go
+++ b/rpc/api.go
@@ -467,7 +467,7 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err
if err := json.Unmarshal(req.Params, &args); err != nil {
return err
}
- *reply = api.xeth().Whisper().Messages(args.Id)
+ *reply = api.xeth().Messages(args.Id)
// case "eth_register":
// // Placeholder for actual type
diff --git a/whisper/envelope.go b/whisper/envelope.go
index ba3e4ccc940c92250b0623da768b5e28e97b6418..c1d84df78b026dda74eb9670c5055898eec092d2 100644
--- a/whisper/envelope.go
+++ b/whisper/envelope.go
@@ -73,6 +73,7 @@ func (self *Envelope) Open(key *ecdsa.PrivateKey) (msg *Message, err error) {
message := &Message{
Flags: data[0],
Sent: int64(self.Expiry - self.TTL),
+ Hash: self.Hash(),
}
data = data[1:]
diff --git a/whisper/envelope_test.go b/whisper/envelope_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..ed1f08365187ef4be0cae0e52a39c865ab92da29
--- /dev/null
+++ b/whisper/envelope_test.go
@@ -0,0 +1,36 @@
+package whisper
+
+import (
+ "bytes"
+ "testing"
+)
+
+func TestEnvelopeOpen(t *testing.T) {
+ payload := []byte("hello world")
+ message := NewMessage(payload)
+
+ envelope, err := message.Wrap(DefaultPoW, Options{})
+ if err != nil {
+ t.Fatalf("failed to wrap message: %v", err)
+ }
+ opened, err := envelope.Open(nil)
+ if err != nil {
+ t.Fatalf("failed to open envelope: %v.", err)
+ }
+ if opened.Flags != message.Flags {
+ t.Fatalf("flags mismatch: have %d, want %d", opened.Flags, message.Flags)
+ }
+ if bytes.Compare(opened.Signature, message.Signature) != 0 {
+ t.Fatalf("signature mismatch: have 0x%x, want 0x%x", opened.Signature, message.Signature)
+ }
+ if bytes.Compare(opened.Payload, message.Payload) != 0 {
+ t.Fatalf("payload mismatch: have 0x%x, want 0x%x", opened.Payload, message.Payload)
+ }
+ if opened.Sent != message.Sent {
+ t.Fatalf("send time mismatch: have %d, want %d", opened.Sent, message.Sent)
+ }
+
+ if opened.Hash != envelope.Hash() {
+ t.Fatalf("message hash mismatch: have 0x%x, want 0x%x", opened.Hash, envelope.Hash())
+ }
+}
diff --git a/whisper/message.go b/whisper/message.go
index 07c67356773295a656ecd6adbfc9b280557d2b2d..69d85b8948de72064714f233fb6e376bc10d9cb0 100644
--- a/whisper/message.go
+++ b/whisper/message.go
@@ -8,12 +8,13 @@ import (
"math/rand"
"time"
+ "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
)
-// Message represents an end-user data packet to trasmit through the Whisper
+// Message represents an end-user data packet to transmit through the Whisper
// protocol. These are wrapped into Envelopes that need not be understood by
// intermediate nodes, just forwarded.
type Message struct {
@@ -22,7 +23,8 @@ type Message struct {
Payload []byte
Sent int64
- To *ecdsa.PublicKey
+ To *ecdsa.PublicKey // Message recipient (identity used to decode the message)
+ Hash common.Hash // Message envelope hash to act as a unique id in de-duplication
}
// Options specifies the exact way a message should be wrapped into an Envelope.
diff --git a/xeth/whisper_filter.go b/xeth/whisper_filter.go
index 9d8a739b7602039446b5eab60213c57025251aef..52e70e041df3e7fe176aaeb5be2f01f06d354ccd 100644
--- a/xeth/whisper_filter.go
+++ b/xeth/whisper_filter.go
@@ -1,26 +1,84 @@
// Contains the external API side message filter for watching, pooling and polling
-// matched whisper messages.
+// matched whisper messages, also serializing data access to avoid duplications.
package xeth
-import "time"
+import (
+ "sync"
+ "time"
+
+ "github.com/ethereum/go-ethereum/common"
+)
// whisperFilter is the message cache matching a specific filter, accumulating
// inbound messages until the are requested by the client.
type whisperFilter struct {
- id int // Filter identifier
- cache []WhisperMessage // Cache of messages not yet polled
- timeout time.Time // Time when the last message batch was queries
+ id int // Filter identifier for old message retrieval
+ ref *Whisper // Whisper reference for old message retrieval
+
+ cache []WhisperMessage // Cache of messages not yet polled
+ skip map[common.Hash]struct{} // List of retrieved messages to avoid duplication
+ update time.Time // Time of the last message query
+
+ lock sync.RWMutex // Lock protecting the filter internals
+}
+
+// newWhisperFilter creates a new serialized, poll based whisper topic filter.
+func newWhisperFilter(id int, ref *Whisper) *whisperFilter {
+ return &whisperFilter{
+ id: id,
+ ref: ref,
+
+ update: time.Now(),
+ skip: make(map[common.Hash]struct{}),
+ }
+}
+
+// messages retrieves all the cached messages from the entire pool matching the
+// filter, resetting the filter's change buffer.
+func (w *whisperFilter) messages() []WhisperMessage {
+ w.lock.Lock()
+ defer w.lock.Unlock()
+
+ w.cache = nil
+ w.update = time.Now()
+
+ w.skip = make(map[common.Hash]struct{})
+ messages := w.ref.Messages(w.id)
+ for _, message := range messages {
+ w.skip[message.ref.Hash] = struct{}{}
+ }
+ return messages
}
// insert injects a new batch of messages into the filter cache.
-func (w *whisperFilter) insert(msgs ...WhisperMessage) {
- w.cache = append(w.cache, msgs...)
+func (w *whisperFilter) insert(messages ...WhisperMessage) {
+ w.lock.Lock()
+ defer w.lock.Unlock()
+
+ for _, message := range messages {
+ if _, ok := w.skip[message.ref.Hash]; !ok {
+ w.cache = append(w.cache, messages...)
+ }
+ }
}
// retrieve fetches all the cached messages from the filter.
func (w *whisperFilter) retrieve() (messages []WhisperMessage) {
+ w.lock.Lock()
+ defer w.lock.Unlock()
+
messages, w.cache = w.cache, nil
- w.timeout = time.Now()
+ w.update = time.Now()
+
return
}
+
+// activity returns the last time instance when client requests were executed on
+// the filter.
+func (w *whisperFilter) activity() time.Time {
+ w.lock.RLock()
+ defer w.lock.RUnlock()
+
+ return w.update
+}
diff --git a/xeth/xeth.go b/xeth/xeth.go
index e7e553036f77d643f7a7d5fcbac4008187d759d7..8cc32c95869aa43c7cbc7c58070f79d815b8e69e 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
}