aboutsummaryrefslogtreecommitdiffstats
path: root/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'rpc')
-rw-r--r--rpc/api.go203
-rw-r--r--rpc/api_test.go64
-rw-r--r--rpc/args.go17
-rw-r--r--rpc/args_test.go28
4 files changed, 113 insertions, 199 deletions
diff --git a/rpc/api.go b/rpc/api.go
index 961440493..9a994ebbd 100644
--- a/rpc/api.go
+++ b/rpc/api.go
@@ -7,7 +7,6 @@ import (
"path"
"strings"
"sync"
- "time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
@@ -15,15 +14,13 @@ import (
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event"
- "github.com/ethereum/go-ethereum/event/filter"
"github.com/ethereum/go-ethereum/state"
"github.com/ethereum/go-ethereum/xeth"
)
var (
- defaultGasPrice = big.NewInt(150000000000)
- defaultGas = big.NewInt(500000)
- filterTickerTime = 5 * time.Minute
+ defaultGasPrice = big.NewInt(150000000000)
+ defaultGas = big.NewInt(500000)
)
type EthereumApi struct {
@@ -31,17 +28,9 @@ type EthereumApi struct {
xethMu sync.RWMutex
mux *event.TypeMux
- quit chan struct{}
- filterManager *filter.FilterManager
-
- logMut sync.RWMutex
- logs map[int]*logFilter
-
- messagesMut sync.RWMutex
- messages map[int]*whisperFilter
- // Register keeps a list of accounts and transaction data
- regmut sync.Mutex
- register map[string][]*NewTxArgs
+ // // Register keeps a list of accounts and transaction data
+ // regmut sync.Mutex
+ // register map[string][]*NewTxArgs
db common.Database
}
@@ -49,16 +38,10 @@ type EthereumApi struct {
func NewEthereumApi(eth *xeth.XEth, dataDir string) *EthereumApi {
db, _ := ethdb.NewLDBDatabase(path.Join(dataDir, "dapps"))
api := &EthereumApi{
- eth: eth,
- mux: eth.Backend().EventMux(),
- quit: make(chan struct{}),
- filterManager: filter.NewFilterManager(eth.Backend().EventMux()),
- logs: make(map[int]*logFilter),
- messages: make(map[int]*whisperFilter),
- db: db,
+ eth: eth,
+ mux: eth.Backend().EventMux(),
+ db: db,
}
- go api.filterManager.Start()
- go api.start()
return api
}
@@ -81,37 +64,8 @@ func (self *EthereumApi) xethWithStateNum(num int64) *xeth.XEth {
return self.xeth().WithState(st)
}
-func (self *EthereumApi) start() {
- timer := time.NewTicker(filterTickerTime)
-done:
- for {
- select {
- case <-timer.C:
- self.logMut.Lock()
- self.messagesMut.Lock()
- for id, filter := range self.logs {
- if time.Since(filter.timeout) > 20*time.Second {
- self.filterManager.UninstallFilter(id)
- delete(self.logs, id)
- }
- }
-
- for id, filter := range self.messages {
- if time.Since(filter.timeout) > 20*time.Second {
- self.xeth().Whisper().Unwatch(id)
- delete(self.messages, id)
- }
- }
- self.logMut.Unlock()
- self.messagesMut.Unlock()
- case <-self.quit:
- break done
- }
- }
-}
-
-func (self *EthereumApi) stop() {
- close(self.quit)
+func (self *EthereumApi) getStateWithNum(num int64) *xeth.State {
+ return self.xethWithStateNum(num).State()
}
// func (self *EthereumApi) Register(args string, reply *interface{}) error {
@@ -145,88 +99,43 @@ func (self *EthereumApi) stop() {
// }
func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) error {
- var id int
- filter := core.NewFilter(self.xeth().Backend())
- filter.SetOptions(toFilterOptions(args))
- filter.LogsCallback = func(logs state.Logs) {
- self.logMut.Lock()
- defer self.logMut.Unlock()
-
- self.logs[id].add(logs...)
- }
- id = self.filterManager.InstallFilter(filter)
- self.logs[id] = &logFilter{timeout: time.Now()}
-
+ opts := toFilterOptions(args)
+ id := self.xeth().RegisterFilter(opts)
*reply = common.ToHex(big.NewInt(int64(id)).Bytes())
return nil
}
func (self *EthereumApi) UninstallFilter(id int, reply *interface{}) error {
- if _, ok := self.logs[id]; ok {
- delete(self.logs, id)
- }
+ *reply = self.xeth().UninstallFilter(id)
- self.filterManager.UninstallFilter(id)
- *reply = true
return nil
}
func (self *EthereumApi) NewFilterString(args *FilterStringArgs, reply *interface{}) error {
- var id int
- filter := core.NewFilter(self.xeth().Backend())
-
- callback := func(block *types.Block) {
- self.logMut.Lock()
- defer self.logMut.Unlock()
-
- self.logs[id].add(&state.StateLog{})
- }
-
- switch args.Word {
- case "pending":
- filter.PendingCallback = callback
- case "latest":
- filter.BlockCallback = callback
- default:
- return NewValidationError("Word", "Must be `latest` or `pending`")
+ if err := args.requirements(); err != nil {
+ return err
}
- id = self.filterManager.InstallFilter(filter)
- self.logs[id] = &logFilter{timeout: time.Now()}
+ id := self.xeth().NewFilterString(args.Word)
*reply = common.ToHex(big.NewInt(int64(id)).Bytes())
-
return nil
}
func (self *EthereumApi) FilterChanged(id int, reply *interface{}) error {
- self.logMut.Lock()
- defer self.logMut.Unlock()
-
- if self.logs[id] != nil {
- *reply = NewLogsRes(self.logs[id].get())
- }
-
+ *reply = NewLogsRes(self.xeth().FilterChanged(id))
return nil
}
func (self *EthereumApi) Logs(id int, reply *interface{}) error {
- self.logMut.Lock()
- defer self.logMut.Unlock()
-
- filter := self.filterManager.GetFilter(id)
- if filter != nil {
- *reply = NewLogsRes(filter.Find())
- }
+ *reply = NewLogsRes(self.xeth().Logs(id))
return nil
}
func (self *EthereumApi) AllLogs(args *FilterOptions, reply *interface{}) error {
- filter := core.NewFilter(self.xeth().Backend())
- filter.SetOptions(toFilterOptions(args))
-
- *reply = NewLogsRes(filter.Find())
+ opts := toFilterOptions(args)
+ *reply = NewLogsRes(self.xeth().AllLogs(opts))
return nil
}
@@ -253,6 +162,11 @@ func (p *EthereumApi) Transact(args *NewTxArgs, reply *interface{}) (err error)
p.register[ags.From] = append(p.register[args.From], args)
}
*/
+
+ if err := args.requirements(); err != nil {
+ return err
+ }
+
// TODO: align default values to have the same type, e.g. not depend on
// common.Value conversions later on
if args.Gas.Cmp(big.NewInt(0)) == 0 {
@@ -304,30 +218,22 @@ func (p *EthereumApi) GetStorageAt(args *GetStorageAtArgs, reply *interface{}) e
}
func (p *EthereumApi) NewWhisperFilter(args *WhisperFilterArgs, reply *interface{}) error {
- var id int
opts := new(xeth.Options)
opts.From = args.From
opts.To = args.To
opts.Topics = args.Topics
- opts.Fn = func(msg xeth.WhisperMessage) {
- p.messagesMut.Lock()
- defer p.messagesMut.Unlock()
- p.messages[id].add(msg) // = append(p.messages[id], msg)
- }
- id = p.xeth().Whisper().Watch(opts)
- p.messages[id] = &whisperFilter{timeout: time.Now()}
+ id := p.xeth().NewWhisperFilter(opts)
*reply = common.ToHex(big.NewInt(int64(id)).Bytes())
return nil
}
-func (self *EthereumApi) MessagesChanged(id int, reply *interface{}) error {
- self.messagesMut.Lock()
- defer self.messagesMut.Unlock()
-
- if self.messages[id] != nil {
- *reply = self.messages[id].get()
- }
+func (p *EthereumApi) UninstallWhisperFilter(id int, reply *interface{}) error {
+ *reply = p.xeth().UninstallWhisperFilter(id)
+ return nil
+}
+func (self *EthereumApi) MessagesChanged(id int, reply *interface{}) error {
+ *reply = self.xeth().MessagesChanged(id)
return nil
}
@@ -347,7 +253,7 @@ func (p *EthereumApi) GetBlockByNumber(blocknum int64, includetx bool) (*BlockRe
func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error {
// Spec at https://github.com/ethereum/wiki/wiki/Generic-JSON-RPC
- rpclogger.Debugf("%s %s", req.Method, req.Params)
+ rpclogger.Infof("%s %s", req.Method, req.Params)
switch req.Method {
case "web3_sha3":
args := new(Sha3Args)
@@ -684,11 +590,7 @@ func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error
return err
}
- if _, ok := p.messages[args.Id]; ok {
- delete(p.messages, args.Id)
- }
-
- *reply = true
+ return p.UninstallWhisperFilter(args.Id, reply)
case "shh_getFilterChanges":
args := new(FilterIdArgs)
if err := json.Unmarshal(req.Params, &args); err != nil {
@@ -734,7 +636,7 @@ func (self *EthereumApi) xeth() *xeth.XEth {
return self.eth
}
-func toFilterOptions(options *FilterOptions) core.FilterOptions {
+func toFilterOptions(options *FilterOptions) *core.FilterOptions {
var opts core.FilterOptions
// Convert optional address slice/string to byte slice
@@ -767,38 +669,5 @@ func toFilterOptions(options *FilterOptions) core.FilterOptions {
}
opts.Topics = topics
- return opts
-}
-
-type whisperFilter struct {
- messages []xeth.WhisperMessage
- timeout time.Time
- id int
-}
-
-func (w *whisperFilter) add(msgs ...xeth.WhisperMessage) {
- w.messages = append(w.messages, msgs...)
-}
-func (w *whisperFilter) get() []xeth.WhisperMessage {
- w.timeout = time.Now()
- tmp := w.messages
- w.messages = nil
- return tmp
-}
-
-type logFilter struct {
- logs state.Logs
- timeout time.Time
- id int
-}
-
-func (l *logFilter) add(logs ...state.Log) {
- l.logs = append(l.logs, logs...)
-}
-
-func (l *logFilter) get() state.Logs {
- l.timeout = time.Now()
- tmp := l.logs
- l.logs = nil
- return tmp
+ return &opts
}
diff --git a/rpc/api_test.go b/rpc/api_test.go
index ec03822c5..727ade007 100644
--- a/rpc/api_test.go
+++ b/rpc/api_test.go
@@ -2,9 +2,9 @@ package rpc
import (
"encoding/json"
- "sync"
+ // "sync"
"testing"
- "time"
+ // "time"
)
func TestWeb3Sha3(t *testing.T) {
@@ -24,33 +24,33 @@ func TestWeb3Sha3(t *testing.T) {
}
}
-func TestFilterClose(t *testing.T) {
- t.Skip()
- api := &EthereumApi{
- logs: make(map[int]*logFilter),
- messages: make(map[int]*whisperFilter),
- quit: make(chan struct{}),
- }
-
- filterTickerTime = 1
- api.logs[0] = &logFilter{}
- api.messages[0] = &whisperFilter{}
- var wg sync.WaitGroup
- wg.Add(1)
- go api.start()
- go func() {
- select {
- case <-time.After(500 * time.Millisecond):
- api.stop()
- wg.Done()
- }
- }()
- wg.Wait()
- if len(api.logs) != 0 {
- t.Error("expected logs to be empty")
- }
-
- if len(api.messages) != 0 {
- t.Error("expected messages to be empty")
- }
-}
+// func TestFilterClose(t *testing.T) {
+// t.Skip()
+// api := &EthereumApi{
+// logs: make(map[int]*logFilter),
+// messages: make(map[int]*whisperFilter),
+// quit: make(chan struct{}),
+// }
+
+// filterTickerTime = 1
+// api.logs[0] = &logFilter{}
+// api.messages[0] = &whisperFilter{}
+// var wg sync.WaitGroup
+// wg.Add(1)
+// go api.start()
+// go func() {
+// select {
+// case <-time.After(500 * time.Millisecond):
+// api.stop()
+// wg.Done()
+// }
+// }()
+// wg.Wait()
+// if len(api.logs) != 0 {
+// t.Error("expected logs to be empty")
+// }
+
+// if len(api.messages) != 0 {
+// t.Error("expected messages to be empty")
+// }
+// }
diff --git a/rpc/args.go b/rpc/args.go
index 992ea1eed..bd6be5a0f 100644
--- a/rpc/args.go
+++ b/rpc/args.go
@@ -145,6 +145,13 @@ func (args *NewTxArgs) UnmarshalJSON(b []byte) (err error) {
return nil
}
+func (args *NewTxArgs) requirements() error {
+ if len(args.From) == 0 {
+ return NewValidationError("From", "Is required")
+ }
+ return nil
+}
+
type GetStorageArgs struct {
Address string
BlockNumber int64
@@ -602,6 +609,16 @@ func (args *FilterStringArgs) UnmarshalJSON(b []byte) (err error) {
return nil
}
+func (args *FilterStringArgs) requirements() error {
+ switch args.Word {
+ case "latest", "pending":
+ break
+ default:
+ return NewValidationError("Word", "Must be `latest` or `pending`")
+ }
+ return nil
+}
+
type FilterIdArgs struct {
Id int
}
diff --git a/rpc/args_test.go b/rpc/args_test.go
index 872c535fa..0d8dc4085 100644
--- a/rpc/args_test.go
+++ b/rpc/args_test.go
@@ -219,6 +219,34 @@ func TestNewTxArgsEmpty(t *testing.T) {
}
}
+func TestNewTxArgsReqs(t *testing.T) {
+ args := new(NewTxArgs)
+ args.From = "0xb60e8dd61c5d32be8058bb8eb970870f07233155"
+
+ err := args.requirements()
+ switch err.(type) {
+ case nil:
+ break
+ default:
+ t.Errorf("Get %T", err)
+ }
+}
+
+func TestNewTxArgsReqsFromBlank(t *testing.T) {
+ args := new(NewTxArgs)
+ args.From = ""
+
+ err := args.requirements()
+ switch err.(type) {
+ case nil:
+ t.Error("Expected error but didn't get one")
+ case *ValidationError:
+ break
+ default:
+ t.Error("Wrong type of error")
+ }
+}
+
func TestGetStorageArgs(t *testing.T) {
input := `["0x407d73d8a49eeb85d32cf465507dd71d507100c1", "latest"]`
expected := new(GetStorageArgs)
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-29 09:03:49 +0800