code: func NewPullEngineWithFilter(participant PullAdapter, sleepTime time.Duration, df DigestFilter,
  	config PullEngineConfig) *PullEngine {
  	engine := &PullEngine{
  		PullAdapter:        participant,
  		stopFlag:           int32(0),
  		state:              util.NewSet(),
  		item2owners:        make(map[string][]string),
  		peers2nonces:       make(map[string]uint64),
  		nonces2peers:       make(map[uint64]string),
  		acceptingDigests:   int32(0),
  		acceptingResponses: int32(0),
  		incomingNONCES:     util.NewSet(),
  		outgoingNONCES:     util.NewSet(),
  		digFilter:          df,
  		digestWaitTime:     config.DigestWaitTime,
  		requestWaitTime:    config.RequestWaitTime,
  		responseWaitTime:   config.ResponseWaitTime,
  	}
  
  	go func() {
  		for !engine.toDie() {
  			time.Sleep(sleepTime)
  			if engine.toDie() {
  				return
  			}
  			engine.initiatePull()
  		}
  	}()
  
  	return engine
  }
docs: // NewPullEngineWithFilter creates an instance of a PullEngine with a certain sleep time
  // between pull initiations, and uses the given filters when sending digests and responses