harness-drone/server/ws.go

package server

import (
	"bufio"
	"encoding/json"
	"io"
	"net/http"
	"strconv"
	"time"

	"github.com/drone/drone/pkg/bus"

	log "github.com/Sirupsen/logrus"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"

	// "github.com/koding/websocketproxy"
)

const (
	// Time allowed to write the message to the client.
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the client.
	pongWait = 60 * time.Second

	// Send pings to client with this period. Must be less than pongWait.
	pingPeriod = (pongWait * 9) / 10
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin:     func(r *http.Request) bool { return true },
}

// GetRepoEvents will upgrade the connection to a Websocket and will stream
// event updates to the browser.
func GetRepoEvents(c *gin.Context) {
	bus_ := ToBus(c)
	repo := ToRepo(c)
	c.Writer.Header().Set("Content-Type", "text/event-stream")

	eventc := make(chan *bus.Event, 1)
	bus_.Subscribe(eventc)
	defer func() {
		bus_.Unsubscribe(eventc)
		log.Infof("closed event stream")
	}()

	c.Stream(func(w io.Writer) bool {
		select {
		case event := <-eventc:
			if event == nil {
				log.Infof("nil event received")
				return false
			}
			if event.Kind == bus.EventRepo &&
				event.Name == repo.FullName {
				d := map[string]interface{}{}
				json.Unmarshal(event.Msg, &d)
				c.SSEvent("message", d)
			}
		case <-c.Writer.CloseNotify():
			return false
		}
		return true
	})
}

func GetStream(c *gin.Context) {
	store := ToDatastore(c)
	repo := ToRepo(c)
	runner := ToRunner(c)
	commitseq, _ := strconv.Atoi(c.Params.ByName("build"))
	buildseq, _ := strconv.Atoi(c.Params.ByName("number"))

	// agent, err := store.BuildAgent(repo.FullName, build)
	// if err != nil {
	// 	c.Fail(404, err)
	// 	return
	// }

	commit, err := store.CommitSeq(repo, commitseq)
	if err != nil {
		c.Fail(404, err)
		return
	}
	build, err := store.BuildSeq(commit, buildseq)
	if err != nil {
		c.Fail(404, err)
		return
	}

	rc, err := runner.Logs(build)
	if err != nil {
		c.Fail(404, err)
		return
	}

	// upgrade the websocket
	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		c.Fail(400, err)
		return
	}

	var ticker = time.NewTicker(pingPeriod)
	var out = make(chan []byte)
	defer func() {
		log.Infof("closed stdout websocket")
		ticker.Stop()
		rc.Close()
		ws.Close()
	}()

	go func() {
		for {
			select {
			case <-c.Writer.CloseNotify():
				rc.Close()
				ws.Close()
				return
			case line := <-out:
				ws.WriteMessage(websocket.TextMessage, line)
			case <-ticker.C:
				ws.SetWriteDeadline(time.Now().Add(writeWait))
				err := ws.WriteMessage(websocket.PingMessage, []byte{})
				if err != nil {
					rc.Close()
					ws.Close()
					return
				}
			}
		}
	}()

	go func() {
		rd := bufio.NewReader(rc)
		for {
			str, err := rd.ReadBytes('\n')

			if err != nil {
				break
			}
			if len(str) == 0 {
				break
			}

			out <- str
		}
		rc.Close()
		ws.Close()
	}()

	readWebsocket(ws)

	// url_, err := url.Parse("ws://" + agent.Addr)
	// if err != nil {
	// 	c.Fail(500, err)
	// 	return
	// }
	// url_.Path = fmt.Sprintf("/stream/%s/%v/%v", repo.FullName, build, task)
	// proxy := websocketproxy.NewProxy(url_)
	// proxy.ServeHTTP(c.Writer, c.Request)

	// log.Debugf("closed websocket")
}

// readWebsocket will block while reading the websocket data
func readWebsocket(ws *websocket.Conn) {
	defer ws.Close()
	ws.SetReadLimit(512)
	ws.SetReadDeadline(time.Now().Add(pongWait))
	ws.SetPongHandler(func(string) error {
		ws.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})
	for {
		_, _, err := ws.ReadMessage()
		if err != nil {
			break
		}
	}
}