GoCollaborate ver:0.4.2

Author: HastingsYoung

CHANGELOG.md

@@ -2,6 +2,10 @@
 ## Updates
 **(Please note that no downward compability will be guaranteed before the formal release 1.0.0 )**
 ### 0.4.x
+#### 0.4.2
+- Update handler API -> Add `Background` as Job loader
+- Support asynchronously loading source data to Job
+- Update examples and documentation
 #### 0.4.1
 - Refactor handler code
 - Clean up code, remove unnecessary method body

Readme.md

@@ -106,17 +106,27 @@ import (
 	"github.com/GoCollaborate/src/artifacts/task"
 	"github.com/GoCollaborate/src/wrappers/taskHelper"
 	"net/http"
+	"time"
 )
 
-func ExampleJobHandler(w http.ResponseWriter, r *http.Request) *task.Job {
+func ExampleJobHandler(w http.ResponseWriter, r *http.Request, bg *task.Background) {
 	job := task.MakeJob()
-	job.Tasks(&task.Task{task.SHORT,
-		task.BASE, "exampleFunc",
-		task.Collection{1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4},
-		task.Collection{0},
-		task.NewTaskContext(struct{}{}), 0})
+
+	job.Tasks(
+		&task.Task{
+			task.SHORT,
+			task.BASE,
+			"exampleFunc",
+			task.Collection{1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4},
+			task.Collection{0},
+			task.NewTaskContext(struct{}{}),
+			0,
+		},
+	)
+
 	job.Stacks("core.ExampleTask.Mapper", "core.ExampleTask.Reducer")
-	return job
+
+	bg.Mount(job)
 }
 
 func ExampleFunc(source *task.Collection,
@@ -154,7 +164,6 @@ func (r *SimpleReducer) Reduce(maps map[int]*task.Task) (map[int]*task.Task, err
 	return maps, nil
 }
 
-
 ```
 ### Run
 Here we create the entry file and a simple implementation of map-reduce interface, and next we will run with std arguments:

artifacts/task/background.go

@@ -0,0 +1,22 @@
+package task
+
+type Background chan *Job
+
+func NewBackground() *Background {
+	bg := make(Background)
+	return &bg
+}
+
+func (bg *Background) Done() *Job {
+	return <-*bg
+}
+
+func (bg *Background) Mount(job *Job) {
+	go func() {
+		*bg <- job
+	}()
+}
+
+func (bg *Background) Close() {
+	close(*bg)
+}

artifacts/task/task.go

@@ -55,6 +55,10 @@ func (t *taskPriority) GetPriority() taskPriority {
 	return *t
 }
 
+func NewCollection() *Collection {
+	return &Collection{}
+}
+
 func (cg *Collection) Append(cs ...interface{}) *Collection {
 	*cg = append(*cg, cs...)
 	return cg

collaborate.go

@@ -54,9 +54,9 @@ func Set(key string, val ...interface{}) interface{} {
 		fs := store.GetInstance()
 
 		methods := val[0].([]string)
-		handlers := make([]func(w http.ResponseWriter, r *http.Request) *task.Job, len(val)-1)
+		handlers := make([]func(w http.ResponseWriter, r *http.Request, bg *task.Background), len(val)-1)
 		for i, v := range val[1:] {
-			handlers[i] = v.(func(w http.ResponseWriter, r *http.Request) *task.Job)
+			handlers[i] = v.(func(w http.ResponseWriter, r *http.Request, bg *task.Background))
 		}
 
 		// register jobs
@@ -65,9 +65,9 @@ func Set(key string, val ...interface{}) interface{} {
 		fs := store.GetInstance()
 
 		methods := val[0].([]string)
-		handlers := make([]func(w http.ResponseWriter, r *http.Request) *task.Job, len(val)-1)
+		handlers := make([]func(w http.ResponseWriter, r *http.Request, bg *task.Background), len(val)-1)
 		for i, v := range val[1:] {
-			handlers[i] = v.(func(w http.ResponseWriter, r *http.Request) *task.Job)
+			handlers[i] = v.(func(w http.ResponseWriter, r *http.Request, bg *task.Background))
 		}
 
 		// register jobs

collaborator/collaborator.go

@@ -150,7 +150,14 @@ func (clbt *Collaborator) Catchup() {
 
 			from := c.Local.GetFullExposureCard()
 			to := e
-			var in *message.CardMessage = message.NewCardMessageWithOptions(c.GetCluster(), &from, to, dgst.GetCards(), dgst.GetTimeStamp(), message.CardMessage_SYNC)
+			var in *message.CardMessage = message.NewCardMessageWithOptions(
+				c.GetCluster(),
+				&from,
+				to,
+				dgst.GetCards(),
+				dgst.GetTimeStamp(),
+				message.CardMessage_SYNC,
+			)
 			var out *message.CardMessage = message.NewCardMessage()
 			var out2 *message.CardMessage = message.NewCardMessage()
 			// first exchange
@@ -385,23 +392,30 @@ func (clbt *Collaborator) HandleLocal(router *mux.Router, jobFunc *store.JobFunc
 
 	var (
 		f = func(w http.ResponseWriter, r *http.Request) {
-			job := jobFunc.F(w, r)
 			var (
+				bg      = task.NewBackground()
 				counter = 0
 			)
-			for s := job.Front(); s != nil; s = s.Next() {
-				exes, err := job.Exes(counter)
-				if err != nil {
-					logger.GetLoggerInstance().LogError(err)
-					break
-				}
-				err = clbt.LocalDistribute(&s.TaskSet, exes)
-				if err != nil {
-					logger.GetLoggerInstance().LogError(err)
-					break
+
+			jobFunc.F(w, r, bg)
+
+			go func() {
+				job := bg.Done()
+				defer bg.Close()
+				for s := job.Front(); s != nil; s = s.Next() {
+					exes, err := job.Exes(counter)
+					if err != nil {
+						logger.GetLoggerInstance().LogError(err)
+						break
+					}
+					err = clbt.LocalDistribute(&s.TaskSet, exes)
+					if err != nil {
+						logger.GetLoggerInstance().LogError(err)
+						break
+					}
+					counter++
 				}
-				counter++
-			}
+			}()
 		}
 		fs = store.GetInstance()
 	)
@@ -423,24 +437,31 @@ func (clbt *Collaborator) HandleShared(router *mux.Router, jobFunc *store.JobFun
 
 	var (
 		f = func(w http.ResponseWriter, r *http.Request) {
-			job := jobFunc.F(w, r)
 			var (
+				bg      = task.NewBackground()
 				counter = 0
 			)
-			for s := job.Front(); s != nil; s = s.Next() {
-				exes, err := job.Exes(counter)
-				if err != nil {
-					logger.GetLoggerInstance().LogError(err)
-					break
-				}
 
-				err = clbt.SharedDistribute(&s.TaskSet, exes)
-				if err != nil {
-					logger.GetLoggerInstance().LogError(err)
-					break
+			jobFunc.F(w, r, bg)
+
+			go func() {
+				job := bg.Done()
+				defer bg.Close()
+				for s := job.Front(); s != nil; s = s.Next() {
+					exes, err := job.Exes(counter)
+					if err != nil {
+						logger.GetLoggerInstance().LogError(err)
+						break
+					}
+
+					err = clbt.SharedDistribute(&s.TaskSet, exes)
+					if err != nil {
+						logger.GetLoggerInstance().LogError(err)
+						break
+					}
+					counter++
 				}
-				counter++
-			}
+			}()
 		}
 		fs = store.GetInstance()
 	)

store/store.go

@@ -75,7 +75,7 @@ type FS struct {
 }
 
 type JobFunc struct {
-	F         func(w http.ResponseWriter, r *http.Request) *task.Job
+	F         func(w http.ResponseWriter, r *http.Request, bg *task.Background)
 	Methods   []string
 	Signature string
 }
@@ -188,14 +188,14 @@ func (fs *FS) GetShared(key string) (*JobFunc, error) {
 	return new(JobFunc), constants.ErrJobNotExist
 }
 
-func (fs *FS) AddLocal(methods []string, jobs ...func(w http.ResponseWriter, r *http.Request) *task.Job) {
+func (fs *FS) AddLocal(methods []string, jobs ...func(w http.ResponseWriter, r *http.Request, bg *task.Background)) {
 	for _, f := range jobs {
 		signature := utils.StripRouteToAPIRoute(utils.ReflectFuncName(f))
 		fs.LocalJobs[signature] = &JobFunc{f, methods, signature}
 	}
 }
 
-func (fs *FS) AddShared(methods []string, jobs ...func(w http.ResponseWriter, r *http.Request) *task.Job) {
+func (fs *FS) AddShared(methods []string, jobs ...func(w http.ResponseWriter, r *http.Request, bg *task.Background)) {
 	for _, f := range jobs {
 		signature := utils.StripRouteToAPIRoute(utils.ReflectFuncName(f))
 		fs.SharedJobs[signature] = &JobFunc{f, methods, signature}