put updated agent behind flag

server/stream.go

@@ -1,10 +1,13 @@
 package server
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 	"time"
 
+	"github.com/cncd/logging"
+	"github.com/cncd/pubsub"
 	"github.com/drone/drone/cache"
 	"github.com/drone/drone/model"
 	"github.com/drone/drone/router/middleware/session"
@@ -194,3 +197,158 @@ func reader(ws *websocket.Conn) {
 		}
 	}
 }
+
+//
+// CANARY IMPLEMENTATION
+//
+// This file is a complete disaster because I'm trying to wedge in some
+// experimental code. Please pardon our appearance during renovations.
+//
+
+func LogStream2(c *gin.Context) {
+	repo := session.Repo(c)
+	buildn, _ := strconv.Atoi(c.Param("build"))
+	jobn, _ := strconv.Atoi(c.Param("number"))
+
+	build, err := store.GetBuildNumber(c, repo, buildn)
+	if err != nil {
+		logrus.Debugln("stream cannot get build number.", err)
+		c.AbortWithError(404, err)
+		return
+	}
+	job, err := store.GetJobNumber(c, build, jobn)
+	if err != nil {
+		logrus.Debugln("stream cannot get job number.", err)
+		c.AbortWithError(404, err)
+		return
+	}
+	if job.Status != model.StatusRunning {
+		logrus.Debugln("stream not found.")
+		c.AbortWithStatus(404)
+		return
+	}
+
+	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		if _, ok := err.(websocket.HandshakeError); !ok {
+			logrus.Errorf("Cannot upgrade websocket. %s", err)
+		}
+		return
+	}
+	logrus.Debugf("Successfull upgraded websocket")
+
+	ticker := time.NewTicker(pingPeriod)
+	logc := make(chan []byte, 10)
+
+	ctx, cancel := context.WithCancel(
+		context.Background(),
+	)
+	defer func() {
+		cancel()
+		ticker.Stop()
+		close(logc)
+		logrus.Debugf("Successfully closing websocket")
+	}()
+
+	go func() {
+		// TODO remove global variable
+		config.logger.Tail(ctx, fmt.Sprint(job.ID), func(entries ...*logging.Entry) {
+			for _, entry := range entries {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					logc <- entry.Data
+				}
+			}
+		})
+		cancel()
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case buf, ok := <-logc:
+				if ok {
+					ws.SetWriteDeadline(time.Now().Add(writeWait))
+					ws.WriteMessage(websocket.TextMessage, buf)
+				}
+			case <-ticker.C:
+				err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait))
+				if err != nil {
+					return
+				}
+			}
+		}
+	}()
+	reader(ws)
+}
+
+func EventStream2(c *gin.Context) {
+	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		if _, ok := err.(websocket.HandshakeError); !ok {
+			logrus.Errorf("Cannot upgrade websocket. %s", err)
+		}
+		return
+	}
+	logrus.Debugf("Successfull upgraded websocket")
+
+	user := session.User(c)
+	repo := map[string]bool{}
+	if user != nil {
+		repo, _ = cache.GetRepoMap(c, user)
+	}
+
+	ticker := time.NewTicker(pingPeriod)
+	eventc := make(chan []byte, 10)
+
+	ctx, cancel := context.WithCancel(
+		context.Background(),
+	)
+	defer func() {
+		cancel()
+		ticker.Stop()
+		close(eventc)
+		logrus.Debugf("Successfully closing websocket")
+	}()
+
+	go func() {
+		// TODO remove this from global config
+		config.pubsub.Subscribe(c, "topic/events", func(m pubsub.Message) {
+			name := m.Labels["repo"]
+			priv := m.Labels["private"]
+			if repo[name] || priv == "false" {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					eventc <- m.Data
+				}
+			}
+		})
+		cancel()
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case buf, ok := <-eventc:
+				if ok {
+					ws.SetWriteDeadline(time.Now().Add(writeWait))
+					ws.WriteMessage(websocket.TextMessage, buf)
+				}
+			case <-ticker.C:
+				err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait))
+				if err != nil {
+					return
+				}
+			}
+		}
+	}()
+	reader(ws)
+}