fixed map

Author: rajatkb

Promise.go

@@ -173,10 +173,17 @@ func (obj *Promise) Catch(callback Callback) *Promise {
 //Finally ... a synchronous call to finally do something at the end of Promise Chain
 func (obj *Promise) Finally(callback func(interface{})) interface{} {
 	obj.valueWaitLock.Wait()
+	value := obj.value
+	vp, ok := value.(*Promise)
+	if ok {
+		vp.valueWaitLock.Wait()
+		value = vp.value
+	}
 	if callback != nil {
-		callback(obj.value)
+		callback(value)
 	}
-	return obj.value
+
+	return value
 }
 
 /**
@@ -200,12 +207,21 @@ func Map(promises []*Promise, cb Callback) []*Promise {
 }
 
 //Reduce ... asynchronous reducer , does not waits for all promise to be resolve , it launches reduce callback as soon as first result is available
-// func Reduce(promises []*Promise, reducer func(acc interface{}, value interface{}), start interface{}) *Promise {
-
-// 	return Create(func(resolve Callback, reject Callback) {
-
-// 	})
-// }
+// It will be used to process both errors and values. So reduces should account for that.
+func Reduce(promises []*Promise, reducer func(index int, acc interface{}, value interface{}) interface{}, acc interface{}) *Promise {
+	return Create(func(resolve Callback, reject Callback) {
+		promise := Resolve(acc)
+		count := 0
+		for asyncValue := range AsyncGenerator(promises) {
+			index := count
+			promise = promise.Then(func(acc interface{}) (interface{}, error) {
+				return reducer(index, acc, asyncValue), nil
+			})
+			count++
+		}
+		resolve(promise)
+	})
+}
 
 //// FUNCTIONAL DONE /////////////////////
 

README.md

@@ -257,6 +257,32 @@ for value := range Promise.AsyncGenerator(promises) {
 
 ```
 
+*My Map is all parallel , can I have a reduce, that pipelines with request delays ? 👩‍⚖️
+ Yes the Reduce provided here internally uses AsyncGenerator to grab data and quicly
+ launch promise task to to luanch reducer for each data. Effectively pipelining with
+ the promises passsed to it. 
+```golang
+
+promises := make([]*Promise.Promise, 3)
+for i := 0; i < len(promises); i++ {
+    index := i
+    promises[i] = Promise.Create(func(resolve Promise.Callback, reject Promise.Callback) {
+        time.Sleep(time.Duration(index)*time.Second + time.Duration(10))
+        resolve(index + 1)
+    })
+}
+
+value := Promise.Reduce(promises, func(index int, acci interface{}, valuei interface{}) interface{} {
+    acc, _ := acci.(int)
+    value, _ := valuei.(int)
+    acc = value + acc
+    return acc
+}, 0).Finally(nil)
+
+```
+
+
+
 
 ### Development
 

test/promise_test.go

@@ -461,3 +461,24 @@ func TestAsyncGenerator(t *testing.T) {
 		t.Errorf("Failed async generator")
 	}
 }
+
+func TestReducer(t *testing.T) {
+	promises := make([]*Promise.Promise, 3)
+	for i := 0; i < len(promises); i++ {
+		index := i
+		promises[i] = Promise.Create(func(resolve Promise.Callback, reject Promise.Callback) {
+			time.Sleep(time.Duration(index)*time.Second + time.Duration(10))
+			resolve(index + 1)
+		})
+	}
+
+	value := Promise.Reduce(promises, func(index int, acci interface{}, valuei interface{}) interface{} {
+		acc, _ := acci.(int)
+		value, _ := valuei.(int)
+		acc = value + acc
+		return acc
+	}, 0).Finally(nil)
+	if value != 6 {
+		t.Errorf("Reduce did not work")
+	}
+}