func (s *PullStorer) IntervalChunks(ctx context.Context, bin uint8, from, to uint64, limit int) (chs []swarm.Address, topmost uint64, err error) {

	type result struct {
		chs     []swarm.Address
		topmost uint64
	}
	s.metrics.TotalSubscribePullRequests.Inc()
	defer s.metrics.TotalSubscribePullRequestsComplete.Inc()

	v, _, err := s.intervalsSF.Do(ctx, fmt.Sprintf("%v-%v-%v-%v", bin, from, to, limit), func(ctx context.Context) (interface{}, error) {
		// call iterator, iterate either until upper bound or limit reached
		// return addresses, topmost is the topmost bin ID
		var (
			timer  *time.Timer
			timerC <-chan time.Time
		)
		s.metrics.SubscribePullsStarted.Inc()
		ch, dbClosed, stop := s.SubscribePull(ctx, bin, from, to)
		defer func(start time.Time) {
			stop()
			if timer != nil {
				timer.Stop()
			}
			s.metrics.SubscribePullsComplete.Inc()
		}(time.Now())

		var nomore bool

	LOOP:
		for limit > 0 {
			select {
			case v, ok := <-ch:
				if !ok {
					nomore = true
					break LOOP
				}
				chs = append(chs, v.Address)
				if v.BinID > topmost {
					topmost = v.BinID
				}
				limit--
				if timer == nil {
					timer = time.NewTimer(batchTimeout)
				} else {
					if !timer.Stop() {
						<-timer.C
					}
					timer.Reset(batchTimeout)
				}
				timerC = timer.C
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-timerC:
				// return batch if new chunks are not received after some time
				break LOOP
			}
		}

		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-dbClosed:
			return nil, ErrDbClosed
		default:
		}

		if nomore {
			// end of interval reached. no more chunks so interval is complete
			// return requested `to`. it could be that len(chs) == 0 if the interval
			// is empty
			topmost = to
		}

		return &result{chs: chs, topmost: topmost}, nil
	})

	if err != nil {
		s.metrics.SubscribePullsFailures.Inc()
		return nil, 0, err
	}
	r := v.(*result)
	return r.chs, r.topmost, nil
}
// IntervalChunks collects chunk for a requested interval.
IntervalChunks collects chunk for a requested interval.
// IntervalChunks collects chunk for a requested interval .
func (s *PullStorer) IntervalChunks(ctx context.Context, bin uint8, from, to uint64, limit int) (chs []swarm.Address, topmost uint64, err error) {
	type result struct {
		chs     []swarm.Address
		topmost uint64
	}
	s.metrics.TotalSubscribePullRequests.Inc()
	defer s.metrics.TotalSubscribePullRequestsComplete.Inc()
	v, _, err := s.intervalsSF.Do(ctx, fmt.Sprintf("%v-%v-%v-%v", bin, from, to, limit), func(ctx context.Context) (interface{}, error) {
		var (
			timer  *time.Timer
			timerC <-chan time.Time
		)
		s.metrics.SubscribePullsStarted.Inc()
		ch, dbClosed, stop := s.SubscribePull(ctx, bin, from, to)
		defer func(start time.Time) {
			stop()
			if timer != nil {
				timer.Stop()
			}
			s.metrics.SubscribePullsComplete.Inc()
		}(time.Now())
		var nomore bool
	LOOP:
		for limit > 0 {
			select {
			case v, ok := <-ch:
				if !ok {
					nomore = true
					break LOOP
				}
				chs = append(chs, v.Address)
				if v.BinID > topmost {
					topmost = v.BinID
				}
				limit--
				if timer == nil {
					timer = time.NewTimer(batchTimeout)
				} else {
					if !timer.Stop() {
						<-timer.C
					}
					timer.Reset(batchTimeout)
				}
				timerC = timer.C
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-timerC:
				break LOOP
			}
		}
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-dbClosed:
			return nil, ErrDbClosed
		default:
		}
		if nomore {
			topmost = to
		}
		return &result{chs: chs, topmost: topmost}, nil
	})
	if err != nil {
		s.metrics.SubscribePullsFailures.Inc()
		return nil, 0, err
	}
	r := v.(*result)
	return r.chs, r.topmost, nil
}
func ( s * PullStorer ) IntervalChunks ( ctx context . Context , bin uint8 , from , to uint64 , limit int ) ( chs [ ] swarm . Address , topmost uint64 , err error ) { type result struct { chs [ ] swarm . Address topmost uint64 } s . metrics . TotalSubscribePullRequests . Inc ( ) defer s . metrics . TotalSubscribePullRequestsComplete . Inc ( ) v , _ , err := s . intervalsSF . Do ( ctx , fmt . Sprintf ( " " , bin , from , to , limit ) , func ( ctx context . Context ) ( interface { } , error ) { var ( timer * time . Timer timerC <- chan time . Time ) s . metrics . SubscribePullsStarted . Inc ( ) ch , dbClosed , stop := s . SubscribePull ( ctx , bin , from , to ) defer func ( start time . Time ) { stop ( ) if timer != nil { timer . Stop ( ) } s . metrics . SubscribePullsComplete . Inc ( ) } ( time . Now ( ) ) var nomore bool LOOP : for limit > 0 { select { case v , ok := <- ch : if ! ok { nomore = true break LOOP } chs = append ( chs , v . Address ) if v . BinID > topmost { topmost = v . BinID } limit -- if timer == nil { timer = time . NewTimer ( batchTimeout ) } else { if ! timer . Stop ( ) { <- timer . C } timer . Reset ( batchTimeout ) } timerC = timer . C case <- ctx . Done ( ) : return nil , ctx . Err ( ) case <- timerC : break LOOP } } select { case <- ctx . Done ( ) : return nil , ctx . Err ( ) case <- dbClosed : return nil , ErrDbClosed default : } if nomore { topmost = to } return & result { chs : chs , topmost : topmost } , nil } ) if err != nil { s . metrics . SubscribePullsFailures . Inc ( ) return nil , 0 , err } r := v . ( * result ) return r . chs , r . topmost , nil }
IntervalChunks collects chunk for a requested interval.
IntervalChunks collects chunk for a requested interval .
// call iterator, iterate either until upper bound or limit reached // return addresses, topmost is the topmost bin ID // return batch if new chunks are not received after some time // end of interval reached. no more chunks so interval is complete // return requested `to`. it could be that len(chs) == 0 if the interval // is empty