Merge pull request #3 from mimiro-io/feature/add-concurrent-processing-example

added sample code to process entities in parallel

Author: gra-moore

sample/sample_transform.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	ct "github.com/mimiro-io/common-http-transform"
 	egdm "github.com/mimiro-io/entity-graph-data-model"
+	"sync"
 )
 
 // EnrichConfig is a function that can be used to enrich the config by reading additional files or environment variables
@@ -30,7 +31,69 @@ type SampleTransform struct {
 func (dl *SampleTransform) Stop(_ context.Context) error { return nil }
 
 func (dl *SampleTransform) Transform(ec *egdm.EntityCollection) (*egdm.EntityCollection, ct.TransformError) {
-	return ec, nil
+	result, err := processEntitiesConcurrently(ec.Entities, 10, processEntity)
+	if err != nil {
+		return nil, ct.Err(err, 1)
+	}
+	return &egdm.EntityCollection{Entities: result}, nil
+}
+
+type Result struct {
+	Index  int
+	Entity *egdm.Entity
+	Error  error
+}
+
+func processEntitiesConcurrently(entities []*egdm.Entity, concurrency int, handler func(entity *egdm.Entity) (*egdm.Entity, error)) ([]*egdm.Entity, error) {
+	var wg sync.WaitGroup
+	results := make([]Result, len(entities))
+	resultsChan := make(chan Result, len(entities))
+
+	// Number of goroutines to use
+	itemsPerGoroutine := (len(entities) + concurrency - 1) / concurrency
+
+	for i := 0; i < concurrency; i++ {
+		start := i * itemsPerGoroutine
+		end := start + itemsPerGoroutine
+		if end > len(entities) {
+			end = len(entities)
+		}
+
+		wg.Add(1)
+		go func(start, end, index int) {
+			defer wg.Done()
+			for j := start; j < end; j++ {
+				result, err := handler(entities[j])
+				resultsChan <- Result{Index: j, Entity: result, Error: err}
+			}
+		}(start, end, i)
+	}
+
+	go func() {
+		wg.Wait()
+		close(resultsChan)
+	}()
+
+	for result := range resultsChan {
+		if result.Error != nil {
+			return nil, result.Error
+		}
+		results[result.Index] = result
+	}
+
+	// filter out non nil
+	finalResults := make([]*egdm.Entity, 0)
+	for _, result := range results {
+		if result.Entity != nil {
+			finalResults = append(finalResults, result.Entity)
+		}
+	}
+
+	return finalResults, nil
+}
+
+func processEntity(entity *egdm.Entity) (*egdm.Entity, error) {
+	return entity, nil
 }
 
 // NewSampleTransform is a factory function that creates a new instance of the sample transform

sample/sample_transform_test.go

@@ -55,6 +55,12 @@ func TestNewSampleDataLayer(t *testing.T) {
 				"props": {
 					"http://example.com/name": "John Smith"
 				}
+			},
+			{
+				"id": "http://example.com/2",
+				"props": {
+					"http://example.com/name": "James Shadwell"
+				}
 			}
 		]`