refactor server code for better testing
This commit is contained in:
parent
de675ba718
commit
31e20b242e
@ -60,6 +60,7 @@ func getTask(wrk *worker.Worker) http.HandlerFunc {
|
|||||||
http.Error(writer, "encoding", http.StatusInternalServerError)
|
http.Error(writer, "encoding", http.StatusInternalServerError)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
writer.Header().Set("X-Correlation-Id", requestID)
|
||||||
writer.Header().Set("Content-Type", "application/json")
|
writer.Header().Set("Content-Type", "application/json")
|
||||||
writer.Header().Set("Content-Length", strconv.Itoa(len(data)))
|
writer.Header().Set("Content-Length", strconv.Itoa(len(data)))
|
||||||
writer.Header().Set("Content-Version", strconv.Itoa(len(info.Attempts)))
|
writer.Header().Set("Content-Version", strconv.Itoa(len(info.Attempts)))
|
||||||
|
@ -8,10 +8,11 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"sync"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gopkg.in/yaml.v2"
|
"gopkg.in/yaml.v2"
|
||||||
|
|
||||||
|
"nano-run/worker"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Config struct {
|
type Config struct {
|
||||||
@ -74,58 +75,62 @@ func (cfg Config) SaveFile(file string) error {
|
|||||||
return ioutil.WriteFile(file, data, 0600)
|
return ioutil.WriteFile(file, data, 0600)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (cfg Config) Run(global context.Context) error {
|
func (cfg Config) Create(global context.Context) (*Server, error) {
|
||||||
units, err := Units(cfg.ConfigDirectory)
|
units, err := Units(cfg.ConfigDirectory)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return nil, err
|
||||||
}
|
}
|
||||||
workers, err := Workers(cfg.WorkingDirectory, units)
|
workers, err := Workers(cfg.WorkingDirectory, units)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return nil, err
|
||||||
}
|
}
|
||||||
defer func() {
|
ctx, cancel := context.WithCancel(global)
|
||||||
for _, wrk := range workers {
|
srv := &Server{
|
||||||
wrk.Close()
|
Handler: Handler(units, workers),
|
||||||
}
|
workers: workers,
|
||||||
}()
|
units: units,
|
||||||
handler := Handler(units, workers)
|
done: make(chan struct{}),
|
||||||
|
cancel: cancel,
|
||||||
|
}
|
||||||
|
go srv.run(ctx)
|
||||||
|
return srv, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (cfg Config) Run(global context.Context) error {
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(global)
|
ctx, cancel := context.WithCancel(global)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
srv, err := cfg.Create(global)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer srv.Close()
|
||||||
|
|
||||||
server := http.Server{
|
server := http.Server{
|
||||||
Addr: cfg.Bind,
|
Addr: cfg.Bind,
|
||||||
Handler: handler,
|
Handler: srv,
|
||||||
}
|
}
|
||||||
|
|
||||||
var wg sync.WaitGroup
|
done := make(chan struct{})
|
||||||
|
|
||||||
wg.Add(1)
|
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
|
||||||
defer cancel()
|
defer cancel()
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
t, c := context.WithTimeout(context.Background(), cfg.GracefulShutdown)
|
t, c := context.WithTimeout(context.Background(), cfg.GracefulShutdown)
|
||||||
_ = server.Shutdown(t)
|
_ = server.Shutdown(t)
|
||||||
c()
|
c()
|
||||||
|
close(done)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
wg.Add(1)
|
|
||||||
go func() {
|
|
||||||
defer wg.Done()
|
|
||||||
defer cancel()
|
|
||||||
err := Run(ctx, workers)
|
|
||||||
if err != nil {
|
|
||||||
log.Println("workers stopped:", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
if cfg.TLS.Enable {
|
if cfg.TLS.Enable {
|
||||||
err = server.ListenAndServeTLS(cfg.TLS.Cert, cfg.TLS.Key)
|
err = server.ListenAndServeTLS(cfg.TLS.Cert, cfg.TLS.Key)
|
||||||
} else {
|
} else {
|
||||||
err = server.ListenAndServe()
|
err = server.ListenAndServe()
|
||||||
}
|
}
|
||||||
cancel()
|
cancel()
|
||||||
wg.Wait()
|
<-done
|
||||||
return err
|
return ctx.Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
func limitRequest(maxSize int64, handler http.Handler) http.Handler {
|
func limitRequest(maxSize int64, handler http.Handler) http.Handler {
|
||||||
@ -142,3 +147,33 @@ func limitRequest(maxSize int64, handler http.Handler) http.Handler {
|
|||||||
handler.ServeHTTP(writer, request)
|
handler.ServeHTTP(writer, request)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type Server struct {
|
||||||
|
http.Handler
|
||||||
|
workers []*worker.Worker
|
||||||
|
units []Unit
|
||||||
|
cancel func()
|
||||||
|
done chan struct{}
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (srv *Server) Close() {
|
||||||
|
for _, wrk := range srv.workers {
|
||||||
|
wrk.Close()
|
||||||
|
}
|
||||||
|
srv.cancel()
|
||||||
|
<-srv.done
|
||||||
|
}
|
||||||
|
|
||||||
|
func (srv *Server) Err() error {
|
||||||
|
return srv.err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (srv *Server) run(ctx context.Context) {
|
||||||
|
err := Run(ctx, srv.workers)
|
||||||
|
if err != nil {
|
||||||
|
log.Println("workers stopped:", err)
|
||||||
|
}
|
||||||
|
srv.err = err
|
||||||
|
close(srv.done)
|
||||||
|
}
|
||||||
|
Loading…
x
Reference in New Issue
Block a user