153 lines
3.0 KiB
Go
153 lines
3.0 KiB
Go
package fasthttp
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"sync"
|
|
"sync/atomic"
|
|
|
|
"git.loafle.net/commons/logging-go"
|
|
"git.loafle.net/commons/server-go"
|
|
"github.com/valyala/fasthttp"
|
|
)
|
|
|
|
type Server struct {
|
|
ServerHandler ServerHandler
|
|
|
|
ctx server.ServerCtx
|
|
stopChan chan struct{}
|
|
stopWg sync.WaitGroup
|
|
|
|
hs *fasthttp.Server
|
|
}
|
|
|
|
func (s *Server) ListenAndServe() error {
|
|
var (
|
|
err error
|
|
listener net.Listener
|
|
)
|
|
if nil == s.ServerHandler {
|
|
return fmt.Errorf("Server: server handler must be specified")
|
|
}
|
|
s.ServerHandler.Validate()
|
|
|
|
if s.stopChan != nil {
|
|
return fmt.Errorf(s.serverMessage("already running. Stop it before starting it again"))
|
|
}
|
|
|
|
s.ctx = s.ServerHandler.ServerCtx()
|
|
if nil == s.ctx {
|
|
return fmt.Errorf(s.serverMessage("ServerCtx is nil"))
|
|
}
|
|
|
|
s.hs = &fasthttp.Server{
|
|
Handler: s.httpHandler,
|
|
Name: s.ServerHandler.GetName(),
|
|
Concurrency: s.ServerHandler.GetConcurrency(),
|
|
ReadBufferSize: s.ServerHandler.GetReadBufferSize(),
|
|
WriteBufferSize: s.ServerHandler.GetWriteBufferSize(),
|
|
ReadTimeout: s.ServerHandler.GetReadTimeout(),
|
|
WriteTimeout: s.ServerHandler.GetWriteTimeout(),
|
|
}
|
|
|
|
if err = s.ServerHandler.Init(s.ctx); nil != err {
|
|
return err
|
|
}
|
|
|
|
if listener, err = s.ServerHandler.Listener(s.ctx); nil != err {
|
|
return err
|
|
}
|
|
|
|
s.stopChan = make(chan struct{})
|
|
|
|
s.stopWg.Add(1)
|
|
return s.handleServer(listener)
|
|
}
|
|
|
|
func (s *Server) Shutdown(ctx context.Context) error {
|
|
if s.stopChan == nil {
|
|
return fmt.Errorf("server must be started before stopping it")
|
|
}
|
|
close(s.stopChan)
|
|
s.stopWg.Wait()
|
|
|
|
s.ServerHandler.Destroy(s.ctx)
|
|
|
|
s.stopChan = nil
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Server) serverMessage(msg string) string {
|
|
return fmt.Sprintf("Server[%s]: %s", s.ServerHandler.GetName(), msg)
|
|
}
|
|
|
|
func (s *Server) handleServer(listener net.Listener) error {
|
|
var (
|
|
err error
|
|
stopping atomic.Value
|
|
)
|
|
|
|
defer func() {
|
|
if nil != listener {
|
|
listener.Close()
|
|
}
|
|
s.ServerHandler.OnStop(s.ctx)
|
|
|
|
logging.Logger().Infof(s.serverMessage("Stopped"))
|
|
|
|
s.stopWg.Done()
|
|
}()
|
|
|
|
if err = s.ServerHandler.OnStart(s.ctx); nil != err {
|
|
return err
|
|
}
|
|
|
|
hsCloseChan := make(chan error)
|
|
go func() {
|
|
if err := s.hs.Serve(listener); nil != err {
|
|
if nil == stopping.Load() {
|
|
hsCloseChan <- err
|
|
return
|
|
}
|
|
}
|
|
hsCloseChan <- nil
|
|
}()
|
|
|
|
logging.Logger().Infof(s.serverMessage("Started"))
|
|
|
|
select {
|
|
case err, _ := <-hsCloseChan:
|
|
if nil != err {
|
|
return err
|
|
}
|
|
case <-s.stopChan:
|
|
stopping.Store(true)
|
|
listener.Close()
|
|
<-hsCloseChan
|
|
listener = nil
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Server) httpHandler(ctx *fasthttp.RequestCtx) {
|
|
var (
|
|
servlet Servlet
|
|
err error
|
|
)
|
|
|
|
if servlet = s.ServerHandler.(ServerHandler).Servlet(s.ctx, ctx); nil == servlet {
|
|
s.onError(ctx, fasthttp.StatusNotFound, err)
|
|
return
|
|
}
|
|
servletCtx := servlet.ServletCtx(s.ctx)
|
|
|
|
servlet.Handle(servletCtx, ctx)
|
|
}
|
|
|
|
func (s *Server) onError(ctx *fasthttp.RequestCtx, status int, reason error) {
|
|
s.ServerHandler.(ServerHandler).OnError(s.ctx, ctx, status, reason)
|
|
}
|