WIP: prototype hacking/refactoring

lib/lib.go

@@ -1,16 +1,11 @@
 package lib
 
 import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"io"
-	"io/ioutil"
 	"log"
-	"net/http"
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/jimenez/mesoscon-demo/lib/mesosproto"
+	"github.com/jimenez/mesoscon-demo/lib/transport"
 )
 
 const ENDPOINT = "/master/api/v1/scheduler"
@@ -32,23 +27,13 @@ func New(master, name string) *DemoLib {
 	}
 }
 
-func (lib *DemoLib) handleEvents(body io.Reader) {
-	dec := json.NewDecoder(body)
-	for {
-		var event mesosproto.Event
-
-		if err := dec.Decode(&event); err != nil {
-			if err == io.EOF {
-				break
-			}
-			if event.GetType() == mesosproto.Event_UPDATE {
-				taskStatus := event.GetUpdate().GetStatus()
-				log.Println("Status for", taskStatus.GetTaskId().GetValue(), "is", taskStatus.GetState().String())
-			}
-			continue
-		}
-
+func (lib *DemoLib) handleEvents(s transport.Subscription) {
+	ech := s.Events()
+	for event := range ech {
 		switch event.GetType() {
+		case mesosproto.Event_UPDATE:
+			taskStatus := event.GetUpdate().GetStatus()
+			log.Println("Status for", taskStatus.GetTaskId().GetValue(), "is", taskStatus.GetState().String())
 		case mesosproto.Event_SUBSCRIBED:
 			lib.frameworkID = event.GetSubscribed().GetFrameworkId()
 			log.Println("framework", lib.name, "subscribed succesfully (", lib.frameworkID.String(), ")")
@@ -59,42 +44,15 @@ func (lib *DemoLib) handleEvents(body io.Reader) {
 			log.Println("framework", lib.name, "received", len(event.GetOffers().GetOffers()), "offer(s)")
 		}
 	}
+	log.Println("subscription terminated:", s.Err())
 }
 
 func (lib *DemoLib) Subscribe() error {
-	call := mesosproto.Call{
-		Type: mesosproto.Call_SUBSCRIBE.Enum(),
-		Subscribe: &mesosproto.Call_Subscribe{
-			FrameworkInfo: lib.frameworkInfo,
-		},
-	}
-
-	body, err := proto.Marshal(&call)
+	s, err := transport.Subscribe(lib.master, lib.frameworkInfo, false)
 	if err != nil {
 		return err
 	}
-
-	req, err := http.NewRequest("POST", "http://"+lib.master+ENDPOINT, bytes.NewReader(body))
-	if err != nil {
-		return err
-	}
-	req.Header.Set("Content-Type", "application/x-protobuf")
-	req.Header.Set("Accept", "application/json")
-
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return err
-	}
-
-	if resp.StatusCode != 200 {
-		body, err := ioutil.ReadAll(resp.Body)
-		if err != nil {
-			return err
-		}
-		return fmt.Errorf("%s", body)
-	}
-
-	go lib.handleEvents(resp.Body)
+	go lib.handleEvents(s)
 	return nil
 
 }

lib/records/doc.go

@@ -0,0 +1,9 @@
+package records
+
+// Package records implements the Mesos variant of RecordIO decoding, whereby
+// each record is prefixed by a line that indicates the length (decimal, printed
+// in ASCII) of the record. The octets of the record immediately follow the
+// length-line. Zero-length records are allowed.
+//
+// This package does not enforce any particular record format: that choice is
+// left up to the caller in the form of the Unmarshaler.

lib/records/records.go

@@ -0,0 +1,119 @@
+package records
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net/textproto"
+	"strconv"
+)
+
+const DefaultMaxEventLength = uint64(1024 * 1024)
+
+var (
+	empty          = []byte{}
+	MaxEventLength = DefaultMaxEventLength // should not be larger than what int will hold
+)
+
+type Decoder interface {
+	Decode(v interface{}) error
+}
+
+// A ProtocolError describes a protocol violation such as an invalid `event-length`.
+type ProtocolError string
+
+func (p ProtocolError) Error() string {
+	return string(p)
+}
+
+type DecoderFunc func(v interface{}) error
+
+func (df DecoderFunc) Decode(v interface{}) error {
+	return df(v)
+}
+
+type Unmarshaler func([]byte, interface{}) error
+
+type decoder struct {
+	in        *textproto.Reader
+	un        Unmarshaler
+	state     stateFn
+	remaining int
+	records   chan []byte
+	buf       []byte
+}
+
+type stateFn func(d *decoder) (stateFn, error)
+
+func NewDecoder(in io.Reader, un Unmarshaler) Decoder {
+	bio := bufio.NewReader(in)
+	return &decoder{
+		in:      textproto.NewReader(bio),
+		un:      un,
+		records: make(chan []byte, 1),
+		state:   headerState,
+	}
+}
+
+func (d *decoder) Decode(v interface{}) (err error) {
+stateLoop:
+	for err == nil {
+		if d.state, err = d.state(d); err == nil {
+			select {
+			case r := <-d.records:
+				err = d.un(r, v)
+				break stateLoop
+			default:
+			}
+		}
+	}
+	return
+}
+
+func headerState(d *decoder) (stateFn, error) {
+	ll, err := d.in.ReadLine()
+	if err != nil {
+		//TODO(jdef) check for EOF?
+		return headerState, err
+	}
+	nbytes, err := strconv.ParseUint(ll, 10, 64)
+	if err != nil {
+		return failedState, ProtocolError(fmt.Sprintf("protocol violation, failed to parse event-length: %v", err))
+	}
+	// enforce max message size here
+	if nbytes > MaxEventLength {
+		// TODO(jdef) enter a "skip-bytes" state instead? if so, we may need to indicate
+		// that the error is temporary
+		return failedState, ProtocolError(fmt.Sprintf("protocol violation, event-length %d exceeds max allowed (%d)", nbytes, MaxEventLength))
+	}
+	if d.remaining = int(nbytes); d.remaining == 0 {
+		d.buf = empty
+	} else {
+		// TODO(jdef) attempt to reuse existing buffer?
+		d.buf = make([]byte, d.remaining)
+	}
+	return eventState, nil
+}
+
+func failedState(d *decoder) (stateFn, error) {
+	return failedState, ProtocolError("decoder is in a failed state and will not recover")
+}
+
+func eventState(d *decoder) (stateFn, error) {
+	var err error
+	if d.remaining > 0 {
+		off := len(d.buf) - d.remaining
+		n, e := d.in.R.Read(d.buf[off:])
+		if n > 0 {
+			d.remaining -= n
+		}
+		err = e
+	}
+	if d.remaining == 0 {
+		d.records <- d.buf
+		// don't communicate an error in the same step that we generated a record,
+		// let the next stateFn deal with any stream errors (e.g. EOF)
+		return headerState, nil
+	}
+	return eventState, err
+}

lib/records/records_test.go

@@ -0,0 +1,63 @@
+package records
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+)
+
+func ExampleDecoder() {
+	un := Unmarshaler(func(b []byte, v interface{}) error {
+		s, ok := v.([]string)
+		if !ok {
+			return errors.New("unexpected object store type")
+		}
+		if len(s) == 0 {
+			return errors.New("not enough space in object store")
+		}
+		s[0] = string(b)
+		return nil
+	})
+	d := NewDecoder(bytes.NewBufferString("5\nhello0\n6\nworld!"), un)
+	s := []string{""}
+	for {
+		err := d.Decode(s)
+		if err != nil {
+			if err != io.EOF {
+				fmt.Println(err)
+			}
+			break
+		} else if s[0] == "" {
+			fmt.Println("--empty--")
+		} else {
+			fmt.Println(s[0])
+		}
+	}
+	// Output:
+	// hello
+	// --empty--
+	// world!
+}
+
+func ExampleDecoder_json() {
+	type Demo struct {
+		Hello string
+	}
+
+	s := &Demo{}
+	un := Unmarshaler(json.Unmarshal)
+	d := NewDecoder(bytes.NewBufferString("18\n{\"hello\": \"world\"}"), un)
+
+	err := d.Decode(s)
+	if err != nil {
+		if err != io.EOF {
+			fmt.Println(err)
+		}
+	} else {
+		fmt.Printf("%+v", *s)
+	}
+	// Output:
+	// {Hello:world}
+}