153 lines
2.8 KiB
Go
153 lines
2.8 KiB
Go
package server
|
|
|
|
import (
|
|
"fmt"
|
|
"net"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"git.loafle.net/commons_go/logging"
|
|
)
|
|
|
|
func New(sh ServerHandler) Server {
|
|
s := &server{
|
|
sh: sh,
|
|
}
|
|
return s
|
|
}
|
|
|
|
type Server interface {
|
|
Start() error
|
|
Stop()
|
|
Context() ServerContext
|
|
}
|
|
|
|
type server struct {
|
|
ctx ServerContext
|
|
|
|
sh ServerHandler
|
|
|
|
listener net.Listener
|
|
|
|
Stats ServerStats
|
|
|
|
stopChan chan struct{}
|
|
stopWg sync.WaitGroup
|
|
}
|
|
|
|
func (s *server) Start() error {
|
|
if nil == s.sh {
|
|
panic("Server: server handler must be specified.")
|
|
}
|
|
s.sh.Validate()
|
|
|
|
if s.stopChan != nil {
|
|
panic("Server: server is already running. Stop it before starting it again")
|
|
}
|
|
var err error
|
|
s.ctx = s.sh.ServerContext()
|
|
|
|
if err = s.sh.Init(s.ctx); nil != err {
|
|
logging.Logger().Panic(fmt.Sprintf("Server: Initialization of server has been failed %v", err))
|
|
}
|
|
|
|
if s.listener, err = s.sh.Listen(s.ctx); nil != err {
|
|
return err
|
|
}
|
|
|
|
s.stopChan = make(chan struct{})
|
|
|
|
s.stopWg.Add(1)
|
|
go handleServer(s)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *server) Stop() {
|
|
if s.stopChan == nil {
|
|
panic("Server: server must be started before stopping it")
|
|
}
|
|
close(s.stopChan)
|
|
s.stopWg.Wait()
|
|
s.stopChan = nil
|
|
|
|
s.sh.OnStop()
|
|
|
|
logging.Logger().Info(fmt.Sprintf("Server[%s] is stopped", s.sh.GetName()))
|
|
}
|
|
|
|
func handleServer(s *server) {
|
|
defer s.stopWg.Done()
|
|
|
|
logging.Logger().Info(fmt.Sprintf("Server[%s] is started", s.sh.GetName()))
|
|
s.sh.OnStart(s.ctx)
|
|
|
|
var conn net.Conn
|
|
var err error
|
|
var stopping atomic.Value
|
|
|
|
for {
|
|
acceptChan := make(chan struct{})
|
|
go func() {
|
|
if conn, err = s.listener.Accept(); err != nil {
|
|
if stopping.Load() == nil {
|
|
logging.Logger().Error(fmt.Sprintf("Server: Cannot accept new connection: [%s]", err))
|
|
}
|
|
}
|
|
close(acceptChan)
|
|
}()
|
|
|
|
select {
|
|
case <-s.stopChan:
|
|
stopping.Store(true)
|
|
s.listener.Close()
|
|
<-acceptChan
|
|
return
|
|
case <-acceptChan:
|
|
s.Stats.incAcceptCalls()
|
|
}
|
|
|
|
if err != nil {
|
|
s.Stats.incAcceptErrors()
|
|
select {
|
|
case <-s.stopChan:
|
|
return
|
|
case <-time.After(time.Second):
|
|
}
|
|
continue
|
|
}
|
|
|
|
s.stopWg.Add(1)
|
|
go handleConnection(s, conn)
|
|
}
|
|
}
|
|
|
|
func handleConnection(s *server, conn net.Conn) {
|
|
defer s.stopWg.Done()
|
|
|
|
var err error
|
|
if conn, err = s.sh.OnConnect(conn); nil != err {
|
|
logging.Logger().Error(fmt.Sprintf("Server: connecting[%s] failed %v", conn.RemoteAddr(), err))
|
|
return
|
|
}
|
|
|
|
logging.Logger().Debug(fmt.Sprintf("Server: Client[%s] is connected.", conn.RemoteAddr()))
|
|
|
|
clientStopChan := make(chan struct{})
|
|
handleDoneChan := make(chan struct{})
|
|
|
|
go s.sh.Handle(conn, clientStopChan, handleDoneChan)
|
|
|
|
select {
|
|
case <-s.stopChan:
|
|
close(clientStopChan)
|
|
conn.Close()
|
|
<-handleDoneChan
|
|
case <-handleDoneChan:
|
|
close(clientStopChan)
|
|
conn.Close()
|
|
logging.Logger().Debug(fmt.Sprintf("Server: Client[%s] is disconnected.", conn.RemoteAddr()))
|
|
}
|
|
}
|