skipping pushing message to client if there are no messages; fixes flaky test and de-intensifies read loop

caddyhugo.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"html/template"
 	"net/http"
-	_ "net/http/pprof"
 	"os/exec"
 	"path/filepath"
 	"strings"
@@ -31,6 +30,7 @@ func init() {
 	// ... there are others. See the godoc.
 }
 
+// CaddyHugo implements the plugin
 type CaddyHugo struct {
 	ServerType string
 	Site       *httpserver.SiteConfig
@@ -49,6 +49,7 @@ type CaddyHugo struct {
 	ltime uint64
 }
 
+// Build rebuilds the cached state of the site. TODO: determine if this republishes
 func (ch *CaddyHugo) Build() error {
 	err := ch.HugoSites.Build(hugolib.BuildCfg{ResetState: true})
 	if err != nil {
@@ -58,7 +59,8 @@ func (ch *CaddyHugo) Build() error {
 	return nil
 }
 
-func (ch CaddyHugo) BasePath() string {
+// BasePath returns the directory that the CaddyHugo internal/author pages are under
+func (ch *CaddyHugo) BasePath() string {
 	return "/hugo"
 }
 
@@ -71,6 +73,7 @@ func (ch *CaddyHugo) docFilename(orig string) string {
 	return filepath.Join(ch.Dir, docname(orig))
 }
 
+// Publish really renders new content into the public directory
 func (ch *CaddyHugo) Publish() error {
 	cmd := exec.Command("hugo")
 	cmd.Dir = ch.Dir
@@ -82,7 +85,8 @@ func (ch *CaddyHugo) Publish() error {
 	return nil
 }
 
-func (ch CaddyHugo) TmplData(r *http.Request, docref *docref) interface{} {
+// TmplData collects data for template execution
+func (ch *CaddyHugo) TmplData(r *http.Request, docref *docref) interface{} {
 	var doc *acedoc.Document
 	if docref != nil {
 		doc = docref.doc

content.go

@@ -79,7 +79,7 @@ func GetContent(siteRoot string, sites *hugolib.HugoSites) ([]Content, error) {
 // If ctype is empty string, "default" is used. The return value is the filename,
 // which may be modified from the title and includes the content type if other than
 // default, but does not include the full directory relative to the site.
-func (ch CaddyHugo) NewContent(name, ctype string) (string, error) {
+func (ch *CaddyHugo) NewContent(name, ctype string) (string, error) {
 	if filepath.Ext(name) != ".md" {
 		name += ".md"
 	}
@@ -101,7 +101,6 @@ func (ch CaddyHugo) NewContent(name, ctype string) (string, error) {
 		out, err := cmd.CombinedOutput()
 		if err != nil {
 			return filename, fmt.Errorf("error running 'hugo new': %v; %v", err, string(out))
-			return filename, err
 		}
 	}
 

deltas.go

@@ -2,6 +2,7 @@ package caddyhugo
 
 import (
 	"fmt"
+	"log"
 	"net/http"
 	"time"
 
@@ -9,6 +10,12 @@ import (
 	"github.com/gorilla/websocket"
 )
 
+// DeltaConn identifies the methods used in the original websocket implementation.
+type DeltaConn interface {
+	ReadJSON(v interface{}) error
+	WriteJSON(v interface{}) error
+}
+
 const (
 	IdleWebsocketTimeout = 10 * time.Minute
 	WebsocketFileTicker  = 1 * time.Second
@@ -50,25 +57,44 @@ func (ch *CaddyHugo) DeltaWebsocket(w http.ResponseWriter, r *http.Request) (int
 		return http.StatusBadRequest, err
 	}
 
+	return ch.handleConn(conn, doc)
+}
+
+func (ch *CaddyHugo) Message(deltas ...acedoc.Delta) Message {
+	return Message{
+		Deltas: deltas,
+		LTime:  ch.LTime(),
+	}
+}
+
+func (ch *CaddyHugo) handleConn(conn DeltaConn, doc *docref) (int, error) {
 	const idlePing = 15 * time.Second
 	const idlePingShort = 1 * time.Millisecond
-	var timer *time.Timer
 
-	timer = time.AfterFunc(idlePing, func() {
-		conn.WriteJSON(Message{
-			Deltas: []acedoc.Delta{},
-			LTime:  ch.LTime(),
-		})
-		timer.Reset(idlePing)
-	})
+	errCh := make(chan error)
+	doneCh := make(chan struct{})
+	defer func() {
+		close(doneCh)
+		close(errCh)
+		for err := range errCh {
+			log.Println(err)
+		}
+	}()
+
+	timer := time.NewTimer(idlePing)
+	resetTimer := func(d time.Duration) {
+		if !timer.Stop() {
+			<-timer.C
+		}
+		timer.Reset(d)
+	}
+
+	wroteMessagesCh := make(chan Message, 2)
 
 	client := doc.doc.Client(acedoc.DeltaHandlerFunc(func(ds []acedoc.Delta) error {
-		timer.Reset(idlePing)
-		err := conn.WriteJSON(Message{
-			Deltas: ds,
-			LTime:  ch.LTime(),
-		})
-		return err
+		m := ch.Message(ds...)
+		wroteMessagesCh <- m
+		return conn.WriteJSON(m)
 	}))
 
 	ch.mtx.Lock()
@@ -82,21 +108,49 @@ func (ch *CaddyHugo) DeltaWebsocket(w http.ResponseWriter, r *http.Request) (int
 		ch.mtx.Unlock()
 	}()
 
+	readMessagesCh := make(chan Message, 2)
+	go func() {
 		for {
 			var message Message
+
 			err := conn.ReadJSON(&message)
 			if err != nil {
-			return http.StatusBadRequest, err
+				errCh <- err
+				return
 			}
-
 			ch.ObserveLTime(message.LTime)
-		timer.Reset(idlePingShort)
+
+			if len(message.Deltas) == 0 {
+				time.Sleep(10 * time.Microsecond)
+				continue
+			}
 
 			err = client.PushDeltas(message.Deltas...)
 			if err != nil {
-			return http.StatusBadRequest, err
+				errCh <- err
+				return
 			}
 
+			select {
+			case readMessagesCh <- message:
+			case <-doneCh:
+				return
+			}
+
+		}
+	}()
+
+	for {
+		select {
+		case <-timer.C:
+			conn.WriteJSON(ch.Message())
+		case <-readMessagesCh:
+			resetTimer(idlePingShort)
+		case <-wroteMessagesCh:
+			resetTimer(idlePing)
+		case <-doneCh:
+			return 200, nil
+		}
 	}
 }
 

doc_test.go

@@ -1,6 +1,7 @@
 package caddyhugo
 
 import (
+	"encoding/json"
 	"io/ioutil"
 	"os"
 	"os/exec"
@@ -13,23 +14,23 @@ import (
 )
 
 type World struct {
-	CH         CaddyHugo
+	CH         *CaddyHugo
 	BlogFolder string
 }
 
-func (w World) Clean() {
+func (w *World) Clean() {
 	if w.BlogFolder != "" {
 		os.RemoveAll(w.BlogFolder)
 	}
 }
 
-func NewWorld(t *testing.T) World {
+func NewWorld(t *testing.T) *World {
 	dir, err := ioutil.TempDir("", "caddy-hugo2-test-")
 	if err != nil {
 		t.Fatalf("error initializing test environment: %v", err)
 	}
 
-	w := World{BlogFolder: dir}
+	w := &World{BlogFolder: dir}
 
 	cmd := exec.Command("hugo", "new", "site", dir)
 	cmd.Dir = dir
@@ -38,6 +39,7 @@ func NewWorld(t *testing.T) World {
 		t.Fatalf("error initializing test site: %v\n\n%v", err, string(out))
 	}
 
+	w.CH = &CaddyHugo{}
 	w.CH.Setup(dir)
 
 	return w
@@ -81,12 +83,224 @@ func TestEdits(t *testing.T) {
 	doc.doc.Apply(send...)
 
 	<-time.After(5 * time.Second)
+
+	mtx.Lock()
+	defer mtx.Unlock()
 	if len(received) != len(send) {
 		t.Errorf("expected %d deltas, received %d; expected: %v, received: %v", len(send), len(received), send, received)
 	}
 
 }
 
+type WebsocketTester struct {
+	receivedPointer int
+	received        [][]byte
+	wroteMessages   []Message
+	wroteDeltas     []acedoc.Delta
+	mtx             sync.Mutex
+}
+
+func (ws *WebsocketTester) ReadJSON(v interface{}) error {
+	ws.mtx.Lock()
+	defer ws.mtx.Unlock()
+
+	if len(ws.received) <= ws.receivedPointer {
+		return nil
+	}
+
+	err := json.Unmarshal(ws.received[ws.receivedPointer], v)
+	ws.receivedPointer++
+	return err
+}
+
+func (ws *WebsocketTester) WriteJSON(v interface{}) error {
+	ws.mtx.Lock()
+	defer ws.mtx.Unlock()
+
+	m, ok := v.(Message)
+	if !ok {
+		panic("wrong type written to WebsocketTester")
+	}
+
+	if len(m.Deltas) == 0 {
+		return nil
+	}
+
+	ws.wroteMessages = append(ws.wroteMessages, m)
+	ws.wroteDeltas = append(ws.wroteDeltas, m.Deltas...)
+
+	return nil
+}
+
+func (ws *WebsocketTester) ReceiveJSON(v interface{}) error {
+	ws.mtx.Lock()
+	defer ws.mtx.Unlock()
+
+	out, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+
+	ws.received = append(ws.received, out)
+	return nil
+}
+
+func TestDeltasSingle(t *testing.T) {
+	w := NewWorld(t)
+	defer w.Clean()
+
+	const title = "test"
+
+	_, err := w.CH.NewContent(title, "")
+	if err != nil {
+		t.Fatal("couldn't create new content:", err)
+	}
+
+	client := new(WebsocketTester)
+
+	doc, err := w.CH.client("content/" + title + ".md")
+	if err != nil {
+		t.Fatal("couldn't establish docref for client 0:", err)
+	}
+
+	go w.CH.handleConn(client, doc)
+
+	a := acedoc.Insert(0, 0, "a")
+
+	// pretend to get one sent from the "browser"
+	client.ReceiveJSON(w.CH.Message(a))
+
+	// wait to make sure it was processed
+	time.Sleep(50 * time.Millisecond)
+
+	// we shouldn't have written back to the client,
+	// so we expect to have written 0 messages
+	if len(client.wroteMessages) != 0 {
+		t.Errorf("client wrote %d messages, should have written %d", len(client.wroteMessages), 0)
+	}
+
+	// we received one, so make sure that's counted properly
+	if len(client.received) != 1 {
+		t.Errorf("client has %d messages, should have received %d", len(client.received), 1)
+	}
+}
+
+func TestDeltasDouble(t *testing.T) {
+	w := NewWorld(t)
+	defer w.Clean()
+
+	const title = "test"
+
+	_, err := w.CH.NewContent(title, "")
+	if err != nil {
+		t.Fatal("couldn't create new content:", err)
+	}
+
+	clientA := new(WebsocketTester)
+	clientB := new(WebsocketTester)
+
+	doc, err := w.CH.client("content/" + title + ".md")
+	if err != nil {
+		t.Fatal("couldn't establish docref for client 0:", err)
+	}
+
+	go w.CH.handleConn(clientA, doc)
+	go w.CH.handleConn(clientB, doc)
+
+	// send the first message, simulating the browser on clientA
+	clientA.ReceiveJSON(w.CH.Message(acedoc.Insert(0, 0, "a")))
+
+	time.Sleep(100 * time.Millisecond)
+
+	clientA.mtx.Lock()
+	clientB.mtx.Lock()
+
+	// so we expect clientA to have written 0 messages, and
+	// clientB to have written 1
+	if len(clientA.wroteMessages) != 0 || len(clientB.wroteMessages) != 1 {
+		t.Errorf("clientA wrote %d messages, should have written 0. clientB wrote %d, should have written 1", len(clientA.wroteMessages), len(clientB.wroteMessages))
+	}
+
+	// we received one via clientA and zero via clientB, so make sure
+	// that's counted properly
+	if len(clientA.received) != 1 || len(clientB.received) != 0 {
+		t.Errorf("clientA has %d messages, should have received 1; clientB has %d messages, should have received 0", len(clientA.received), len(clientB.received))
+	}
+
+	clientA.mtx.Unlock()
+	clientB.mtx.Unlock()
+
+	// send the second message, via clientB
+	clientB.ReceiveJSON(w.CH.Message(acedoc.Insert(0, 0, "b")))
+
+	time.Sleep(400 * time.Millisecond)
+
+	clientA.mtx.Lock()
+	clientB.mtx.Lock()
+
+	// so we expect clientA to have written 1 message this time, and
+	// clientB to have written nothing new, so 1 still
+	if len(clientA.wroteMessages) != 1 || len(clientB.wroteMessages) != 1 {
+		t.Errorf("clientA wrote %d messages, should have written 1. clientB wrote %d, should have written 1 (just from before)", len(clientA.wroteMessages), len(clientB.wroteMessages))
+	}
+
+	// we received zero (new) via clientA and one via clientB, so make sure
+	// that's counted properly
+	if len(clientA.received) != 1 || len(clientB.received) != 1 {
+		t.Errorf("clientA has %d messages, should have received 1; clientB has %d messages, should have received 1", len(clientA.received), len(clientB.received))
+	}
+	clientA.mtx.Unlock()
+	clientB.mtx.Unlock()
+}
+
+func TestDeltasMulti(t *testing.T) {
+	w := NewWorld(t)
+	defer w.Clean()
+
+	const title = "test"
+
+	_, err := w.CH.NewContent(title, "")
+	if err != nil {
+		t.Fatal("couldn't create new content:", err)
+	}
+
+	clients := []*WebsocketTester{{}, {}, {}}
+
+	doc, err := w.CH.client("content/" + title + ".md")
+	if err != nil {
+		t.Fatal("couldn't establish docref:", err)
+	}
+
+	go w.CH.handleConn(clients[0], doc)
+	go w.CH.handleConn(clients[1], doc)
+	go w.CH.handleConn(clients[2], doc)
+
+	a := acedoc.Insert(0, 0, "a")
+	b := acedoc.Insert(0, 0, "b")
+	c := acedoc.Insert(0, 0, "c")
+
+	clients[0].ReceiveJSON(w.CH.Message(a))
+	clients[1].ReceiveJSON(w.CH.Message(b))
+	clients[2].ReceiveJSON(w.CH.Message(c))
+
+	time.Sleep(400 * time.Millisecond)
+
+	for i, client := range clients {
+		client.mtx.Lock()
+		// all clients should have "written" 2 deltas (could be the same
+		// message) that came from the other clients
+		if len(client.wroteDeltas) != 2 {
+			t.Errorf("client %d wrote %d deltas, should have written 2", i, len(client.wroteDeltas))
+		}
+
+		// all clients "received" 1 message from the "browser"
+		if len(client.received) != 1 {
+			t.Errorf("client %d has %d messages, should have received 1", i, len(client.received))
+		}
+		client.mtx.Unlock()
+	}
+}
+
 func TestPagesInPagesOut(t *testing.T) {
 	w := NewWorld(t)
 	defer w.Clean()

http.go

@@ -69,7 +69,7 @@ func (ch *CaddyHugo) ServeHTTPWithNext(next httpserver.Handler, c *caddy.Control
 	return 404, nil
 }
 
-func (ch CaddyHugo) ServeNewContent(w http.ResponseWriter, r *http.Request) (int, error) {
+func (ch *CaddyHugo) ServeNewContent(w http.ResponseWriter, r *http.Request) (int, error) {
 	name := r.FormValue("name")
 	ctype := r.FormValue("type")
 
@@ -83,7 +83,7 @@ func (ch CaddyHugo) ServeNewContent(w http.ResponseWriter, r *http.Request) (int
 	http.Redirect(w, r, filepath.Join("/hugo/edit/", "content", filename), http.StatusFound)
 	return http.StatusFound, nil
 }
-func (ch CaddyHugo) Middleware(c *caddy.Controller) httpserver.Middleware {
+func (ch *CaddyHugo) Middleware(c *caddy.Controller) httpserver.Middleware {
 	return func(next httpserver.Handler) httpserver.Handler {
 		return httpserver.HandlerFunc(func(w http.ResponseWriter, r *http.Request) (int, error) {
 			return ch.ServeHTTPWithNext(next, c, w, r)
@@ -91,11 +91,11 @@ func (ch CaddyHugo) Middleware(c *caddy.Controller) httpserver.Middleware {
 	}
 }
 
-func (ch CaddyHugo) Auth(r *http.Request) bool {
+func (ch *CaddyHugo) Auth(r *http.Request) bool {
 	return true
 }
 
-func (ch CaddyHugo) Match(r *http.Request) bool {
+func (ch *CaddyHugo) Match(r *http.Request) bool {
 	if strings.HasPrefix(r.URL.Path, "/media/") {
 		return true
 	}
@@ -107,7 +107,7 @@ func (ch CaddyHugo) Match(r *http.Request) bool {
 	return strings.HasPrefix(r.URL.Path, "/hugo/")
 }
 
-func (ch CaddyHugo) Admin() httpserver.Handler {
+func (ch *CaddyHugo) Admin() httpserver.Handler {
 	return httpserver.HandlerFunc(func(w http.ResponseWriter, r *http.Request) (int, error) {
 		err := ch.adminTmpl.Execute(w, ch.TmplData(r, nil))
 		if err != nil {
@@ -120,7 +120,7 @@ func (ch CaddyHugo) Admin() httpserver.Handler {
 	})
 }
 
-func (ch CaddyHugo) AuthorHome() httpserver.Handler {
+func (ch *CaddyHugo) AuthorHome() httpserver.Handler {
 	return httpserver.HandlerFunc(func(w http.ResponseWriter, r *http.Request) (int, error) {
 		td := ch.TmplData(r, nil)
 		err := ch.authorTmpl.Execute(w, td)

media.go

@@ -3,9 +3,9 @@ package caddyhugo
 import (
 	"fmt"
 	"image"
-	_ "image/gif"
+	_ "image/gif" // for processing images
 	"image/jpeg"
-	_ "image/png"
+	_ "image/png" // for processing images
 	"io"
 	"net/http"
 	"os"
@@ -217,6 +217,9 @@ func (ch *CaddyHugo) uploadMedia(w http.ResponseWriter, r *http.Request) (int, e
 
 	for {
 		part, err := mr.NextPart()
+		if err == io.EOF {
+			break
+		}
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return 400, nil
@@ -344,7 +347,6 @@ func parseSizeString(str string, actual image.Rectangle) (image.Rectangle, error
 		// w was the only dimension given, so set it to the greater dimension
 		// of the actual image size
 		if actual.Dx() > actual.Dy() {
-			w = w
 			h = 0
 		} else {
 			h = w

templates.go

@@ -14,11 +14,11 @@ import (
 	"github.com/mholt/caddy/caddyhttp/httpserver"
 )
 
-func (t tmplData) Content() ([]Content, error) {
+func (t *tmplData) Content() ([]Content, error) {
 	return GetContent(t.Site.Root, t.HugoSites)
 }
 
-func (t tmplData) ContentTypes() ([]string, error) {
+func (t *tmplData) ContentTypes() ([]string, error) {
 	nameMap := map[string]struct{}{"default": struct{}{}}
 
 	names, err := t.contentTypes(path.Join(t.Site.Root, "archetypes"))
@@ -45,7 +45,7 @@ func (t tmplData) ContentTypes() ([]string, error) {
 	return out, nil
 }
 
-func (t tmplData) contentTypes(dir string) ([]string, error) {
+func (t *tmplData) contentTypes(dir string) ([]string, error) {
 	layoutDir, err := os.Open(path.Join(t.Site.Root, "archetypes"))
 	if err != nil {
 		fmt.Println("opening layout dir", err)
@@ -65,12 +65,12 @@ func (t tmplData) contentTypes(dir string) ([]string, error) {
 type tmplData struct {
 	Site *httpserver.SiteConfig
 	R    *http.Request
-	CaddyHugo
+	*CaddyHugo
 	Doc    *acedoc.Document
 	docref *docref
 }
 
-func (t tmplData) LoadContent() (string, error) {
+func (t *tmplData) LoadContent() (string, error) {
 	return t.Doc.Contents(), nil
 }
 
@@ -79,7 +79,7 @@ func baseNoExt(name string) string {
 	return base[:len(base)-len(path.Ext(base))]
 }
 
-func (t tmplData) IframeSource() string {
+func (t *tmplData) IframeSource() string {
 	name := baseNoExt(t.docref.name)
 	ctype := baseNoExt(path.Dir(t.docref.name))
 	if ctype == "content" {