roysc/plugeth
1
0
Fork 0
forked from cerc-io/plugeth
Code Pull Requests Activity

rpc, whisper, xeth: polish whisper RPC interface

Browse Source
This commit is contained in:
Péter Szilágyi 2015-04-17 16:45:44 +03:00
parent 182d484aa7
commit 3563c59b12
6 changed files with 132 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
rpc/api.go
View File

@ -408,18 +408,18 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err
*reply = newHexData(res)
case "shh_version":
*reply = api.xeth().WhisperVersion()
case "shh_post":
args := new(WhisperMessageArgs)
if err := json.Unmarshal(req.Params, &args); err != nil {
return err
}
err := api.xeth().Whisper().Post(args.Payload, args.To, args.From, args.Topics, args.Priority, args.Ttl)
if err != nil {
return err
}
*reply = true
case "shh_newIdentity":
*reply = api.xeth().Whisper().NewIdentity()
// case "shh_removeIdentity":
@ -434,32 +434,35 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err
return err
}
*reply = api.xeth().Whisper().HasIdentity(args.Identity)
case "shh_newGroup", "shh_addToGroup":
return NewNotImplementedError(req.Method)
case "shh_newFilter":
args := new(WhisperFilterArgs)
if err := json.Unmarshal(req.Params, &args); err != nil {
return err
}
opts := new(xeth.Options)
// opts.From = args.From
opts.To = args.To
opts.Topics = args.Topics
id := api.xeth().NewWhisperFilter(opts)
id := api.xeth().NewWhisperFilter(args.To, args.From, args.Topics)
*reply = newHexNum(big.NewInt(int64(id)).Bytes())
case "shh_uninstallFilter":
args := new(FilterIdArgs)
if err := json.Unmarshal(req.Params, &args); err != nil {
return err
}
*reply = api.xeth().UninstallWhisperFilter(args.Id)
case "shh_getFilterChanges":
// Retrieve all the new messages arrived since the last request
args := new(FilterIdArgs)
if err := json.Unmarshal(req.Params, &args); err != nil {
return err
}
*reply = api.xeth().MessagesChanged(args.Id)
case "shh_getMessages":
// Retrieve all the cached messages matching a specific, existing filter
args := new(FilterIdArgs)
if err := json.Unmarshal(req.Params, &args); err != nil {
return err

13
whisper/whisper.go
View File

@ -58,6 +58,8 @@ type Whisper struct {
quit chan struct{}
}
// New creates a Whisper client ready to communicate through the Ethereum P2P
// network.
func New() *Whisper {
whisper := &Whisper{
filters: filter.New(),
@ -148,7 +150,7 @@ func (self *Whisper) Stop() {
glog.V(logger.Info).Infoln("Whisper stopped")
}
// Messages retrieves the currently pooled messages matching a filter id.
// Messages retrieves all the currently pooled messages matching a filter id.
func (self *Whisper) Messages(id int) []*Message {
messages := make([]*Message, 0)
if filter := self.filters.Get(id); filter != nil {
@ -163,15 +165,6 @@ func (self *Whisper) Messages(id int) []*Message {
return messages
}
// func (self *Whisper) RemoveIdentity(key *ecdsa.PublicKey) bool {
// k := string(crypto.FromECDSAPub(key))
// if _, ok := self.keys[k]; ok {
// delete(self.keys, k)
// return true
// }
// return false
// }
// handlePeer is called by the underlying P2P layer when the whisper sub-protocol
// connection is negotiated.
func (self *Whisper) handlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error {

141
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,78 @@ 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 {
// 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.NewTopicsFromStrings(topics...),
}
var i int
filter.Fn = func(msg *whisper.Message) {
opts.Fn(NewWhisperMessage(msg))
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)
}
}
i = self.Whisper.Watch(filter)
return i
// 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 {
filter := whisper.Filter{
To: crypto.ToECDSAPub(common.FromHex(to)),
From: crypto.ToECDSAPub(common.FromHex(from)),
Topics: whisper.NewTopicsFromStrings(topics...),
}
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
}

26
xeth/whisper_filter.go Normal file
View File

@ -0,0 +1,26 @@
// Contains the external API side message filter for watching, pooling and polling
// matched whisper messages.
package xeth
import "time"
// 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
}
// insert injects a new batch of messages into the filter cache.
func (w *whisperFilter) insert(msgs ...WhisperMessage) {
w.cache = append(w.cache, msgs...)
}
// retrieve fetches all the cached messages from the filter.
func (w *whisperFilter) retrieve() (messages []WhisperMessage) {
messages, w.cache = w.cache, nil
w.timeout = time.Now()
return
}

31
xeth/whisper_message.go Normal file
View File

@ -0,0 +1,31 @@
// Contains the external API representation of a whisper message.
package xeth
import (
"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"`
}
// 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,
}
}

27
xeth/xeth.go
View File

@ -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