// We traverse all items in the given category with following way
// due to Yahoo Shopping API limitation(max 30 items at once, 100 page for the given condition)
// 1. get items in price ascending order
// 2. when we reach 100th page, set the last item's price to `minPrice` and fetch more 100 pages
// 3. when we get 0 items, it means we reached the end.
// Ideally, we want to refetch only updated items since we need to do full-reindex with the current approach.
// But currently we don't have a way to get item's updated time(API doesn't return it) and set `from` parameter for search
We traverse all items in the given category with following way
due to Yahoo Shopping API limitation(max 30 items at once, 100 page for the given condition)
1. get items in price ascending order
2. when we reach 100th page, set the last item's price to `minPrice` and fetch more 100 pages
3. when we get 0 items, it means we reached the end.
Ideally, we want to refetch only updated items since we need to do full-reindex with the current approach.
But currently we don't have a way to get item's updated time(API doesn't return it) and set `from` parameter for search
// We traverse all items in the given category with following way // due to Yahoo Shopping API limitation ( max 30 items at once 100 page for the given condition ) // 1 . get items in price ascending order // 2 . when we reach 100th page set the last item ' s price to ` minPrice ` and fetch more 100 pages // 3 . when we get 0 items it means we reached the end . // Ideally we want to refetch only updated items since we need to do full - reindex with the current approach . // But currently we don ' t have a way to get item ' s updated time ( API doesn ' t return it ) and set ` from ` parameter for search
func (w *categoryItemsFetcher) start(ctx context.Context) {
	rateLimiter := rate.NewLimiter(1, 1)
	go func() {
		for {
			w.pool <- w
			select {
			case <-ctx.Done():
				return
			case categoryID := <-w.categoryID:
				totalPublishedCount := 0
				cursor := w.yahooShoppingAPIClient.NewCategoryItemCursor(categoryID, item_fetcher.MinFetchItemPrice, item_fetcher.MaxFetchItemPrice)
				for {
					if err := rateLimiter.Wait(ctx); err != nil {
						w.logger.Error("rateLimiter.Wait failed", zap.Error(err))
					}
					res, err := cursor.Next(ctx)
					if err == yahoo_shopping.Done {
						w.logger.Info(fmt.Sprintf(
							"fetched all items in category %d", categoryID),
							zap.Int("categoryID", categoryID),
							zap.Int("total", totalPublishedCount),
						)
						break
					}
					if err != nil {
						w.logger.Error("cursor.Next failed",
							zap.Error(err),
							zap.Int("categoryID", categoryID),
							zap.Int("minPrice", cursor.CurMinPrice()),
							zap.Int("page", cursor.CurPage()),
						)
						break
					}
					items, err := mapYahooShoppingItemsToIndexItems(res.Hits, w.ysCategoryIDItemCategoryMap)
					if err != nil {
						w.logger.Error(
							"mapYahooShoppingItemsToIndexItems failed for some items",
							zap.Error(err),
							zap.Int("totalCount", len(res.Hits)),
							zap.Int("failedCount", len(res.Hits)-len(items)),
						)
					}
					wg := sync.WaitGroup{}
					var publishedCount int64
					for _, item := range items {
						if !item.IsIndexable() {
							continue
						}
						item := item
						wg.Add(1)
						go func() {
							defer wg.Done()
							itemJSON, err := json.Marshal(item)
							if err != nil {
								w.logger.Error(
									"json.Marshal item failed",
									zap.Error(err),
									zap.Any("item", item),
								)
								return
							}
							res := w.pubsubItemUpdateTopic.Publish(ctx, &pubsub.Message{
								Data:        itemJSON,
								OrderingKey: item_fetcher.ItemOrderingKey(item),
							})
							if _, err := res.Get(ctx); err != nil {
								w.logger.Error("publish item update failed",
									zap.Error(err),
									zap.String("itemId", item.ID),
								)
								return
							}
							atomic.AddInt64(&publishedCount, 1)
						}()
					}
					wg.Wait()
					totalPublishedCount += int(publishedCount)
					if totalPublishedCount%300 == 0 {
						w.logger.Info(fmt.Sprintf(
							"published %d items", totalPublishedCount),
							zap.Int("categoryID", categoryID),
							zap.Int("minPrice", cursor.CurMinPrice()),
							zap.Int("page", cursor.CurPage()),
						)
					}
				}
				w.wg.Done()
			}
		}
	}()
}
func ( w * categoryItemsFetcher ) start ( ctx context . Context ) { rateLimiter := rate . NewLimiter ( 1 , 1 ) go func ( ) { for { w . pool <- w select { case <- ctx . Done ( ) : return case categoryID := <- w . categoryID : totalPublishedCount := 0 cursor := w . yahooShoppingAPIClient . NewCategoryItemCursor ( categoryID , item_fetcher . MinFetchItemPrice , item_fetcher . MaxFetchItemPrice ) for { if err := rateLimiter . Wait ( ctx ) ; err != nil { w . logger . Error ( " " , zap . Error ( err ) ) } res , err := cursor . Next ( ctx ) if err == yahoo_shopping . Done { w . logger . Info ( fmt . Sprintf ( " " , categoryID ) , zap . Int ( " " , categoryID ) , zap . Int ( " " , totalPublishedCount ) , ) break } if err != nil { w . logger . Error ( " " , zap . Error ( err ) , zap . Int ( " " , categoryID ) , zap . Int ( " " , cursor . CurMinPrice ( ) ) , zap . Int ( " " , cursor . CurPage ( ) ) , ) break } items , err := mapYahooShoppingItemsToIndexItems ( res . Hits , w . ysCategoryIDItemCategoryMap ) if err != nil { w . logger . Error ( " " , zap . Error ( err ) , zap . Int ( " " , len ( res . Hits ) ) , zap . Int ( " " , len ( res . Hits ) - len ( items ) ) , ) } wg := sync . WaitGroup { } var publishedCount int64 for _ , item := range items { if ! item . IsIndexable ( ) { continue } item := item wg . Add ( 1 ) go func ( ) { defer wg . Done ( ) itemJSON , err := json . Marshal ( item ) if err != nil { w . logger . Error ( " " , zap . Error ( err ) , zap . Any ( " " , item ) , ) return } res := w . pubsubItemUpdateTopic . Publish ( ctx , & pubsub . Message { Data : itemJSON , OrderingKey : item_fetcher . ItemOrderingKey ( item ) , } ) if _ , err := res . Get ( ctx ) ; err != nil { w . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , item . ID ) , ) return } atomic . AddInt64 ( & publishedCount , 1 ) } ( ) } wg . Wait ( ) totalPublishedCount += int ( publishedCount ) if totalPublishedCount % 300 == 0 { w . logger . Info ( fmt . Sprintf ( " " , totalPublishedCount ) , zap . Int ( " " , categoryID ) , zap . Int ( " " , cursor . CurMinPrice ( ) ) , zap . Int ( " " , cursor . CurPage ( ) ) , ) } } w . wg . Done ( ) } } } ( ) }
We traverse all items in the given category with following way
due to Yahoo Shopping API limitation(max 30 items at once, 100 page for the given condition)
1. get items in price ascending order
2. when we reach 100th page, set the last item's price to `minPrice` and fetch more 100 pages
3. when we get 0 items, it means we reached the end.
We traverse all items in the given category with following way due to Yahoo Shopping API limitation ( max 30 items at once 100 page for the given condition ) 1 . get items in price ascending order 2 . when we reach 100th page set the last item ' s price to ` minPrice ` and fetch more 100 pages 3 . when we get 0 items it means we reached the end .