1/*
  2
  3The remote package provides the pieces to allow Ginkgo test suites to report to remote listeners.
  4This is used, primarily, to enable streaming parallel test output but has, in principal, broader applications (e.g. streaming test output to a browser).
  5
  6*/
  7
  8package remote
  9
 10import (
 11	"encoding/json"
 12	"io/ioutil"
 13	"net"
 14	"net/http"
 15	"sync"
 16
 17	"github.com/onsi/ginkgo/internal/spec_iterator"
 18
 19	"github.com/onsi/ginkgo/config"
 20	"github.com/onsi/ginkgo/reporters"
 21	"github.com/onsi/ginkgo/types"
 22)
 23
 24/*
 25Server spins up on an automatically selected port and listens for communication from the forwarding reporter.
 26It then forwards that communication to attached reporters.
 27*/
 28type Server struct {
 29	listener        net.Listener
 30	reporters       []reporters.Reporter
 31	alives          []func() bool
 32	lock            *sync.Mutex
 33	beforeSuiteData types.RemoteBeforeSuiteData
 34	parallelTotal   int
 35	counter         int
 36}
 37
 38//Create a new server, automatically selecting a port
 39func NewServer(parallelTotal int) (*Server, error) {
 40	listener, err := net.Listen("tcp", "127.0.0.1:0")
 41	if err != nil {
 42		return nil, err
 43	}
 44	return &Server{
 45		listener:        listener,
 46		lock:            &sync.Mutex{},
 47		alives:          make([]func() bool, parallelTotal),
 48		beforeSuiteData: types.RemoteBeforeSuiteData{Data: nil, State: types.RemoteBeforeSuiteStatePending},
 49		parallelTotal:   parallelTotal,
 50	}, nil
 51}
 52
 53//Start the server.  You don't need to `go s.Start()`, just `s.Start()`
 54func (server *Server) Start() {
 55	httpServer := &http.Server{}
 56	mux := http.NewServeMux()
 57	httpServer.Handler = mux
 58
 59	//streaming endpoints
 60	mux.HandleFunc("/SpecSuiteWillBegin", server.specSuiteWillBegin)
 61	mux.HandleFunc("/BeforeSuiteDidRun", server.beforeSuiteDidRun)
 62	mux.HandleFunc("/AfterSuiteDidRun", server.afterSuiteDidRun)
 63	mux.HandleFunc("/SpecWillRun", server.specWillRun)
 64	mux.HandleFunc("/SpecDidComplete", server.specDidComplete)
 65	mux.HandleFunc("/SpecSuiteDidEnd", server.specSuiteDidEnd)
 66
 67	//synchronization endpoints
 68	mux.HandleFunc("/BeforeSuiteState", server.handleBeforeSuiteState)
 69	mux.HandleFunc("/RemoteAfterSuiteData", server.handleRemoteAfterSuiteData)
 70	mux.HandleFunc("/counter", server.handleCounter)
 71	mux.HandleFunc("/has-counter", server.handleHasCounter) //for backward compatibility
 72
 73	go httpServer.Serve(server.listener)
 74}
 75
 76//Stop the server
 77func (server *Server) Close() {
 78	server.listener.Close()
 79}
 80
 81//The address the server can be reached it.  Pass this into the `ForwardingReporter`.
 82func (server *Server) Address() string {
 83	return "http://" + server.listener.Addr().String()
 84}
 85
 86//
 87// Streaming Endpoints
 88//
 89
 90//The server will forward all received messages to Ginkgo reporters registered with `RegisterReporters`
 91func (server *Server) readAll(request *http.Request) []byte {
 92	defer request.Body.Close()
 93	body, _ := ioutil.ReadAll(request.Body)
 94	return body
 95}
 96
 97func (server *Server) RegisterReporters(reporters ...reporters.Reporter) {
 98	server.reporters = reporters
 99}
