Added testing and examples for products lister

Author: breathbath

examples/products/main.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/erply/api-go-wrapper/pkg/api"
 	"github.com/erply/api-go-wrapper/pkg/api/auth"
+	sharedCommon "github.com/erply/api-go-wrapper/pkg/api/common"
 	"github.com/erply/api-go-wrapper/pkg/api/products"
 	"net/http"
 	"time"
@@ -32,7 +33,13 @@ func main() {
 		panic(err)
 	}
 
-	fmt.Printf("%+v\n", prods)
+	fmt.Printf("GetProductsBulk:\n%+v\n", prods)
+
+	prods, err = GetProductsInParallel(apiClient)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Printf("GetProductsInParallel:\n%+v\n", prods)
 }
 
 func GetProductsBulk(cl *api.Client) (prods []products.Product, err error) {
@@ -65,3 +72,42 @@ func GetProductsBulk(cl *api.Client) (prods []products.Product, err error) {
 
 	return
 }
+
+func GetProductsInParallel(cl *api.Client) ([]products.Product, error) {
+	productsDataProvider := products.NewListingDataProvider(cl.ProductManager)
+
+	lister := sharedCommon.NewLister(
+		sharedCommon.ListingSettings{
+			MaxRequestsCountPerSecond: 5,
+			StreamBufferLength:        10,
+			MaxItemsPerRequest:        10,
+			MaxFetchersCount:          2,
+		},
+		productsDataProvider,
+		func(sleepTime time.Duration) {
+			time.Sleep(sleepTime)
+		},
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second * 5)
+	defer cancel()
+
+	prodsChan := lister.Get(ctx, map[string]interface{}{})
+
+	prods := make([]products.Product, 0)
+	var err error
+	doneChan := make(chan struct{}, 1)
+	go func() {
+		defer close(doneChan)
+		for prod := range prodsChan {
+			if prod.Err != nil {
+				err = prod.Err
+				return
+			}
+			prods = append(prods, prod.Payload.(products.Product))
+		}
+	}()
+
+	<-doneChan
+	return prods, err
+}

pkg/api/common/listing.go

@@ -26,7 +26,6 @@ type ItemsStream chan Item
 type Item struct {
 	Err           error
 	TotalCount    int
-	ProgressCount int
 	Payload       interface{}
 }
 
@@ -35,8 +34,8 @@ func setListingSettingsDefaults(settingsFromInput ListingSettings) ListingSettin
 		settingsFromInput.MaxRequestsCountPerSecond = DefaultMaxRequestsCountPerSecond
 	}
 
-	if settingsFromInput.MaxItemsPerRequest == 0 || settingsFromInput.MaxItemsPerRequest > MaxCountPerBulkRequestItem {
-		settingsFromInput.MaxItemsPerRequest = MaxCountPerBulkRequestItem
+	if settingsFromInput.MaxItemsPerRequest == 0 || settingsFromInput.MaxItemsPerRequest > MaxCountPerBulkRequestItem * MaxCountPerBulkRequestItem {
+		settingsFromInput.MaxItemsPerRequest = MaxCountPerBulkRequestItem * MaxCountPerBulkRequestItem
 	}
 
 	if settingsFromInput.MaxFetchersCount == 0 {
@@ -131,25 +130,21 @@ func (p *Lister) getCursors(ctx context.Context, totalCount int) chan []Cursor {
 			p.listingSettings.MaxItemsPerRequest = MaxCountPerBulkRequestItem*MaxBulkRequestsCount
 		}
 
-		for ; leftCount > 0; { //leftCount 1000, p.listingSettings.MaxItemsPerRequest 100
+		for ; leftCount > 0; {
 			countToFetchForBulkRequest := leftCount
 			if leftCount > p.listingSettings.MaxItemsPerRequest {
 				countToFetchForBulkRequest = p.listingSettings.MaxItemsPerRequest
 			}
 
-			//countToFetchForBulkRequest 100
-
 			bulkItemsCount := CeilDivisionInt(countToFetchForBulkRequest, MaxCountPerBulkRequestItem)
 			if bulkItemsCount > MaxBulkRequestsCount {
 				bulkItemsCount = MaxBulkRequestsCount
 			}
-			//bulkItemsCount 1
 
 			limit := CeilDivisionInt(p.listingSettings.MaxItemsPerRequest, bulkItemsCount)
 			if limit > MaxCountPerBulkRequestItem {
 				limit = MaxCountPerBulkRequestItem
 			}
-			//limit 100
 
 			cursorsForBulkRequest := make([]Cursor, 0, bulkItemsCount)
 			for i := 0; i < bulkItemsCount; i++ {
@@ -244,8 +239,3 @@ func (p *Lister) mergeChannels(ctx context.Context, childChans ...ItemsStream) I
 func CeilDivisionInt(x, y int) int {
 	return int(math.Ceil(float64(x) / float64(y)))
 }
-
-func AddDefaultPaginationOption(filters map[string]interface{}, limit, page int) {
-	filters["recordsOnPage"] = limit
-	filters["pageNo"] = page
-}