Refactor rate converter separating scheduler from converter logic to improve testability

currencies/converter_info.go

@@ -5,30 +5,23 @@ import "time"
 // ConverterInfo holds information about converter setup
 type ConverterInfo interface {
 	Source() string
-	FetchingInterval() time.Duration
 	LastUpdated() time.Time
 	Rates() *map[string]map[string]float64
 	AdditionalInfo() interface{}
 }
 
 type converterInfo struct {
-	source           string
-	fetchingInterval time.Duration
-	lastUpdated      time.Time
-	rates            *map[string]map[string]float64
-	additionalInfo   interface{}
+	source         string
+	lastUpdated    time.Time
+	rates          *map[string]map[string]float64
+	additionalInfo interface{}
 }
 
 // Source returns converter's URL source
 func (ci converterInfo) Source() string {
 	return ci.source
 }
 
-// FetchingInterval returns converter's fetching interval in nanoseconds
-func (ci converterInfo) FetchingInterval() time.Duration {
-	return ci.fetchingInterval
-}
-
 // LastUpdated returns converter's last updated time
 func (ci converterInfo) LastUpdated() time.Time {
 	return ci.lastUpdated

currencies/rate_converter.go

@@ -2,83 +2,43 @@ package currencies
 
 import (
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"net/http"
 	"sync/atomic"
 	"time"
 
 	"github.com/golang/glog"
+	"github.com/prebid/prebid-server/errortypes"
+	"github.com/prebid/prebid-server/util/timeutil"
 )
 
 // RateConverter holds the currencies conversion rates dictionary
 type RateConverter struct {
 	httpClient          httpClient
-	done                chan bool
-	updateNotifier      chan<- int
-	fetchingInterval    time.Duration
 	staleRatesThreshold time.Duration
 	syncSourceURL       string
 	rates               atomic.Value // Should only hold Rates struct
 	lastUpdated         atomic.Value // Should only hold time.Time
 	constantRates       Conversions
+	time                timeutil.Time
 }
 
 // NewRateConverter returns a new RateConverter
 func NewRateConverter(
 	httpClient httpClient,
 	syncSourceURL string,
-	fetchingInterval time.Duration,
 	staleRatesThreshold time.Duration,
 ) *RateConverter {
-	return NewRateConverterWithNotifier(
-		httpClient,
-		syncSourceURL,
-		fetchingInterval,
-		staleRatesThreshold,
-		nil, // no notifier channel specified, won't send any notifications
-	)
-}
-
-// NewRateConverterDefault returns a RateConverter with default values.
-// By default there will be no currencies conversions done.
-// `currencies.ConstantRate` will be used.
-func NewRateConverterDefault() *RateConverter {
-	return NewRateConverter(&http.Client{}, "", time.Duration(0), time.Duration(0))
-}
-
-// NewRateConverterWithNotifier returns a new RateConverter
-// it allow to pass an update chan in which the number of ticks will be passed after each tick
-// allowing clients to listen on updates
-// Do not use this method
-func NewRateConverterWithNotifier(
-	httpClient httpClient,
-	syncSourceURL string,
-	fetchingInterval time.Duration,
-	staleRatesThreshold time.Duration,
-	updateNotifier chan<- int,
-) *RateConverter {
-	rc := &RateConverter{
+	return &RateConverter{
 		httpClient:          httpClient,
-		done:                make(chan bool),
-		updateNotifier:      updateNotifier,
-		fetchingInterval:    fetchingInterval,
 		staleRatesThreshold: staleRatesThreshold,
 		syncSourceURL:       syncSourceURL,
 		rates:               atomic.Value{},
 		lastUpdated:         atomic.Value{},
 		constantRates:       NewConstantRates(),
+		time:                &timeutil.RealTime{},
 	}
-
-	// In case host do not want to support currency lookup
-	// we just stop here and do nothing
-	if rc.fetchingInterval == time.Duration(0) {
-		return rc
-	}
-
-	rc.Update()                   // Make sure to populate data before to return the converter
-	go rc.startPeriodicFetching() // Start periodic ticking
-
-	return rc
 }
 
 // fetch allows to retrieve the currencies rates from the syncSourceURL provided
@@ -93,6 +53,11 @@ func (rc *RateConverter) fetch() (*Rates, error) {
 		return nil, err
 	}
 
+	if response.StatusCode >= 400 {
+		message := fmt.Sprintf("The currency rates request failed with status code %d", response.StatusCode)
+		return nil, &errortypes.BadServerResponse{Message: message}
+	}
+
 	defer response.Body.Close()
 
 	bytesJSON, err := ioutil.ReadAll(response.Body)
@@ -110,14 +75,14 @@ func (rc *RateConverter) fetch() (*Rates, error) {
 }
 
 // Update updates the internal currencies rates from remote sources
-func (rc *RateConverter) Update() error {
+func (rc *RateConverter) update() error {
 	rates, err := rc.fetch()
 	if err == nil {
 		rc.rates.Store(rates)
-		rc.lastUpdated.Store(time.Now())
+		rc.lastUpdated.Store(rc.time.Now())
 	} else {
-		if rc.CheckStaleRates() {
-			rc.ClearRates()
+		if rc.checkStaleRates() {
+			rc.clearRates()
 			glog.Errorf("Error updating conversion rates, falling back to constant rates: %v", err)
 		} else {
 			glog.Errorf("Error updating conversion rates: %v", err)
@@ -127,37 +92,8 @@ func (rc *RateConverter) Update() error {
 	return err
 }
 
-// startPeriodicFetching starts the periodic fetching at the given interval
-// triggers a first fetch when called before the first tick happen in order to initialize currencies rates map
-// returns a chan in which the number of data updates everytime a new update was done
-func (rc *RateConverter) startPeriodicFetching() {
-
-	ticker := time.NewTicker(rc.fetchingInterval)
-	updatesTicksCount := 0
-
-	for {
-		select {
-		case <-ticker.C:
-			// Retries are handled by clients directly.
-			rc.Update()
-			updatesTicksCount++
-			if rc.updateNotifier != nil {
-				rc.updateNotifier <- updatesTicksCount
-			}
-		case <-rc.done:
-			if ticker != nil {
-				ticker.Stop()
-				ticker = nil
-			}
-			return
-		}
-	}
-}
-
-// StopPeriodicFetching stops the periodic fetching while keeping the latest currencies rates map
-func (rc *RateConverter) StopPeriodicFetching() {
-	rc.done <- true
-	close(rc.done)
+func (rc *RateConverter) Run() error {
+	return rc.update()
 }
 
 // LastUpdated returns time when currencies rates were updated
@@ -178,18 +114,19 @@ func (rc *RateConverter) Rates() Conversions {
 	return rc.constantRates
 }
 
-// ClearRates sets the rates to nil
-func (rc *RateConverter) ClearRates() {
+// clearRates sets the rates to nil
+func (rc *RateConverter) clearRates() {
 	// atomic.Value field rates must be of type *Rates so we cast nil to that type
 	rc.rates.Store((*Rates)(nil))
 }
 
-// CheckStaleRates checks if loaded third party conversion rates are stale
-func (rc *RateConverter) CheckStaleRates() bool {
+// checkStaleRates checks if loaded third party conversion rates are stale
+func (rc *RateConverter) checkStaleRates() bool {
 	if rc.staleRatesThreshold <= 0 {
 		return false
 	}
-	currentTime := time.Now().UTC()
+
+	currentTime := rc.time.Now().UTC()
 	if lastUpdated := rc.lastUpdated.Load(); lastUpdated != nil {
 		delta := currentTime.Sub(lastUpdated.(time.Time).UTC())
 		if delta.Seconds() > rc.staleRatesThreshold.Seconds() {
@@ -202,14 +139,11 @@ func (rc *RateConverter) CheckStaleRates() bool {
 // GetInfo returns setup information about the converter
 func (rc *RateConverter) GetInfo() ConverterInfo {
 	var rates *map[string]map[string]float64
-	if rc.Rates() != nil {
-		rates = rc.Rates().GetRates()
-	}
+	rates = rc.Rates().GetRates()
 	return converterInfo{
-		source:           rc.syncSourceURL,
-		fetchingInterval: rc.fetchingInterval,
-		lastUpdated:      rc.LastUpdated(),
-		rates:            rates,
+		source:      rc.syncSourceURL,
+		lastUpdated: rc.LastUpdated(),
+		rates:       rates,
 	}
 }