100
101func (server *Server) specSuiteWillBegin(writer http.ResponseWriter, request *http.Request) {
102	body := server.readAll(request)
103
104	var data struct {
105		Config  config.GinkgoConfigType `json:"config"`
106		Summary *types.SuiteSummary     `json:"suite-summary"`
107	}
108
109	json.Unmarshal(body, &data)
110
111	for _, reporter := range server.reporters {
112		reporter.SpecSuiteWillBegin(data.Config, data.Summary)
113	}
114}
115
116func (server *Server) beforeSuiteDidRun(writer http.ResponseWriter, request *http.Request) {
117	body := server.readAll(request)
118	var setupSummary *types.SetupSummary
119	json.Unmarshal(body, &setupSummary)
120
121	for _, reporter := range server.reporters {
122		reporter.BeforeSuiteDidRun(setupSummary)
123	}
124}
125
126func (server *Server) afterSuiteDidRun(writer http.ResponseWriter, request *http.Request) {
127	body := server.readAll(request)
128	var setupSummary *types.SetupSummary
129	json.Unmarshal(body, &setupSummary)
130
131	for _, reporter := range server.reporters {
132		reporter.AfterSuiteDidRun(setupSummary)
133	}
134}
135
136func (server *Server) specWillRun(writer http.ResponseWriter, request *http.Request) {
137	body := server.readAll(request)
138	var specSummary *types.SpecSummary
139	json.Unmarshal(body, &specSummary)
140
141	for _, reporter := range server.reporters {
142		reporter.SpecWillRun(specSummary)
143	}
144}
145
146func (server *Server) specDidComplete(writer http.ResponseWriter, request *http.Request) {
147	body := server.readAll(request)
148	var specSummary *types.SpecSummary
149	json.Unmarshal(body, &specSummary)
150
151	for _, reporter := range server.reporters {
152		reporter.SpecDidComplete(specSummary)
153	}
154}
155
156func (server *Server) specSuiteDidEnd(writer http.ResponseWriter, request *http.Request) {
157	body := server.readAll(request)
158	var suiteSummary *types.SuiteSummary
159	json.Unmarshal(body, &suiteSummary)
160
161	for _, reporter := range server.reporters {
162		reporter.SpecSuiteDidEnd(suiteSummary)
163	}
164}
165
166//
167// Synchronization Endpoints
168//
169
170func (server *Server) RegisterAlive(node int, alive func() bool) {
171	server.lock.Lock()
172	defer server.lock.Unlock()
173	server.alives[node-1] = alive
174}
175
176func (server *Server) nodeIsAlive(node int) bool {
177	server.lock.Lock()
178	defer server.lock.Unlock()
179	alive := server.alives[node-1]
180	if alive == nil {
181		return true
182	}
183	return alive()
184}
185
186func (server *Server) handleBeforeSuiteState(writer http.ResponseWriter, request *http.Request) {
187	if request.Method == "POST" {
188		dec := json.NewDecoder(request.Body)
189		dec.Decode(&(server.beforeSuiteData))
190	} else {
191		beforeSuiteData := server.beforeSuiteData
192		if beforeSuiteData.State == types.RemoteBeforeSuiteStatePending && !server.nodeIsAlive(1) {
193			beforeSuiteData.State = types.RemoteBeforeSuiteStateDisappeared
194		}
195		enc := json.NewEncoder(writer)
196		enc.Encode(beforeSuiteData)
197	}
198}
199
200func (server *Server) handleRemoteAfterSuiteData(writer http.ResponseWriter, request *http.Request) {
201	afterSuiteData := types.RemoteAfterSuiteData{
202		CanRun: true,
203	}
204	for i := 2; i <= server.parallelTotal; i++ {
205		afterSuiteData.CanRun = afterSuiteData.CanRun && !server.nodeIsAlive(i)
206	}
207
208	enc := json.NewEncoder(writer)
209	enc.Encode(afterSuiteData)
210}
211
212func (server *Server) handleCounter(writer http.ResponseWriter, request *http.Request) {
213	c := spec_iterator.Counter{}
214	server.lock.Lock()
215	c.Index = server.counter
216	server.counter++
217	server.lock.Unlock()
218
219	json.NewEncoder(writer).Encode(c)
220}
221
222func (server *Server) handleHasCounter(writer http.ResponseWriter, request *http.Request) {
223	writer.Write([]byte(""))
224}