<commit_before>package apidApigeeSync

import (
	"bytes"
	"encoding/json"
	"errors"
	"io/ioutil"
	"net/http"
	"net/url"
	"path"
	"time"

	"github.com/30x/apid-core"
	"github.com/apigee-labs/transicator/common"
)

var token string
var downloadDataSnapshot, downloadBootSnapshot, changeFinished bool
var lastSequence string

func addHeaders(req *http.Request) {
	req.Header.Add("Authorization", "Bearer "+token)
	req.Header.Set("apid_instance_id", apidInfo.InstanceID)
	req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
	req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339))
}

func postPluginDataDelivery(e apid.Event) {

	if ede, ok := e.(apid.EventDeliveryEvent); ok {

		if ev, ok := ede.Event.(*common.ChangeList); ok {
			if lastSequence != ev.LastSequence {
				lastSequence = ev.LastSequence
				err := updateLastSequence(lastSequence)
				if err != nil {
					log.Panic("Unable to update Sequence in DB")
				}
			}
			changeFinished = true

		} else if _, ok := ede.Event.(*common.Snapshot); ok {
			if downloadBootSnapshot == false {
				downloadBootSnapshot = true
				log.Debug("Updated bootstrap SnapshotInfo")
			} else {
				downloadDataSnapshot = true
				log.Debug("Updated data SnapshotInfo")
			}
		}
	}
}

/*
 * Polls change agent for changes. In event of errors, uses a doubling
 * backoff from 200ms up to a max delay of the configPollInterval value.
 */
func updatePeriodicChanges() {

	var backOffFunc func()
	pollInterval := config.GetDuration(configPollInterval)
	for {
		err := pollChangeAgent()
		if err != nil {
			log.Debugf("Error connecting to changeserver: %v", err)
			if backOffFunc == nil {
				log.Error("creating backoff")
				backOffFunc = createBackOff(200*time.Millisecond, pollInterval)
			}
			backOffFunc()
		} else {
			log.Error("resetting backoff")
			backOffFunc = nil
		}
	}
}

/*
 * Long polls the change agent with a 45 second block. Parses the response from
 * change agent and raises an event.
 */
func pollChangeAgent() error {

	if downloadDataSnapshot != true {<commit_msg>Change back off detection to be time-based again<commit_after>package apidApigeeSync

import (
	"bytes"
	"encoding/json"
	"errors"
	"io/ioutil"
	"net/http"
	"net/url"
	"path"
	"time"

	"github.com/30x/apid-core"
	"github.com/apigee-labs/transicator/common"
)

var token string
var downloadDataSnapshot, downloadBootSnapshot, changeFinished bool
var lastSequence string

func addHeaders(req *http.Request) {
	req.Header.Add("Authorization", "Bearer "+token)
	req.Header.Set("apid_instance_id", apidInfo.InstanceID)
	req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
	req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339))
}

func postPluginDataDelivery(e apid.Event) {

	if ede, ok := e.(apid.EventDeliveryEvent); ok {

		if ev, ok := ede.Event.(*common.ChangeList); ok {
			if lastSequence != ev.LastSequence {
				lastSequence = ev.LastSequence
				err := updateLastSequence(lastSequence)
				if err != nil {
					log.Panic("Unable to update Sequence in DB")
				}
			}
			changeFinished = true

		} else if _, ok := ede.Event.(*common.Snapshot); ok {
			if downloadBootSnapshot == false {
				downloadBootSnapshot = true
				log.Debug("Updated bootstrap SnapshotInfo")
			} else {
				downloadDataSnapshot = true
				log.Debug("Updated data SnapshotInfo")
			}
		}
	}
}

/*
 * Polls change agent for changes. In event of errors, uses a doubling
 * backoff from 200ms up to a max delay of the configPollInterval value.
 */
func updatePeriodicChanges() {

	var backOffFunc func()
	pollInterval := config.GetDuration(configPollInterval)
	for {
		start := time.Now().Second()
		err := pollChangeAgent()
		end := time.Now().Second()
		if err != nil {
			log.Debugf("Error connecting to changeserver: %v", err)
		}
		if end-start <= 1 {
			if backOffFunc == nil {
				backOffFunc = createBackOff(200*time.Millisecond, pollInterval)
			}
			backOffFunc()
		} else {
			backOffFunc = nil
		}
	}
}

/*
 * Long polls the change agent with a 45 second block. Parses the response from
 * change agent and raises an event.
 */
func pollChangeAgent() error {

	if downloadDataSnapshot != true {
{'fraction_non_alphanumeric': 0.11923726768042481, 'fraction_numerical': 0.006999758629012792, 'mean_word_length': 3.869565217391304, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '175709', 'n_tokens_mistral': 1544, 'n_tokens_neox': 1361, 'n_words': 348}