server-go/web/fasthttp/server.go

151 lines
2.9 KiB
Go
Raw Normal View History

2018-04-05 15:15:29 +00:00
package fasthttp
2018-04-04 04:01:26 +00:00
import (
"context"
"fmt"
"net"
"sync"
2018-04-06 01:18:19 +00:00
"sync/atomic"
2018-04-04 04:01:26 +00:00
"git.loafle.net/commons/logging-go"
"git.loafle.net/commons/server-go"
"github.com/valyala/fasthttp"
)
type Server struct {
ServerHandler ServerHandler
2018-04-04 13:28:35 +00:00
ctx server.ServerCtx
stopChan chan struct{}
stopWg sync.WaitGroup
2018-04-05 15:15:29 +00:00
hs *fasthttp.Server
2018-04-04 04:01:26 +00:00
}
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{})
2018-04-04 13:28:35 +00:00
2018-04-04 04:01:26 +00:00
s.stopWg.Add(1)
return s.handleServer(listener)
}
func (s *Server) Shutdown(ctx context.Context) error {
if s.stopChan == nil {
2018-04-04 13:28:35 +00:00
return fmt.Errorf("server must be started before stopping it")
2018-04-04 04:01:26 +00:00
}
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 (
2018-04-06 01:18:19 +00:00
err error
stopping atomic.Value
2018-04-04 04:01:26 +00:00
)
defer func() {
if nil != listener {
listener.Close()
}
s.ServerHandler.OnStop(s.ctx)
logging.Logger().Infof(s.serverMessage("Stopped"))
2018-04-06 01:18:19 +00:00
s.stopWg.Done()
2018-04-04 04:01:26 +00:00
}()
if err = s.ServerHandler.OnStart(s.ctx); nil != err {
return err
}
2018-04-06 01:18:19 +00:00
hsCloseChan := make(chan error)
go func() {
if err := s.hs.Serve(listener); nil != err {
if nil == stopping.Load() {
hsCloseChan <- err
return
}
}
hsCloseChan <- nil
}()
2018-04-04 04:01:26 +00:00
logging.Logger().Infof(s.serverMessage("Started"))
select {
2018-04-06 01:18:19 +00:00
case err, _ := <-hsCloseChan:
if nil != err {
return err
}
2018-04-04 04:01:26 +00:00
case <-s.stopChan:
2018-04-06 01:18:19 +00:00
stopping.Store(true)
2018-04-04 04:01:26 +00:00
listener.Close()
2018-04-06 01:18:19 +00:00
<-hsCloseChan
2018-04-04 04:01:26 +00:00
listener = nil
}
return nil
}
func (s *Server) httpHandler(ctx *fasthttp.RequestCtx) {
var (
servlet Servlet
err error
)
2018-04-04 16:51:34 +00:00
if servlet = s.ServerHandler.(ServerHandler).Servlet(s.ctx, ctx); nil == servlet {
2018-04-04 04:01:26 +00:00
s.onError(ctx, fasthttp.StatusInternalServerError, err)
return
}
}
func (s *Server) onError(ctx *fasthttp.RequestCtx, status int, reason error) {
2018-04-04 13:28:35 +00:00
s.ServerHandler.(ServerHandler).OnError(s.ctx, ctx, status, reason)
2018-04-04 04:01:26 +00:00
}