Method poll

graphql/subscription/poller.go:164–233

(req *pollRequest)

}

func (p *Poller) poll(req *pollRequest) {
	p.RLock()
	resolver := p.resolver
	p.RUnlock()

	pollID := uint64(0)
	for {
		pollID++
		time.Sleep(x.Config.GraphQL.GetDuration("poll-interval"))

		globalEpoch := atomic.LoadUint64(p.globalEpoch)
		if req.localEpoch != globalEpoch || globalEpoch == math.MaxUint64 {
			// There is a schema change since local epoch is diffrent from global schema epoch.
			// We'll terminate all the subscription for this bucket. So, that all client can
			// reconnect and listen for new schema.
			p.terminateSubscriptions(req.bucketID)
		}

		ctx := x.AttachAccessJwt(context.Background(), &http.Request{Header: req.graphqlReq.Header})
		res := resolver.Resolve(ctx, req.graphqlReq)

		currentHash := farm.Fingerprint64(res.Data.Bytes())

		if req.prevHash == currentHash {
			if pollID%2 != 0 {
				// Don't update if there is no change in response.
				continue
			}
			// Every second poll, we'll check if there is any active subscription for the
			// current goroutine. If not we'll terminate this poll.
			p.Lock()
			subscribers, ok := p.pollRegistry[req.bucketID]
			if !ok || len(subscribers) == 0 {
				delete(p.pollRegistry, req.bucketID)
				p.Unlock()
				return
			}
			for subscriberID, subscriber := range subscribers {
				if !subscriber.expiry.IsZero() && time.Now().After(subscriber.expiry) {
					p.terminateSubscription(req.bucketID, subscriberID)
				}

			}
			p.Unlock()
			continue
		}
		req.prevHash = currentHash

		p.Lock()
		subscribers, ok := p.pollRegistry[req.bucketID]
		if !ok || len(subscribers) == 0 {
			// There is no subscribers to push the update. So, kill the current polling
			// go routine.
			delete(p.pollRegistry, req.bucketID)
			p.Unlock()
			return
		}