forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 13:16:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/filters: add notifier interface doc

Browse source 
Signed-off-by: jsvisa <delweng@gmail.com>
This commit is contained in:
jsvisa 2023-06-27 00:00:51 +08:00
parent c7bfb244df
commit 89cce937ab
1 changed files with 33 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
eth/filters/api.go
View file

@ -261,6 +261,8 @@ func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
return rpcSub, nil
}
// notifier is used for broadcasting data(eg: logs) to rpc receivers
// used in unit testing
type notifier interface {
Notify(id rpc.ID, data interface{}) error
Closed() <-chan interface{}
@ -277,6 +279,37 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
return rpcSub, err
}
// logs is the inner implmention of logs filter
func (api *FilterAPI) logs(ctx context.Context, notifier notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error {
isLiveFilter := true
if crit.FromBlock != nil {
isLiveFilter = crit.FromBlock.Sign() < 0
}
// do live filter only
if isLiveFilter {
return api.liveLogs(ctx, notifier, rpcSub, crit)
}
// if toBlock is limited and handled, no need to subscribe live logs anymore
if toBlock := crit.ToBlock; toBlock != nil {
if header := api.sys.backend.CurrentHeader(); header != nil && toBlock.Sign() > 0 && toBlock.Cmp(header.Number) <= 0 {
return errors.New("historical only log subscription is not supported")
}
}
go func() {
// do historical sync first
_, err := api.histLogs(ctx, notifier, rpcSub, crit)
if err != nil {
return
}
// then subscribe from the header
api.liveLogs(ctx, notifier, rpcSub, crit)
}()
return nil
}
// liveLogs only retrieves live logs
func (api *FilterAPI) liveLogs(ctx context.Context, notify notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error {
matchedLogs := make(chan []*types.Log)
@ -386,37 +419,6 @@ func (api *FilterAPI) histLogs(ctx context.Context, notifier notifier, rpcSub *r
}
}
// logs is the inner implmention of logs filter
func (api *FilterAPI) logs(ctx context.Context, notifier notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error {
isLiveFilter := true
if crit.FromBlock != nil {
isLiveFilter = crit.FromBlock.Sign() < 0
}
// do live filter only
if isLiveFilter {
return api.liveLogs(ctx, notifier, rpcSub, crit)
}
// if toBlock is limited and handled, no need to subscribe live logs anymore
if toBlock := crit.ToBlock; toBlock != nil {
if header := api.sys.backend.CurrentHeader(); header != nil && toBlock.Sign() > 0 && toBlock.Cmp(header.Number) <= 0 {
return errors.New("historical only log subscription is not supported")
}
}
go func() {
// do historical sync first
_, err := api.histLogs(ctx, notifier, rpcSub, crit)
if err != nil {
return
}
// then subscribe from the header
api.liveLogs(ctx, notifier, rpcSub, crit)
}()
return nil
}
// FilterCriteria represents a request to create a new filter.
// Same as ethereum.FilterQuery but with UnmarshalJSON() method.
type FilterCriteria ethereum.FilterQuery