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

refactor notifying logs

Browse source 
Signed-off-by: jsvisa <delweng@gmail.com>
This commit is contained in:
Sina Mahmoodi 2023-08-10 18:05:39 +02:00 committed by jsvisa
parent d78c6c74eb
commit 3083f77c87
1 changed files with 17 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
eth/filters/api.go
View file

@ -299,7 +299,7 @@ func (api *FilterAPI) logs(ctx context.Context, notifier notifier, rpcSub *rpc.S
} }
// liveLogs only retrieves live logs. // liveLogs only retrieves live logs.
func (api *FilterAPI) liveLogs(notify notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error { func (api *FilterAPI) liveLogs(notifier notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error {
matchedLogs := make(chan []*types.Log) matchedLogs := make(chan []*types.Log)
logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), matchedLogs) logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), matchedLogs)
if err != nil { if err != nil {
@ -310,14 +310,11 @@ func (api *FilterAPI) liveLogs(notify notifier, rpcSub *rpc.Subscription, crit F
for { for {
select { select {
case logs := <-matchedLogs: case logs := <-matchedLogs:
for _, log := range logs { notifyLogsIf(notifier, rpcSub.ID, logs, nil)
log := log
notify.Notify(rpcSub.ID, &log)
}
case <-rpcSub.Err(): // client send an unsubscribe request case <-rpcSub.Err(): // client send an unsubscribe request
logsSub.Unsubscribe() logsSub.Unsubscribe()
return return
case <-notify.Closed(): // connection dropped case <-notifier.Closed(): // connection dropped
logsSub.Unsubscribe() logsSub.Unsubscribe()
return return
} }
@ -375,10 +372,7 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
continue continue
} }
if liveOnly { if liveOnly {
for _, log := range logs { notifyLogsIf(notifier, rpcSub.ID, logs, nil)
log := log
notifier.Notify(rpcSub.ID, &log)
}
continue continue
} }
reorgBlock := logs[0].BlockNumber reorgBlock := logs[0].BlockNumber
@ -392,19 +386,13 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
if logs[0].Removed { if logs[0].Removed {
// Send removed logs notification up until the point // Send removed logs notification up until the point
// we have delivered logs from old chain. // we have delivered logs from old chain.
for _, log := range logs { notifyLogsIf(notifier, rpcSub.ID, logs, func(log *types.Log) bool {
if log.BlockNumber <= delivered { return log.BlockNumber <= delivered
log := log })
notifier.Notify(rpcSub.ID, &log)
}
}
} else { } else {
// New logs are emitted for the whole new chain since the reorg block. // New logs are emitted for the whole new chain since the reorg block.
// Send them all. // Send them all.
for _, log := range logs { notifyLogsIf(notifier, rpcSub.ID, logs, nil)
log := log
notifier.Notify(rpcSub.ID, &log)
}
} }
case log := <-histLogs: case log := <-histLogs:
// If ChainReorg is detected, we need to deliver the last block's full logs // If ChainReorg is detected, we need to deliver the last block's full logs
@ -472,6 +460,15 @@ func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, histLogs chan<
} }
} }
func notifyLogsIf(notifier notifier, id rpc.ID, logs []*types.Log, cond func(log *types.Log) bool) {
for _, log := range logs {
if cond == nil || cond(log) {
log := log
notifier.Notify(id, &log)
}
}
}
// FilterCriteria represents a request to create a new filter. // FilterCriteria represents a request to create a new filter.
// Same as ethereum.FilterQuery but with UnmarshalJSON() method. // Same as ethereum.FilterQuery but with UnmarshalJSON() method.
type FilterCriteria ethereum.FilterQuery type FilterCriteria ethereum.FilterQuery