User websocket for log streaming, Fixes #115 and Fixes #86

api/logs.go

@@ -14,7 +14,7 @@ func LogsURL(app string) (*http.Response, error) {
 	return req.Do()
 }
 
-func Logs(logsURL string, stream bool, n int) (*http.Response, error) {
+func Logs(logsURL string, n int) (*http.Response, error) {
 	u, err := url.Parse(logsURL)
 	if err != nil {
 		return nil, errgo.Mask(err)
@@ -25,9 +25,8 @@ func Logs(logsURL string, stream bool, n int) (*http.Response, error) {
 		URL:      u.Scheme + "://" + u.Host,
 		Endpoint: u.Path,
 		Params: map[string]interface{}{
-			"token":  u.Query().Get("token"),
-			"stream": stream,
-			"n":      n,
+			"token": u.Query().Get("token"),
+			"n":     n,
 		},
 	}
 	return req.Do()

apps/logs.go

@@ -1,20 +1,29 @@
 package apps
 
 import (
-	"bufio"
 	"encoding/json"
 	"fmt"
 	stdio "io"
 	"io/ioutil"
-	"net/http"
+	"net/url"
+	"os"
 	"strings"
+	"time"
 
 	"github.com/Scalingo/cli/Godeps/_workspace/src/gopkg.in/errgo.v1"
 	"github.com/Scalingo/cli/api"
+	"github.com/Scalingo/cli/config"
 	"github.com/Scalingo/cli/debug"
 	"github.com/Scalingo/cli/io"
+	"golang.org/x/net/websocket"
 )
 
+type WSEvent struct {
+	Type      string    `json:"event"`
+	Log       string    `json:"log"`
+	Timestamp time.Time `json:"timestamp"`
+}
+
 type LogsRes struct {
 	LogsURL string   `json:"logs_url"`
 	App     *api.App `json:"app"`
@@ -43,44 +52,76 @@ func Logs(appName string, stream bool, n int) error {
 		return errgo.Mask(err, errgo.Any)
 	}
 
-	res, err = api.Logs(logsRes.LogsURL, stream, n)
+	if err = dumpLogs(logsRes.LogsURL, n); err != nil {
+		return errgo.Mask(err, errgo.Any)
+	}
+
+	if stream {
+		if err = streamLogs(logsRes.LogsURL); err != nil {
+			return errgo.Mask(err, errgo.Any)
+		}
+	}
+	return nil
+}
+
+func dumpLogs(logsURL string, n int) error {
+	res, err := api.Logs(logsURL, n)
 	if err != nil {
 		return errgo.Mask(err, errgo.Any)
 	}
+	defer res.Body.Close()
 
 	if res.StatusCode == 404 {
 		io.Error("There is not log for this application.")
 		return nil
 	}
 
-	if !stream {
-		buffer, _ := ioutil.ReadAll(res.Body)
-		fmt.Println(string(buffer))
-	} else {
-		return streamLogs(res)
+	_, err = stdio.Copy(os.Stdout, res.Body)
+	if err != nil {
+		return errgo.Mask(err, errgo.Any)
 	}
+
 	return nil
 }
 
-func streamLogs(res *http.Response) error {
-	var err error
-	reader := bufio.NewReader(res.Body)
-	for line, _, err := reader.ReadLine(); err == nil; line, _, err = reader.ReadLine() {
-		if len(line) != 0 {
-			parsedLine := strings.SplitN(string(line), ":", 2)
-			if len(parsedLine) != 2 {
-				// Invalid content from server, SSE should be
-				// msgname: content
-				// Anything else is wrong
-				continue
-			}
-			fmt.Println(
-				strings.TrimSpace(parsedLine[1]),
-			)
-		}
+func streamLogs(logsRawURL string) error {
+	var (
+		err    error
+		buffer [2048]byte
+		event  WSEvent
+	)
+
+	logsURL, err := url.Parse(logsRawURL)
+	if err != nil {
+		return errgo.Mask(err, errgo.Any)
 	}
-	if err != stdio.EOF {
+	if logsURL.Scheme == "https" {
+		logsURL.Scheme = "wss"
+	} else {
+		logsURL.Scheme = "ws"
+	}
+
+	conn, err := websocket.Dial(logsURL.String()+"&stream=true", "", "http://scalingo-cli.local/"+config.Version)
+	if err != nil {
 		return errgo.Mask(err, errgo.Any)
 	}
+
+	for {
+		n, err := conn.Read(buffer[:])
+		if err != nil {
+			return errgo.Mask(err, errgo.Any)
+		}
+		err = json.Unmarshal(buffer[:n], &event)
+		if err != nil {
+			return errgo.Mask(err, errgo.Any)
+		}
+		switch event.Type {
+		case "ping":
+			debug.Println("> ", event.Timestamp)
+		case "log":
+			fmt.Println(strings.TrimSpace(event.Log))
+		}
+	}
+
 	return nil
 }