wip

cmd/mistryd/server.go

@@ -41,12 +41,16 @@ type Server struct {
 	// matches a job id.
 	// The stored map type is [string]bool.
 	tq *sync.Map
+
+	fs http.FileSystem
 }
 
 // NewServer accepts a non-nil configuration and an optional logger, and
 // returns a new Server.
 // If logger is nil, server logs are disabled.
 func NewServer(cfg *Config, logger *log.Logger) (*Server, error) {
+	var err error
+
 	if cfg == nil {
 		return nil, errors.New("config cannot be nil")
 	}
@@ -58,12 +62,12 @@ func NewServer(cfg *Config, logger *log.Logger) (*Server, error) {
 	s := new(Server)
 	mux := http.NewServeMux()
 
-	statikFS, err := fs.New()
+	s.fs, err = fs.New()
 	if err != nil {
 		logger.Fatal(err)
 	}
 
-	mux.Handle("/", http.StripPrefix("/", http.FileServer(statikFS)))
+	mux.Handle("/", http.StripPrefix("/", http.FileServer(s.fs)))
 	mux.HandleFunc("/jobs", s.HandleNewJob)
 	mux.HandleFunc("/index/", s.HandleIndex)
 	mux.HandleFunc("/job/", s.HandleShowJob)
@@ -76,7 +80,6 @@ func NewServer(cfg *Config, logger *log.Logger) (*Server, error) {
 	s.pq = NewProjectQueue()
 	s.br = broker.NewBroker(s.Log)
 	s.tq = new(sync.Map)
-	go s.br.ListenForClients()
 	return s, nil
 }
 
@@ -275,52 +278,90 @@ func (s *Server) HandleShowJob(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if state == "ready" {
+		_, ok := s.tq.Load(id)
+		if ok {
+			fmt.Println("before")
+			s.br.CloseClientC[id] <- struct{}{}
+			fmt.Println("after")
+		}
+	}
+
 	if state == "pending" {
 		// For each job id there is only one tailer responsible for
 		// emitting the read bytes to the s.br.Notifier channel.
-		hasTail, ok := s.tq.Load(id)
-		if !ok || hasTail.(bool) == false {
+		_, ok := s.tq.Load(id)
+		if !ok {
 			// Mark the id to the tailers' queue to identify that a
 			// tail reader has been spawned.
 			s.tq.Store(id, true)
 			// Create a channel to communicate the closure of all connections
 			// for the job id to the spawned tailer goroutine.
 			if _, ok := s.br.CloseClientC[id]; !ok {
-				s.br.CloseClientC[id] = make(chan struct{}, 1)
+				s.br.CloseClientC[id] = make(chan struct{})
 			}
 			// Spawns a tailer which tails the build log file and communicates
 			// the read results to the s.br.Notifier channel.
 			go func() {
+				defer func() {
+					log.Println("exiting go")
+				}()
 				s.Log.Printf("[Tailer] Starting for: %s", id)
 				tl, err := tailer.New(buildLogPath)
 				if err != nil {
 					s.Log.Print(err)
 					w.WriteHeader(http.StatusInternalServerError)
 					return
 				}
-				defer tl.Close()
+				defer func() {
+					<-s.br.CloseClientC[id]
+					s.Log.Printf("[Tailer] Exiting for: %s", id)
+					err = tl.Close()
+					if err != nil {
+						// TODO Handle
+					}
+					s.tq.Delete(id)
+				}()
 				scanner := bufio.NewScanner(tl)
 				for scanner.Scan() {
+					fmt.Println("for is true")
 					select {
 					case <-s.br.CloseClientC[id]:
-						s.Log.Printf("[Tailer] Exiting for: %s", id)
-						s.tq.Store(id, false)
+						fmt.Println("closed")
 						return
 					default:
+						fmt.Println("before br.notifier write")
 						s.br.Notifier <- &broker.Event{Msg: []byte(scanner.Text()), ID: id}
+						fmt.Println("after br.notifier write")
 					}
 				}
+				fmt.Println("after for")
 			}()
 		}
 	}
 
-	t, err := template.ParseFiles("./public/templates/show.html")
+	f, err := s.fs.Open("/templates/show.html")
+	if err != nil {
+		s.Log.Print(err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	tmplBody, err := ioutil.ReadAll(f)
+	if err != nil {
+		s.Log.Print(err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	tmpl := template.New("jobshow")
+	tmpl, err = tmpl.Parse(string(tmplBody))
 	if err != nil {
 		s.Log.Print(err)
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
-	t.Execute(w, ji)
+	tmpl.Execute(w, ji)
 }
 
 // HandleServerPush handles the server push logic.
@@ -393,5 +434,6 @@ func (s *Server) HandleServerPush(w http.ResponseWriter, r *http.Request) {
 // non-nil error.
 func (s *Server) ListenAndServe() error {
 	s.Log.Printf("Configuration: %#v", s.cfg)
+	go s.br.ListenForClients()
 	return s.srv.ListenAndServe()
 }

cmd/mistryd/testdata/projects/simple/docker-entrypoint.sh

@@ -1,6 +1,14 @@
 #!/bin/bash
 set -e
 
+echo hi
+sleep 3
+echo hi
+sleep 3
+echo hi
+sleep 3
+echo hi
+
 touch artifacts/out.txt
 
 exit 0

pkg/broker/broker.go

@@ -1,6 +1,7 @@
 package broker
 
 import (
+	"fmt"
 	"log"
 	"sync"
 )
@@ -93,9 +94,12 @@ func (br *Broker) ListenForClients() {
 			br.Log.Printf("[Broker] Removed client. %d registered clients", len(br.clients))
 			cc, ok = br.clientsCount.Load(sb.ID)
 			if ok && cc.(int) == 0 {
+				fmt.Println("zxczxc")
 				br.CloseClientC[sb.ID] <- struct{}{}
+				fmt.Println("asdasd")
 			}
 		case event := <-br.Notifier:
+			fmt.Println("reading br.Notifier")
 			for client, _ := range br.clients {
 				if client.ID == event.ID {
 					client.EventC <- event.Msg

pkg/tailer/tailer.go

@@ -34,6 +34,8 @@ func (t *Tailer) Read(b []byte) (int, error) {
 			return n, nil
 		} else if err != io.EOF {
 			return n, err
+		} else {
+
 		}
 	}
 }