fho_forward/client/listener.go

package client

import (
	"io"
	"log"
	"time"
)

func listen(c *FhoClient, object string, lastUpdatedTime time.Time, fn func(*[]map[string]interface{}) error) error {
	snap, err := c.QuerySnapshot(object, lastUpdatedTime)
	if err != nil {
		log.Println("listen error:", err.Error())
		return err
	}
	for {
		data, err := snap.Next()
		if err == io.EOF {
			log.Println("Error EOF")
			break
		}
		if err != nil {
			log.Println("Error:", err.Error())
			break
		}
		err = fn(data)
		if err != nil {
			return err
		}
	}
	return nil
}

func ListenOnProducts(c *FhoClient, lastUpdatedTime time.Time, fn func([]*Product) error) error {
	listen(c, "products", lastUpdatedTime, func(data *[]map[string]interface{}) error {
		var products []*Product
		for _, rec := range *data {
			_product := ProductFromMap(rec)
			products = append(products, _product)
		}
		return fn(products)
	})
	return nil
}

func ListenOnPrice(c *FhoClient, lastUpdatedTime time.Time, fn func([]*PriceSchedule) error) error {
	err := listen(c, "station_price_updates", lastUpdatedTime, func(data *[]map[string]interface{}) error {
		var prices []*PriceSchedule
		pricesMap := make(map[string]*PriceSchedule)
		for _, rec := range *data {

			_price, err := PriceScheduleFromMap(rec)
			if err != nil {
				return err
			}
			if _, ok := pricesMap[_price.ID]; ok {
				_price = pricesMap[_price.ID]
			} else {
				pricesMap[_price.ID] = _price
				prices = append(prices, _price)
			}
			_product := PriceScheduleProductFromMap(rec)
			_price.PriceScheduleProducts = append(_price.PriceScheduleProducts, _product)
		}
		return fn(prices)
	})
	return err
}