gaspersic/go-ethereum
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge branch 'develop' of github.com-obscure:ethereum/go-ethereum into develop

Browse Source 
Conflicts:
	rpc/api.go
This commit is contained in:
obscuren
2015-04-28 11:17:41 +02:00
parent bac455c011 a05c420371
commit 99027c79fe
20 changed files with 1028 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
xeth/whisper.go
View File

@@ -1,7 +1,9 @@
// Contains the external API to the whisper sub-protocol.
package xeth
import (
"errors"
"fmt"
"time"
"github.com/ethereum/go-ethereum/common"
@@ -12,109 +14,92 @@ import (
var qlogger = logger.NewLogger("XSHH")
// Whisper represents the API wrapper around the internal whisper implementation.
type Whisper struct {
*whisper.Whisper
}
// NewWhisper wraps an internal whisper client into an external API version.
func NewWhisper(w *whisper.Whisper) *Whisper {
return &Whisper{w}
}
func (self *Whisper) Post(payload string, to, from string, topics []string, priority, ttl uint32) error {
if priority == 0 {
priority = 1000
}
if ttl == 0 {
ttl = 100
}
pk := crypto.ToECDSAPub(common.FromHex(from))
if key := self.Whisper.GetIdentity(pk); key != nil || len(from) == 0 {
msg := whisper.NewMessage(common.FromHex(payload))
envelope, err := msg.Wrap(time.Duration(priority*100000), whisper.Options{
TTL: time.Duration(ttl) * time.Second,
To: crypto.ToECDSAPub(common.FromHex(to)),
From: key,
Topics: whisper.NewTopicsFromStrings(topics...),
})
if err != nil {
return err
}
if err := self.Whisper.Send(envelope); err != nil {
return err
}
} else {
return errors.New("unmatched pub / priv for seal")
}
return nil
}
// NewIdentity generates a new cryptographic identity for the client, and injects
// it into the known identities for message decryption.
func (self *Whisper) NewIdentity() string {
key := self.Whisper.NewIdentity()
return common.ToHex(crypto.FromECDSAPub(&key.PublicKey))
identity := self.Whisper.NewIdentity()
return common.ToHex(crypto.FromECDSAPub(&identity.PublicKey))
}
// HasIdentity checks if the the whisper node is configured with the private key
// of the specified public pair.
func (self *Whisper) HasIdentity(key string) bool {
return self.Whisper.HasIdentity(crypto.ToECDSAPub(common.FromHex(key)))
}
// func (self *Whisper) RemoveIdentity(key string) bool {
// return self.Whisper.RemoveIdentity(crypto.ToECDSAPub(common.FromHex(key)))
// }
func (self *Whisper) Watch(opts *Options) int {
filter := whisper.Filter{
To: crypto.ToECDSAPub(common.FromHex(opts.To)),
From: crypto.ToECDSAPub(common.FromHex(opts.From)),
Topics: whisper.NewTopicsFromStrings(opts.Topics...),
// Post injects a message into the whisper network for distribution.
func (self *Whisper) Post(payload string, to, from string, topics []string, priority, ttl uint32) error {
// Decode the topic strings
topicsDecoded := make([][]byte, len(topics))
for i, topic := range topics {
topicsDecoded[i] = common.FromHex(topic)
}
var i int
filter.Fn = func(msg *whisper.Message) {
opts.Fn(NewWhisperMessage(msg))
// Construct the whisper message and transmission options
message := whisper.NewMessage(common.FromHex(payload))
options := whisper.Options{
To: crypto.ToECDSAPub(common.FromHex(to)),
TTL: time.Duration(ttl) * time.Second,
Topics: whisper.NewTopics(topicsDecoded...),
}
i = self.Whisper.Watch(filter)
return i
if len(from) != 0 {
if key := self.Whisper.GetIdentity(crypto.ToECDSAPub(common.FromHex(from))); key != nil {
options.From = key
} else {
return fmt.Errorf("unknown identity to send from: %s", from)
}
}
// Wrap and send the message
pow := time.Duration(priority) * time.Millisecond
envelope, err := message.Wrap(pow, options)
if err != nil {
return err
}
if err := self.Whisper.Send(envelope); err != nil {
return err
}
return nil
}
func (self *Whisper) Messages(id int) (messages []WhisperMessage) {
msgs := self.Whisper.Messages(id)
messages = make([]WhisperMessage, len(msgs))
for i, message := range msgs {
// Watch installs a new message handler to run in case a matching packet arrives
// from the whisper network.
func (self *Whisper) Watch(to, from string, topics [][]string, fn func(WhisperMessage)) int {
// Decode the topic strings
topicsDecoded := make([][][]byte, len(topics))
for i, condition := range topics {
topicsDecoded[i] = make([][]byte, len(condition))
for j, topic := range condition {
topicsDecoded[i][j] = common.FromHex(topic)
}
}
// Assemble and inject the filter into the whisper client
filter := whisper.Filter{
To: crypto.ToECDSAPub(common.FromHex(to)),
From: crypto.ToECDSAPub(common.FromHex(from)),
Topics: whisper.NewFilterTopics(topicsDecoded...),
}
filter.Fn = func(message *whisper.Message) {
fn(NewWhisperMessage(message))
}
return self.Whisper.Watch(filter)
}
// Messages retrieves all the currently pooled messages matching a filter id.
func (self *Whisper) Messages(id int) []WhisperMessage {
pool := self.Whisper.Messages(id)
messages := make([]WhisperMessage, len(pool))
for i, message := range pool {
messages[i] = NewWhisperMessage(message)
}
return
}
type Options struct {
To string
From string
Topics []string
Fn func(msg WhisperMessage)
}
type WhisperMessage struct {
ref *whisper.Message
Payload string `json:"payload"`
To string `json:"to"`
From string `json:"from"`
Sent int64 `json:"sent"`
}
func NewWhisperMessage(msg *whisper.Message) WhisperMessage {
return WhisperMessage{
ref: msg,
Payload: common.ToHex(msg.Payload),
From: common.ToHex(crypto.FromECDSAPub(msg.Recover())),
To: common.ToHex(crypto.FromECDSAPub(msg.To)),
Sent: msg.Sent,
}
return messages
}

84
xeth/whisper_filter.go Normal file
View File

@@ -0,0 +1,84 @@
// Contains the external API side message filter for watching, pooling and polling
// matched whisper messages, also serializing data access to avoid duplications.
package xeth
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 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(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.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
}

37
xeth/whisper_message.go Normal file
View File

@@ -0,0 +1,37 @@
// Contains the external API representation of a whisper message.
package xeth
import (
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/whisper"
)
// WhisperMessage is the external API representation of a whisper.Message.
type WhisperMessage struct {
ref *whisper.Message
Payload string `json:"payload"`
To string `json:"to"`
From string `json:"from"`
Sent int64 `json:"sent"`
TTL int64 `json:"ttl"`
Hash string `json:"hash"`
}
// NewWhisperMessage converts an internal message into an API version.
func NewWhisperMessage(message *whisper.Message) WhisperMessage {
return WhisperMessage{
ref: message,
Payload: common.ToHex(message.Payload),
From: common.ToHex(crypto.FromECDSAPub(message.Recover())),
To: common.ToHex(crypto.FromECDSAPub(message.To)),
Sent: message.Sent.Unix(),
TTL: int64(message.TTL / time.Second),
Hash: common.ToHex(message.Hash.Bytes()),
}
}

68
xeth/xeth.go
View File

@@ -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)
}
@@ -456,35 +456,61 @@ func (self *XEth) AllLogs(earliest, latest int64, skip, max int, address []strin
return filter.Find()
}
func (p *XEth) NewWhisperFilter(opts *Options) int {
// 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
opts.Fn = func(msg WhisperMessage) {
p.messagesMut.Lock()
defer p.messagesMut.Unlock()
p.messages[id].add(msg) // = append(p.messages[id], msg)
// 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)
}
id = p.Whisper().Watch(opts)
p.messages[id] = &whisperFilter{timeout: time.Now()}
// 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].get()
return self.messages[id].messages()
}
return nil
}
// 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].retrieve()
}
return nil
}
@@ -735,22 +761,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