Queue: Make WorkerPools and Queues flushable (#10001)

* Make WorkerPools and Queues flushable

Adds Flush methods to Queues and the WorkerPool
Further abstracts the WorkerPool
Adds a final step to Flush the queues in the defer from PrintCurrentTest
Fixes an issue with Settings inheritance in queues

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Change to for loop

* Add IsEmpty and begin just making the queues composed WorkerPools

* subsume workerpool into the queues and create a flushable interface

* Add manager command

* Move flushall to queue.Manager and add to testlogger

* As per @guillep2k

* as per @guillep2k

* Just make queues all implement flushable and clean up the wrapped queue flushes

* cope with no timeout

Co-authored-by: Lauris BH <lauris@nix.lv>

1 files changed, 83 insertions, 0 deletions

diff --git a/modules/private/manager.go b/modules/private/manager.go
new file mode 100644
index 0000000000..de46c7ab2c
--- /dev/null
+++ b/modules/private/manager.go
@@ -0,0 +1,83 @@
+// Copyright 2020 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package private
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"code.gitea.io/gitea/modules/setting"
+)
+
+// Shutdown calls the internal shutdown function
+func Shutdown() (int, string) {
+	reqURL := setting.LocalURL + fmt.Sprintf("api/internal/manager/shutdown")
+
+	req := newInternalRequest(reqURL, "POST")
+	resp, err := req.Response()
+	if err != nil {
+		return http.StatusInternalServerError, fmt.Sprintf("Unable to contact gitea: %v", err.Error())
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return resp.StatusCode, decodeJSONError(resp).Err
+	}
+
+	return http.StatusOK, "Shutting down"
+}
+
+// Restart calls the internal restart function
+func Restart() (int, string) {
+	reqURL := setting.LocalURL + fmt.Sprintf("api/internal/manager/restart")
+
+	req := newInternalRequest(reqURL, "POST")
+	resp, err := req.Response()
+	if err != nil {
+		return http.StatusInternalServerError, fmt.Sprintf("Unable to contact gitea: %v", err.Error())
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return resp.StatusCode, decodeJSONError(resp).Err
+	}
+
+	return http.StatusOK, "Restarting"
+}
+
+// FlushOptions represents the options for the flush call
+type FlushOptions struct {
+	Timeout     time.Duration
+	NonBlocking bool
+}
+
+// FlushQueues calls the internal flush-queues function
+func FlushQueues(timeout time.Duration, nonBlocking bool) (int, string) {
+	reqURL := setting.LocalURL + fmt.Sprintf("api/internal/manager/flush-queues")
+
+	req := newInternalRequest(reqURL, "POST")
+	if timeout > 0 {
+		req.SetTimeout(timeout+10*time.Second, timeout+10*time.Second)
+	}
+	req = req.Header("Content-Type", "application/json")
+	jsonBytes, _ := json.Marshal(FlushOptions{
+		Timeout:     timeout,
+		NonBlocking: nonBlocking,
+	})
+	req.Body(jsonBytes)
+	resp, err := req.Response()
+	if err != nil {
+		return http.StatusInternalServerError, fmt.Sprintf("Unable to contact gitea: %v", err.Error())
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return resp.StatusCode, decodeJSONError(resp).Err
+	}
+
+	return http.StatusOK, "Flushed"
+}