Package Socket/Server:

- fix UDP as kernel SOCKGRAM
- add unixgram as kernel SOCKGRAM for local socket file
- optimize parallel statment
- optimize code / minor bugs fix
- add IsRunning function
- add timeout shutdown based on periodicly check of IsRunning
- fix close channel before send empty struct to stop process
- optimize clean shutdown
- optimize connection process to use less resources and lock pipeline less time

socket/server/tcp/error.go

@@ -30,8 +30,9 @@ import "fmt"
 
 var (
 	ErrInvalidAddress  = fmt.Errorf("invalid listen address")
-	ErrInvalidHostName = fmt.Errorf("invalid server host name")
-	ErrInvalidHostPort = fmt.Errorf("invalid server host port")
 	ErrContextClosed   = fmt.Errorf("context closed")
 	ErrServerClosed    = fmt.Errorf("server closed")
+	ErrInvalidHandler  = fmt.Errorf("invalid handler")
+	ErrShutdownTimeout = fmt.Errorf("timeout on stopping socket")
+	ErrInvalidInstance = fmt.Errorf("invalid socket instance")
 )

socket/server/tcp/inerface.go

@@ -57,6 +57,7 @@ func New(h libsck.Handler, sizeBuffRead libsiz.Size) ServerTcp {
 		h:  f,
 		c:  c,
 		s:  s,
+		r:  new(atomic.Bool),
 		fe: new(atomic.Value),
 		fi: new(atomic.Value),
 		fs: new(atomic.Value),

socket/server/tcp/listener.go

@@ -33,6 +33,7 @@ import (
 	"crypto/tls"
 	"io"
 	"net"
+	"sync/atomic"
 
 	libptc "github.com/nabbar/golib/network/protocol"
 	libsck "github.com/nabbar/golib/socket"
@@ -80,44 +81,67 @@ func (o *srv) Listen(ctx context.Context) error {
 		e error
 		l net.Listener
 		a = o.getAddress()
+		s = new(atomic.Bool)
 	)
 
-	var fctClose = func() {
-		o.fctInfoSrv("closing listen socket '%s %s'", libptc.NetworkTCP.String(), a)
-		if l != nil {
-			o.fctError(l.Close())
-		}
-	}
-
-	defer fctClose()
-
 	if len(a) == 0 {
 		return ErrInvalidAddress
+	} else if hdl := o.handler(); hdl == nil {
+		return ErrInvalidHandler
 	} else if l, e = o.getListen(a); e != nil {
 		o.fctError(e)
 		return e
 	}
 
+	var fctClose = func() {
+		o.fctInfoSrv("closing listen socket '%s %s'", libptc.NetworkTCP.String(), a)
+
+		if l != nil {
+			_ = l.Close()
+		}
+
+		o.r.Store(false)
+	}
+
+	defer fctClose()
+	s.Store(false)
+
+	go func() {
+		<-ctx.Done()
+		go func() {
+			_ = o.Shutdown()
+		}()
+		return
+	}()
+
+	go func() {
+		<-o.Done()
+
+		e = nil
+		s.Store(true)
+
+		if l != nil {
+			o.fctError(l.Close())
+		}
+
+		return
+	}()
+
+	o.r.Store(true)
 	// Accept new connection or stop if context or shutdown trigger
 	for {
-		select {
-		case <-ctx.Done():
-			return ErrContextClosed
-		case <-o.Done():
-			return nil
-		default:
-			// Accept an incoming connection.
-			if l == nil {
-				return ErrServerClosed
-			}
+		// Accept an incoming connection.
+		if l == nil {
+			return ErrServerClosed
+		} else if s.Load() {
+			return e
+		}
 
-			co, ce := l.Accept()
-
-			if ce != nil {
-				o.fctError(ce)
-			} else {
-				go o.Conn(co)
-			}
+		if co, ce := l.Accept(); ce != nil && !s.Load() {
+			o.fctError(ce)
+		} else {
+			o.fctInfo(co.LocalAddr(), co.RemoteAddr(), libsck.ConnectionNew)
+			go o.Conn(co)
 		}
 	}
 }
@@ -128,12 +152,10 @@ func (o *srv) Conn(conn net.Conn) {
 		_ = conn.Close()
 	}()
 
-	o.fctInfo(conn.LocalAddr(), conn.RemoteAddr(), libsck.ConnectionNew)
-
 	var (
 		err error
 		rdr = bufio.NewReaderSize(conn, o.buffSize())
-		buf []byte
+		msg []byte
 		hdl libsck.Handler
 	)
 
@@ -142,7 +164,7 @@ func (o *srv) Conn(conn net.Conn) {
 	}
 
 	for {
-		buf, err = rdr.ReadBytes('\n')
+		msg, err = rdr.ReadBytes('\n')
 
 		o.fctInfo(conn.LocalAddr(), conn.RemoteAddr(), libsck.ConnectionRead)
 		if err != nil {
@@ -152,7 +174,13 @@ func (o *srv) Conn(conn net.Conn) {
 			break
 		}
 
+		var buf = bytes.NewBuffer(msg)
+
+		if !bytes.HasSuffix(msg, []byte{libsck.EOL}) {
+			buf.Write([]byte{libsck.EOL})
+		}
+
 		o.fctInfo(conn.LocalAddr(), conn.RemoteAddr(), libsck.ConnectionHandler)
-		hdl(bytes.NewBuffer(buf), conn)
+		hdl(buf, conn)
 	}
 }

socket/server/tcp/model.go

@@ -27,10 +27,12 @@
 package tcp
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
 	"net"
 	"sync/atomic"
+	"time"
 
 	libptc "github.com/nabbar/golib/network/protocol"
 
@@ -58,6 +60,7 @@ type srv struct {
 	h *atomic.Value // handler
 	c *atomic.Value // chan []byte
 	s *atomic.Value // chan struct{}
+	r *atomic.Bool  // is Running
 
 	fe *atomic.Value // function error
 	fi *atomic.Value // function info
@@ -67,6 +70,10 @@ type srv struct {
 	ad *atomic.Value // Server address url
 }
 
+func (o *srv) IsRunning() bool {
+	return o.r.Load()
+}
+
 func (o *srv) Done() <-chan struct{} {
 	s := o.s.Load()
 	if s != nil {
@@ -76,14 +83,47 @@ func (o *srv) Done() <-chan struct{} {
 	return closedChanStruct
 }
 
-func (o *srv) Shutdown() {
+func (o *srv) Close() error {
+	return o.Shutdown()
+}
+
+func (o *srv) Shutdown() error {
 	if o == nil {
-		return
+		return ErrInvalidInstance
 	}
 
 	s := o.s.Load()
 	if s != nil {
-		o.s.Store(nil)
+		if c, k := s.(chan struct{}); k {
+			c <- struct{}{}
+			o.s.Store(c)
+		}
+	}
+
+	var (
+		tck      = time.NewTicker(100 * time.Millisecond)
+		ctx, cnl = context.WithTimeout(context.Background(), 25*time.Second)
+	)
+
+	defer func() {
+		if s != nil {
+			o.s.Store(closedChanStruct)
+		}
+
+		tck.Stop()
+		cnl()
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ErrShutdownTimeout
+		case <-tck.C:
+			if o.IsRunning() {
+				continue
+			}
+			return nil
+		}
 	}
 }