<commit_before>package capmetro

import (
	"log"
	"os"
	"time"

	r "github.com/scascketta/CapMetrics/Godeps/_workspace/src/github.com/dancannon/gorethink"
)

const (
	NormalDuration   = 30 * time.Second // Duration to wait between fetching locations when at least one route is active
	ExtendedDuration = 10 * time.Minute // Duration to wait between fetching locations when all routes are inactive
)

var (
	dlog = log.New(os.Stdout, "[DBG] ", log.LstdFlags|log.Lshortfile)
	elog = log.New(os.Stderr, "[ERR] ", log.LstdFlags|log.Lshortfile)
)

// FetchHistory contains the history of vehicle location fetches, including when a vehicle was last updated,
// and how many responses were "stale" for each route.
type FetchHistory struct {
	LastUpdated    map[string]time.Time
	StaleResponses map[string]int
}

func NewFetchHistory() *FetchHistory {
	return &FetchHistory{make(map[string]time.Time), make(map[string]int)}
}

// filterUpdatedVehicles returns the VehicleLocation structs whose timestamp has
// changed since the last update
func filterUpdatedVehicles(vehicles []VehicleLocation, fh *FetchHistory) []VehicleLocation {
	updated := []VehicleLocation{}
	for _, v := range vehicles {
		updateTime, _ := fh.LastUpdated[v.VehicleID]
		fh.LastUpdated[v.VehicleID] = v.Time
		threshold := time.Now().Add(time.Minute)
		// reject location updates if timestamp hasn't changed or is more than 1 minute in the future
		if !updateTime.Equal(v.Time) && v.Time.Before(threshold) {
			updated = append(updated, v)
		}
	}
	return updated
}

func prepRoute(route string, fh *FetchHistory) {
	if _, ok := fh.StaleResponses[route]; !ok {
		fh.StaleResponses[route] = 0
	}
}

// LogVehicleLocations calls setupConn for a *gorethink.Session to pass to logVehicleLocations and closes it afterwards.
func LogVehicleLocations(setupConn func() *r.Session, fh *FetchHistory) func() error {
	return func() error {
		session := setupConn()
		err := logVehicleLocations(session, fh)
		session.Close()
		return err
	}
}

// logVehicleLocations fetches vehicle locations from CapMetro and inserts new
// locations into the database. It also tracks stale responses to determine when
//  to sleep.
func logVehicleLocations(session *r.Session, fh *FetchHistory) error {
	locationsByRoute, err := FetchVehicles()
	if err != nil {
		return err
	}

	for route, rl := range locationsByRoute {
		prepRoute(route, fh)

		updated := filterUpdatedVehicles(rl.Locations, fh)
		if len(updated) == 0 {
			// increment retry count if fetch just before was also stale
			// only subsequent stale responses matter when determining how long to sleep
			fh.StaleResponses[route]++
			dlog.Printf("No vehicles in response for route: %s.", route)
			continue
		} else {
			fh.StaleResponses[route] = 0
		}

		for _, v := range updated {
			dlog.Printf("Vehicle %s updated at %s\n", v.VehicleID, v.Time.Format("2006-01-02T15:04:05-07:00"))
		}

		if len(updated) > 0 {
			_, err = r.Table("vehicle_position").Insert(r.Expr(updated)).Run(session)
			if err != nil {
				return err
			}
			dlog.Printf("Log %d vehicles, route %s.\n", len(updated), route)
		} else {
			dlog.Printf("No new vehicle positions to record for route %s.\n", route)
		}
	}
	return nil
}

// Check if the the routes are inactive
// There must have been MAX_RETRIES previous attempts to fetch data,
// and all attempts must have failed
func routesAreSleeping(maxRetries int, fh *FetchHistory) bool {
	dlog.Println("staleResponses:", fh.StaleResponses)
	for _, retries := range fh.StaleResponses {
		if retries < maxRetries {
			return false
		}
	}
	return true
}

// UpdateInterval changes the interval between fetches if MAX_RETRIES responses have been stale for every route
func UpdateInterval(maxRetries int, fh *FetchHistory) func() (bool, time.Duration) {
	return func() (bool, time.Duration) {
		if routesAreSleeping(maxRetries, fh) {
			for k := range fh.StaleResponses {
				fh.StaleResponses[k] = 0
			}
			return true, ExtendedDuration
		} else {
			return false, NormalDuration
		}
	}
}<commit_msg>Revert "Revert "Make LogVehicleLocations concurrent""<commit_after>package capmetro

import (
	"log"
	"os"
	"time"

	r "github.com/scascketta/CapMetrics/Godeps/_workspace/src/github.com/dancannon/gorethink"
)

