core/rpc: add continue flag to iterating functions

Most of the time we don't need to get all transfers from the DB and
deserialize them.

pkg/core/blockchain.go

@@ -784,7 +784,7 @@ func (bc *Blockchain) processNEP5Transfer(cache *dao.Cached, h util.Uint256, b *
 }
 
 // ForEachNEP5Transfer executes f for each nep5 transfer in log.
-func (bc *Blockchain) ForEachNEP5Transfer(acc util.Uint160, f func(*state.NEP5Transfer) error) error {
+func (bc *Blockchain) ForEachNEP5Transfer(acc util.Uint160, f func(*state.NEP5Transfer) (bool, error)) error {
 	balances, err := bc.dao.GetNEP5Balances(acc)
 	if err != nil {
 		return nil
@@ -794,10 +794,13 @@ func (bc *Blockchain) ForEachNEP5Transfer(acc util.Uint160, f func(*state.NEP5Tr
 		if err != nil {
 			return nil
 		}
-		err = lg.ForEach(f)
+		cont, err := lg.ForEach(f)
 		if err != nil {
 			return err
 		}
+		if !cont {
+			break
+		}
 	}
 	return nil
 }

pkg/core/blockchainer/blockchainer.go

@@ -31,7 +31,7 @@ type Blockchainer interface {
 	GetContractScriptHash(id int32) (util.Uint160, error)
 	GetEnrollments() ([]state.Validator, error)
 	GetGoverningTokenBalance(acc util.Uint160) (*big.Int, uint32)
-	ForEachNEP5Transfer(util.Uint160, func(*state.NEP5Transfer) error) error
+	ForEachNEP5Transfer(util.Uint160, func(*state.NEP5Transfer) (bool, error)) error
 	GetHeaderHash(int) util.Uint256
 	GetHeader(hash util.Uint256) (*block.Header, error)
 	CurrentHeaderHash() util.Uint256

pkg/core/state/nep5.go

@@ -102,9 +102,9 @@ func (lg *NEP5TransferLog) Append(tr *NEP5Transfer) error {
 }
 
 // ForEach iterates over transfer log returning on first error.
-func (lg *NEP5TransferLog) ForEach(f func(*NEP5Transfer) error) error {
+func (lg *NEP5TransferLog) ForEach(f func(*NEP5Transfer) (bool, error)) (bool, error) {
 	if lg == nil || len(lg.Raw) == 0 {
-		return nil
+		return true, nil
 	}
 	transfers := make([]NEP5Transfer, lg.Size())
 	r := io.NewBinReaderFromBuf(lg.Raw[1:])
@@ -112,14 +112,18 @@ func (lg *NEP5TransferLog) ForEach(f func(*NEP5Transfer) error) error {
 		transfers[i].DecodeBinary(r)
 	}
 	if r.Err != nil {
-		return r.Err
+		return false, r.Err
 	}
 	for i := len(transfers) - 1; i >= 0; i-- {
-		if err := f(&transfers[i]); err != nil {
-			return err
+		cont, err := f(&transfers[i])
+		if err != nil {
+			return false, err
+		}
+		if !cont {
+			return false, nil
 		}
 	}
-	return nil
+	return true, nil
 }
 
 // Size returns an amount of transfer written in log.

pkg/core/state/nep5_test.go

@@ -30,13 +30,13 @@ func TestNEP5TransferLog_Append(t *testing.T) {
 	require.Equal(t, len(expected), lg.Size())
 
 	i := len(expected) - 1
-	err := lg.ForEach(func(tr *NEP5Transfer) error {
+	cont, err := lg.ForEach(func(tr *NEP5Transfer) (bool, error) {
 		require.Equal(t, expected[i], tr)
 		i--
-		return nil
+		return true, nil
 	})
 	require.NoError(t, err)
-
+	require.True(t, cont)
 }
 
 func TestNEP5Tracker_EncodeBinary(t *testing.T) {

pkg/network/helper_test.go

@@ -95,7 +95,7 @@ func (chain testChain) GetHeader(hash util.Uint256) (*block.Header, error) {
 func (chain testChain) GetNextBlockValidators() ([]*keys.PublicKey, error) {
 	panic("TODO")
 }
-func (chain testChain) ForEachNEP5Transfer(util.Uint160, func(*state.NEP5Transfer) error) error {
+func (chain testChain) ForEachNEP5Transfer(util.Uint160, func(*state.NEP5Transfer) (bool, error)) error {
 	panic("TODO")
 }
 func (chain testChain) GetNEP5Balances(util.Uint160) *state.NEP5Balances {

pkg/rpc/server/server.go

@@ -599,14 +599,17 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err
 		Sent:     []result.NEP5Transfer{},
 	}
 	cache := make(map[int32]decimals)
-	err = s.chain.ForEachNEP5Transfer(u, func(tr *state.NEP5Transfer) error {
-		if tr.Timestamp < start || tr.Timestamp > end ||
+	err = s.chain.ForEachNEP5Transfer(u, func(tr *state.NEP5Transfer) (bool, error) {
+		if tr.Timestamp > end {
+			return true, nil
+		}
+		if tr.Timestamp < start ||
 			(limit != 0 && (len(bs.Received)+len(bs.Sent) >= limit)) {
-			return nil
+			return false, nil
 		}
 		d, err := s.getDecimals(tr.Asset, cache)
 		if err != nil {
-			return nil
+			return false, err
 		}
 		transfer := result.NEP5Transfer{
 			Timestamp: tr.Timestamp,
@@ -620,7 +623,7 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err
 				transfer.Address = address.Uint160ToString(tr.From)
 			}
 			bs.Received = append(bs.Received, transfer)
-			return nil
+			return true, nil
 		}
 
 		transfer.Amount = amountToString(new(big.Int).Neg(&tr.Amount), d.Value)
@@ -628,7 +631,7 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err
 			transfer.Address = address.Uint160ToString(tr.To)
 		}
 		bs.Sent = append(bs.Sent, transfer)
-		return nil
+		return true, nil
 	})
 	if err != nil {
 		return nil, response.NewInternalServerError("invalid NEP5 transfer log", err)