Merge pull request #10 from alob-mtc/event-loop

added an event loop and reduced unnecessary goroutine creation

.gitignore

@@ -1 +1,2 @@
-.idea/
+.idea/
+tmp/

eventloop.go

@@ -13,13 +13,38 @@ var (
 	GlobalEventLoop *eventLoop
 )
 
-func Init() {
+func InitGlobalEventLoop() Future {
 	once.Do(func() {
+		queue := []func() bool{}
+		var lock sync.Mutex
+		// event loop
+		go func() {
+			for work := range eventBus {
+				lock.Lock()
+				queue = append(queue, *work)
+				lock.Unlock()
+			}
+		}()
+
+		go func() {
+			for {
+				n := len(queue)
+				retry := []func() bool{}
+				for i := 0; i < n; i++ {
+					currentWork := queue[i]
+
+					if done := currentWork(); !done {
+						retry = append(retry, currentWork)
+					}
+				}
+				lock.Lock()
+				queue = queue[n:]
+				queue = append(queue, retry...)
+				lock.Unlock()
+			}
+		}()
 		GlobalEventLoop = &eventLoop{promiseQueue: make([]*Promise, 0)}
 	})
-}
-
-func GetGlobalEventLoop() Future {
 	return GlobalEventLoop
 }
 
@@ -39,6 +64,7 @@ func (e *eventLoop) Await(currentP *Promise) (interface{}, error) {
 	defer currentP.Done()
 	currentP.RegisterHandler()
 	select {
+
 	case err := <-currentP.errChan:
 		return nil, err
 	case rev := <-currentP.rev:
@@ -108,6 +134,7 @@ outer:
 				<-p.done
 			}
 			if currentN := int(atomic.LoadInt64(&e.size)); i == 0 && !(currentN > n) {
+				close(eventBus)
 				break outer
 			}
 		}

promise.go

@@ -5,13 +5,15 @@ import (
 	"sync/atomic"
 )
 
+var eventBus = make(chan *func() bool)
+
 type Promise struct {
-	id      uint64
-	handler bool
-	rev     <-chan interface{}
-	errChan chan error
-	err     bool
-	done    chan struct{}
+	id       uint64
+	handler  bool
+	rev      <-chan interface{}
+	errChan  chan error
+	doneFlag bool
+	done     chan struct{}
 }
 
 func (e *eventLoop) newPromise(rev <-chan interface{}, errChan chan error) *Promise {
@@ -22,7 +24,7 @@ func (e *eventLoop) newPromise(rev <-chan interface{}, errChan chan error) *Prom
 	return currentP
 }
 
-func (p *Promise) Done() {
+func (p *Promise)  Done() {
 	close(p.done)
 }
 
@@ -32,51 +34,52 @@ func (p *Promise) RegisterHandler() {
 
 func (p *Promise) Then(fn func(interface{})) *Promise {
 	p.RegisterHandler()
-	go func() {
-		for {
-			select {
-			default:
-				if p.err {
-					return
-				}
-			case val := <-p.rev:
-				func() {
-					defer func() {
-						if r := recover(); r != nil {
-							switch x := r.(type) {
-							case error:
-								p.errChan <- x
-							default:
-								p.errChan <- fmt.Errorf("%v", x)
-							}
-						} else {
-							p.err = true
-							p.Done()
+	work := func() bool {
+		select {
+		default:
+			if p.doneFlag {
+				return true
+			}
+			return false
+		case val := <-p.rev:
+			go func() {
+				defer func() {
+					if r := recover(); r != nil {
+						switch x := r.(type) {
+						case error:
+							p.errChan <- x
+						default:
+							p.errChan <- fmt.Errorf("%v", x)
 						}
-					}()
-					fn(val)
+					} else {
+						p.doneFlag = true
+						p.Done()
+					}
 				}()
-				return
-			}
+				fn(val)
+			}()
+			return true
 		}
-	}()
+	}
+	eventBus <- &work
 	return p
 }
 
 func (p *Promise) Catch(fn func(err error)) {
 	p.RegisterHandler()
-	go func() {
-		for {
-			select {
-			default:
-				if p.err {
-					return
-				}
-			case err := <-p.errChan:
-				fn(err)
-				p.err = true
-				p.Done()
+	work := func() bool {
+		select {
+		default:
+			if p.doneFlag {
+				return true
 			}
+			return false
+		case err := <-p.errChan:
+			p.doneFlag = true
+			go fn(err)
+			p.Done()
+			return true
 		}
-	}()
+	}
+	eventBus <- &work
 }