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: handle toBlock

Browse source 
Signed-off-by: jsvisa <delweng@gmail.com>
This commit is contained in:
jsvisa 2023-06-08 08:29:05 +00:00
parent 390a4b3018
commit d30d0f0a15
1 changed files with 18 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
eth/filters/api.go
View file

@ -309,7 +309,7 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
return rpcSub, err return rpcSub, err
} }
filterHistLogs := func(n int64) error { filterHistLogs := func(n int64) (int64, error) {
// the ctx will be canceled after this callback(filter.Logs) is called // the ctx will be canceled after this callback(filter.Logs) is called
// and we are run in a background goroutine, use a new context instead // and we are run in a background goroutine, use a new context instead
var ( var (
@ -320,18 +320,22 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
// get the latest block header // get the latest block header
head := api.sys.backend.CurrentHeader().Number.Int64() head := api.sys.backend.CurrentHeader().Number.Int64()
if n >= head { if n >= head {
return nil return head, nil
} }
// do historical sync from n to head to := big.NewInt(head)
f := api.sys.NewRangeFilter(n, head, crit.Addresses, crit.Topics) if toBlock := crit.ToBlock; toBlock != nil && toBlock.Sign() > 0 && toBlock.Cmp(to) < 0 {
to = toBlock
}
// do historical sync from n to min(head, to)
f := api.sys.NewRangeFilter(n, to.Int64(), crit.Addresses, crit.Topics)
logChan, errChan := f.rangeLogsAsync(cctx) logChan, errChan := f.rangeLogsAsync(cctx)
// subscribe rmLogs // subscribe rmLogs
query := ethereum.FilterQuery{FromBlock: big.NewInt(n), ToBlock: big.NewInt(head), Addresses: crit.Addresses, Topics: crit.Topics} query := ethereum.FilterQuery{FromBlock: big.NewInt(n), ToBlock: to, Addresses: crit.Addresses, Topics: crit.Topics}
rmLogsSub, err := api.events.SubscribeLogs(query, rmLogsCh) rmLogsSub, err := api.events.SubscribeLogs(query, rmLogsCh)
if err != nil { if err != nil {
return err return 0, err
} }
var rmLogs []*types.Log var rmLogs []*types.Log
@ -340,7 +344,7 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
select { select {
case <-notifier.Closed(): case <-notifier.Closed():
rmLogsSub.Unsubscribe() rmLogsSub.Unsubscribe()
return errors.New("connection dropped") return 0, errors.New("connection dropped")
case log := <-logChan: case log := <-logChan:
notifier.Notify(rpcSub.ID, &log) notifier.Notify(rpcSub.ID, &log)
case logs := <-rmLogsCh: case logs := <-rmLogsCh:
@ -353,7 +357,7 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
// range filter is done or error, let's also stop the rmLogs subscribe // range filter is done or error, let's also stop the rmLogs subscribe
rmLogsSub.Unsubscribe() rmLogsSub.Unsubscribe()
if err != nil { if err != nil {
return err return 0, err
} }
break FORLOOP break FORLOOP
} }
@ -372,7 +376,12 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
n := crit.FromBlock.Int64() n := crit.FromBlock.Int64()
go func() { go func() {
// do historical sync first // do historical sync first
if err := filterHistLogs(n); err != nil { head, err := filterHistLogs(n)
if err != nil {
return
}
// if toBlock is limited and handled, no need to subscribe live logs anymore
if toBlock := crit.ToBlock; toBlock != nil && toBlock.Sign() > 0 && toBlock.Cmp(big.NewInt(head)) <= 0 {
return return
} }
// then subscribe from the header // then subscribe from the header