Added whisper interface for xeth, added examples, updated RPC

* Added RPC methods for whisper
* Added whisper example

3 files changed, 124 insertions, 9 deletions

diff --git a/rpc/args.go b/rpc/args.go
index aaa017c4e..75eef873d 100644
--- a/rpc/args.go
+++ b/rpc/args.go
@@ -251,3 +251,12 @@ func (a *DbArgs) requirements() error {
 	}
 	return nil
 }
+
+type WhisperMessageArgs struct {
+	Payload  string
+	To       string
+	From     string
+	Topics   []string
+	Priority uint32
+	Ttl      uint32
+}
diff --git a/rpc/message.go b/rpc/message.go
index 5045adb8f..919302921 100644
--- a/rpc/message.go
+++ b/rpc/message.go
@@ -21,6 +21,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+
+	"github.com/ethereum/go-ethereum/xeth"
 )
 
 const (
@@ -270,3 +272,45 @@ func (req *RpcRequest) ToDbGetArgs() (*DbArgs, error) {
 	rpclogger.DebugDetailf("%T %v", args, args)
 	return &args, nil
 }
+
+func (req *RpcRequest) ToWhisperFilterArgs() (*xeth.Options, error) {
+	if len(req.Params) < 1 {
+		return nil, NewErrorResponse(ErrorArguments)
+	}
+
+	var args xeth.Options
+	err := json.Unmarshal(req.Params[0], &args)
+	if err != nil {
+		return nil, NewErrorResponseWithError(ErrorDecodeArgs, err)
+	}
+	rpclogger.DebugDetailf("%T %v", args, args)
+	return &args, nil
+}
+
+func (req *RpcRequest) ToWhisperChangedArgs() (int, error) {
+	if len(req.Params) < 1 {
+		return 0, NewErrorResponse(ErrorArguments)
+	}
+
+	var id int
+	err := json.Unmarshal(req.Params[0], &id)
+	if err != nil {
+		return 0, NewErrorResponse(ErrorDecodeArgs)
+	}
+	rpclogger.DebugDetailf("%T %v", id, id)
+	return id, nil
+}
+
+func (req *RpcRequest) ToWhisperPostArgs() (*WhisperMessageArgs, error) {
+	if len(req.Params) < 1 {
+		return nil, NewErrorResponse(ErrorArguments)
+	}
+
+	var args WhisperMessageArgs
+	err := json.Unmarshal(req.Params[0], &args)
+	if err != nil {
+		return nil, err
+	}
+	rpclogger.DebugDetailf("%T %v", args, args)
+	return &args, nil
+}
diff --git a/rpc/packages.go b/rpc/packages.go
index aa51aad42..8344d6a46 100644
--- a/rpc/packages.go
+++ b/rpc/packages.go
@@ -44,18 +44,22 @@ type EthereumApi struct {
 	xeth          *xeth.XEth
 	filterManager *filter.FilterManager
 
-	mut  sync.RWMutex
-	logs map[int]state.Logs
+	logMut sync.RWMutex
+	logs   map[int]state.Logs
+
+	messagesMut sync.RWMutex
+	messages    map[int][]xeth.WhisperMessage
 
 	db ethutil.Database
 }
 
-func NewEthereumApi(xeth *xeth.XEth) *EthereumApi {
+func NewEthereumApi(eth *xeth.XEth) *EthereumApi {
 	db, _ := ethdb.NewLDBDatabase("dapps")
 	api := &EthereumApi{
-		xeth:          xeth,
-		filterManager: filter.NewFilterManager(xeth.Backend().EventMux()),
+		xeth:          eth,
+		filterManager: filter.NewFilterManager(eth.Backend().EventMux()),
 		logs:          make(map[int]state.Logs),
+		messages:      make(map[int][]xeth.WhisperMessage),
 		db:            db,
 	}
 	go api.filterManager.Start()
@@ -67,8 +71,8 @@ func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) erro
 	var id int
 	filter := core.NewFilter(self.xeth.Backend())
 	filter.LogsCallback = func(logs state.Logs) {
-		self.mut.Lock()
-		defer self.mut.Unlock()
+		self.logMut.Lock()
+		defer self.logMut.Unlock()
 
 		self.logs[id] = append(self.logs[id], logs...)
 	}
@@ -79,8 +83,8 @@ func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) erro
 }
 
 func (self *EthereumApi) FilterChanged(id int, reply *interface{}) error {
-	self.mut.RLock()
-	defer self.mut.RUnlock()
+	self.logMut.RLock()
+	defer self.logMut.RUnlock()
 
 	*reply = toLogs(self.logs[id])
 
@@ -257,6 +261,44 @@ func (p *EthereumApi) DbGet(args *DbArgs, reply *interface{}) error {
 	return nil
 }
 
+func (p *EthereumApi) NewWhisperIdentity(reply *interface{}) error {
+	*reply = p.xeth.Whisper().NewIdentity()
+	return nil
+}
+
+func (p *EthereumApi) NewWhisperFilter(args *xeth.Options, reply *interface{}) error {
+	var id int
+	args.Fn = func(msg xeth.WhisperMessage) {
+		p.messagesMut.Lock()
+		defer p.messagesMut.Unlock()
+		p.messages[id] = append(p.messages[id], msg)
+	}
+	id = p.xeth.Whisper().Watch(args)
+	*reply = id
+	return nil
+}
+
+func (self *EthereumApi) MessagesChanged(id int, reply *interface{}) error {
+	self.messagesMut.RLock()
+	defer self.messagesMut.RUnlock()
+
+	*reply = self.messages[id]
+
+	self.messages[id] = nil // empty the messages
+
+	return nil
+}
+
+func (p *EthereumApi) WhisperPost(args *WhisperMessageArgs, reply *interface{}) error {
+	err := p.xeth.Whisper().Post(args.Payload, args.To, args.From, args.Topics, args.Priority, args.Ttl)
+	if err != nil {
+		return err
+	}
+
+	*reply = true
+	return nil
+}
+
 func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error {
 	// Spec at https://github.com/ethereum/wiki/wiki/Generic-ON-RPC
 	rpclogger.DebugDetailf("%T %s", req.Params, req.Params)
@@ -354,6 +396,26 @@ func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error
 			return err
 		}
 		return p.DbGet(args, reply)
+	case "shh_newIdentity":
+		return p.NewWhisperIdentity(reply)
+	case "shh_newFilter":
+		args, err := req.ToWhisperFilterArgs()
+		if err != nil {
+			return err
+		}
+		return p.NewWhisperFilter(args, reply)
+	case "shh_changed":
+		args, err := req.ToWhisperChangedArgs()
+		if err != nil {
+			return err
+		}
+		return p.MessagesChanged(args, reply)
+	case "shh_post":
+		args, err := req.ToWhisperPostArgs()
+		if err != nil {
+			return nil
+		}
+		return p.WhisperPost(args, reply)
 	default:
 		return NewErrorResponse(fmt.Sprintf("%v %s", ErrorNotImplemented, req.Method))
 	}