const (
	NormalDuration   = 30 * time.Second // Duration to wait between fetching locations when at least one route is active
	ExtendedDuration = 10 * time.Minute // Duration to wait between fetching locations when all routes are inactive
)

var (
	dlog = log.New(os.Stdout, "[DBG] ", log.LstdFlags|log.Lshortfile)
	elog = log.New(os.Stderr, "[ERR] ", log.LstdFlags|log.Lshortfile)
)

// FetchHistory contains the history of vehicle location fetches, including when a vehicle was last updated,
// and how many responses were "stale" for each route.
type FetchHistory struct {
	LastUpdated    map[string]time.Time
	StaleResponses map[string]int
}

func NewFetchHistory() *FetchHistory {
	return &FetchHistory{make(map[string]time.Time), make(map[string]int)}
}

// filterUpdatedVehicles returns the VehicleLocation structs whose timestamp has
// changed since the last update
func filterUpdatedVehicles(vehicles []VehicleLocation, fh *FetchHistory) []VehicleLocation {
	updated := []VehicleLocation{}
	for _, v := range vehicles {
		updateTime, _ := fh.LastUpdated[v.VehicleID]
		fh.LastUpdated[v.VehicleID] = v.Time
		threshold := time.Now().Add(time.Minute)
		// reject location updates if timestamp hasn't changed or is more than 1 minute in the future
		if !updateTime.Equal(v.Time) && v.Time.Before(threshold) {
			updated = append(updated, v)
			dlog.Printf("Vehicle %s updated at %s\n", v.VehicleID, v.Time.Format("2006-01-02T15:04:05-07:00"))
		}
	}
	return updated
}

func prepRoute(route string, fh *FetchHistory) {
	if _, ok := fh.StaleResponses[route]; !ok {
		fh.StaleResponses[route] = 0
	}
}

// LogVehicleLocations calls setupConn for a *gorethink.Session to pass to logVehicleLocations and closes it afterwards.
func LogVehicleLocations(setupConn func() *r.Session, fh *FetchHistory) func() error {
	return func() error {
		session := setupConn()
		err := logVehicleLocations(session, fh)
		session.Close()
		return err
	}
}

// logVehicleLocations fetches vehicle locations from CapMetro and inserts new
// locations into the database. It also tracks stale responses to determine when
//  to sleep.
func logVehicleLocations(session *r.Session, fh *FetchHistory) error {
	locationsByRoute, err := FetchVehicles()
	if err != nil {
		return err
	}

	errChan := make(chan error)
	doneChan := make(chan struct{})
	remaining := len(locationsByRoute)
	for route, rl := range locationsByRoute {
		go func(route string, rl RouteLocations, errChan chan error, doneChan chan struct{}) {
			prepRoute(route, fh)
			updated := filterUpdatedVehicles(rl.Locations, fh)
			if len(updated) == 0 {
				fh.StaleResponses[route]++
				dlog.Printf("No vehicles in response for route: %s.", route)
			} else {
				fh.StaleResponses[route] = 0
				_, err = r.Table("vehicle_position").Insert(r.Expr(updated)).Run(session)
				if err != nil {
					errChan <- err
				} else {
					dlog.Printf("Log %d vehicles, route %s.\n", len(updated), route)
					doneChan <- struct{}{}
				}

			}
		}(route, rl, errChan, doneChan)
	}

	for {
		select {
		case _ = <-doneChan:
			remaining--
		case err := <-errChan:
			return err
		}

		if remaining == 0 {
			break
		}
	}
	return nil
}

// Check if the the routes are inactive
// There must have been MAX_RETRIES previous attempts to fetch data,
// and all attempts must have failed
func routesAreSleeping(maxRetries int, fh *FetchHistory) bool {
	dlog.Println("staleResponses:", fh.StaleResponses)
	for _, retries := range fh.StaleResponses {
		if retries < maxRetries {
			return false
		}
	}
	return true
}

// UpdateInterval changes the interval between fetches if MAX_RETRIES responses have been stale for every route
func UpdateInterval(maxRetries int, fh *FetchHistory) func() (bool, time.Duration) {
	return func() (bool, time.Duration) {
		if routesAreSleeping(maxRetries, fh) {
			for k := range fh.StaleResponses {
				fh.StaleResponses[k] = 0
			}
			return true, ExtendedDuration
		} else {
			return false, NormalDuration
		}
	}
}
{'fraction_non_alphanumeric': 0.11043541717343712, 'fraction_numerical': 0.006810994891753831, 'mean_word_length': 4.107453416149069, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5866217', 'n_tokens_mistral': 2941, 'n_tokens_neox': 2697, 'n_words': 792}