Refactor

Add examples
Simplify job interface (remove batching)
Don't serialize jobSpec struct to job, just job name
Implement query splitting using scatter or property range. Closes #18

Author: CaptainCodeman

common.go

@@ -8,12 +8,12 @@ import (
 )
 
 type (
+	// taskable (urgh, horrible name) is anything that can
+	// be run as a task. It's basically all our datastore models
+	// and used to set the private common values (context?)
 	taskable interface {
-		getID() string
-		setID(id string)
-
-		getQueue() string
-		setQueue(queue string)
+		getCommon() *common
+		setCommon(id string, jobSpec JobSpec, queue string)
 	}
 
 	// common contains properties that are common across all
@@ -22,6 +22,9 @@ type (
 		// Counters holds the task counters map
 		Counters Counters `datastore:"-"`
 
+		// Query is the datastore query spec
+		Query *Query `datastore:"-"`
+
 		// Active indicates if this task is still active
 		Active bool `datastore:"active,noindex"`
 
@@ -43,27 +46,25 @@ type (
 		// private fields used by local instance
 		id        string
 		queue     string
+		jobSpec   JobSpec
 		startTime time.Time
 	}
 )
 
-func (c *common) getID() string {
-	return c.id
-}
-func (c *common) setID(id string) {
-	c.id = id
+func (c *common) getCommon() *common {
+	return c
 }
 
-func (c *common) getQueue() string {
-	return c.queue
-}
-func (c *common) setQueue(queue string) {
+func (c *common) setCommon(id string, jobSpec JobSpec, queue string) {
+	c.id = id
+	c.jobSpec = jobSpec
 	c.queue = queue
 }
 
-func (c *common) start() {
+func (c *common) start(query *Query) {
 	c.Active = true
 	c.Counters = NewCounters()
+	c.Query = query
 	c.Count = 0
 	c.Started = getTime()
 	c.Updated = c.Started
@@ -88,9 +89,17 @@ func (c *common) Load(props []datastore.Property) error {
 
 	c.Counters = make(map[string]int64)
 	for _, prop := range props {
-		if strings.HasPrefix(prop.Name, "counters.") {
-			key := prop.Name[9:len(prop.Name)]
-			c.Counters[key] = prop.Value.(int64)
+		switch prop.Name {
+		case "query":
+			c.Query = &Query{}
+			if err := c.Query.GobDecode(prop.Value.([]byte)); err != nil {
+				return err
+			}
+		default:
+			if strings.HasPrefix(prop.Name, "counters.") {
+				key := prop.Name[9:len(prop.Name)]
+				c.Counters[key] = prop.Value.(int64)
+			}
 		}
 	}
 
@@ -111,5 +120,11 @@ func (c *common) Save() ([]datastore.Property, error) {
 		props = append(props, datastore.Property{Name: "counters." + key, Value: value, NoIndex: true, Multiple: false})
 	}
 
+	b, err := c.Query.GobEncode()
+	if err != nil {
+		return nil, err
+	}
+	props = append(props, datastore.Property{Name: "query", Value: b, NoIndex: true, Multiple: false})
+
 	return props, nil
 }

config.go

@@ -40,6 +40,8 @@ type (
 
 // newConfig creates a new config with default values
 func newConfig() *Config {
+	// TODO: use config as default, allow setting some values per-job
+	// and prevent config changes affecting already-running tasks
 	return &Config{
 		Path:            DefaultPath,
 		DatastorePrefix: "MP_",

example/example1.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"io"
+
+	"net/http"
+
+	"github.com/qedus/nds"
+	"golang.org/x/net/context"
+	"google.golang.org/appengine/datastore"
+
+	"github.com/captaincodeman/datastore-mapper"
+)
+
+type (
+	// simple keysonly iteration and aggregation using counters
+	example1 struct{}
+)
+
+func init() {
+	mapper.RegisterJob(&example1{})
+}
+
+func (x *example1) Query(r *http.Request) (*mapper.Query, error) {
+	q := mapper.NewQuery("photo")
+	q = q.Namespace("")
+	return q, nil
+}
+
+// Next processes the next item
+func (x *example1) Next(c context.Context, w io.Writer, counters mapper.Counters, key *datastore.Key) error {
+	// we need to load the entity ourselves
+	photo := new(Photo)
+	if err := nds.Get(c, key, photo); err != nil {
+		return err
+	}
+
+	counters.Increment(photo.Photographer.Name, 1)
+	return nil
+}

example/example2.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"io"
+
+	"net/http"
+
+	"golang.org/x/net/context"
+	"google.golang.org/appengine/datastore"
+
+	"github.com/captaincodeman/datastore-mapper"
+)
+
+type (
+	// simple eager iteration and aggregation using counters
+	example2 struct {
+		photo *Photo
+	}
+)
+
+func init() {
+	mapper.RegisterJob(&example2{})
+}
+
+func (x *example2) Query(r *http.Request) (*mapper.Query, error) {
+	q := mapper.NewQuery("photo")
+	q = q.Namespace("")
+	return q, nil
+}
+
+// Make creates the entity to load into
+func (x *example2) Make() interface{} {
+	x.photo = new(Photo)
+	return x.photo
+}
+
+// Next processes the next item
+func (x *example2) Next(c context.Context, w io.Writer, counters mapper.Counters, key *datastore.Key) error {
+	// the photo instance was loaded from the query by the mapper
+	counters.Increment(x.photo.Photographer.Name, 1)
+	return nil
+}

example/example3.go

@@ -0,0 +1,79 @@
+package main
+
+import (
+	"io"
+	"time"
+
+	"net/http"
+
+	"github.com/qedus/nds"
+	"golang.org/x/net/context"
+	"google.golang.org/appengine/datastore"
+	"google.golang.org/appengine/log"
+
+	"github.com/captaincodeman/datastore-mapper"
+)
+
+type (
+	// parse request parameters to create query
+	example3 struct {}
+)
+
+const (
+	dateFormat = "2006-01-02"
+)
+
+func init() {
+	mapper.RegisterJob(&example3{})
+}
+
+func (x *example3) Query(r *http.Request) (*mapper.Query, error) {
+	v := r.URL.Query()
+
+	fromStr := v.Get("from")
+	toStr := v.Get("to")
+	now := time.Now().UTC()
+
+	var from time.Time
+	var to time.Time
+	var err error
+
+	// default to previous day but allow any
+	if toStr == "" {
+		to = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
+	} else {
+		to, err = time.Parse(dateFormat, toStr)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if fromStr == "" {
+		from = to.Add(time.Duration(-24) * time.Hour)
+	} else {
+		from, err = time.Parse(dateFormat, fromStr)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	q := mapper.NewQuery("photo")
+	q = q.Namespace("")
+	q = q.Filter("taken >=", from)
+	q = q.Filter("taken <", to)
+
+	return q, nil
+}
+
+// Next processes the next item
+func (x *example3) Next(c context.Context, w io.Writer, counters mapper.Counters, key *datastore.Key) error {
+	photo := new(Photo)
+	if err := nds.Get(c, key, photo); err != nil {
+		log.Errorf(c, err.Error())
+		return err
+	}
+	photo.ID = key.IntID()
+
+	counters.Increment(photo.Photographer.Name, 1)
+	return nil
+}

example/example4.go

@@ -0,0 +1,59 @@
+package main
+
+import (
+	"io"
+
+	"net/http"
+
+	"golang.org/x/net/context"
+	"google.golang.org/appengine/datastore"
+	"google.golang.org/appengine/log"
+
+	"github.com/captaincodeman/datastore-mapper"
+)
+
+type (
+	// lifecycle notifications
+	example4 struct{}
+)
+
+func init() {
+	mapper.RegisterJob(&example4{})
+}
+
+func (x *example4) Query(r *http.Request) (*mapper.Query, error) {
+	q := mapper.NewQuery("photo")
+	q = q.Namespace("")
+	return q, nil
+}
+
+// Next processes the next item
+func (x *example4) Next(c context.Context, w io.Writer, counters mapper.Counters, key *datastore.Key) error {
+	return nil
+}
+
+// job lifecycle notifications
+func (x *example4) JobStarted(c context.Context, id string) {
+	log.Debugf(c, "JobStarted %s", id)
+}
+func (x *example4) JobCompleted(c context.Context, id string) {
+	log.Debugf(c, "JobCompleted %s", id)
+}
+func (x *example4) NamespaceStarted(c context.Context, id string, namespace string) {
+	log.Debugf(c, "NamespaceStarted %s %s", id, namespace)
+}
+func (x *example4) NamespaceCompleted(c context.Context, id string, namespace string) {
+	log.Debugf(c, "NamespaceCompleted %s %s", id, namespace)
+}
+func (x *example4) ShardStarted(c context.Context, id string, namespace string, shard int) {
+	log.Debugf(c, "ShardStarted %s %s %d", id, namespace, shard)
+}
+func (x *example4) ShardCompleted(c context.Context, id string, namespace string, shard int) {
+	log.Debugf(c, "ShardCompleted %s %s %d", id, namespace, shard)
+}
+func (x *example4) SliceStarted(c context.Context, id string, namespace string, shard, slice int) {
+	log.Debugf(c, "SliceStarted %s %s %d %d", id, namespace, shard, slice)
+}
+func (x *example4) SliceCompleted(c context.Context, id string, namespace string, shard, slice int) {
+	log.Debugf(c, "SliceCompleted %s %s %d %d", id, namespace, shard, slice)
+}

example/example5.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"io"
+
+	"encoding/json"
+	"net/http"
+
+	"golang.org/x/net/context"
+	"google.golang.org/appengine/datastore"
+
+	"github.com/captaincodeman/datastore-mapper"
+)
+
+type (
+	// export custom JSON to Cloud Storage
+	example5 struct {
+		photo *Photo
+	}
+
+	photoOutput struct {
+		*Photo
+		// add namespace for bigquery
+		Namespace string `json:"namespace"`
+	}
+)
+
+func init() {
+	mapper.RegisterJob(&example5{})
+}
+
+func (x *example5) Query(r *http.Request) (*mapper.Query, error) {
+	q := mapper.NewQuery("photo")
+	q = q.Namespace("")
+	return q, nil
+}
+
+// Make creates the entity to load into
+func (x *example5) Make() interface{} {
+	x.photo = new(Photo)
+	return x.photo
+}
+
+// Next processes the next item
+func (x *example5) Next(c context.Context, w io.Writer, counters mapper.Counters, key *datastore.Key) error {
+	photo := x.photo
+	photo.ID = key.IntID()
+
+	out := &photoOutput{
+		Photo:     photo,
+		Namespace: key.Namespace(),
+	}
+
+	enc := json.NewEncoder(w)
+	enc.Encode(out)
+
+	return nil
+}