set tcp server as plugin

examples/go.mod

@@ -50,7 +50,7 @@ require (
 	go.osspkg.com/goppy/tcp v0.0.0-00010101000000-000000000000
 	go.osspkg.com/goppy/udp v0.0.2
 	go.osspkg.com/goppy/unixsocket v0.1.0
-	go.osspkg.com/goppy/web v0.1.5
+	go.osspkg.com/goppy/web v0.1.6
 	go.osspkg.com/goppy/ws v0.1.0
 	go.osspkg.com/goppy/xc v0.1.0
 	go.osspkg.com/goppy/xdns v0.1.0
@@ -74,7 +74,7 @@ require (
 	go.osspkg.com/goppy/encryption v0.1.3 // indirect
 	go.osspkg.com/goppy/errors v0.1.0 // indirect
 	go.osspkg.com/goppy/iofile v0.1.3 // indirect
-	go.osspkg.com/goppy/ioutil v0.1.0 // indirect
+	go.osspkg.com/goppy/ioutil v0.1.1 // indirect
 	go.osspkg.com/goppy/orm v0.1.4 // indirect
 	go.osspkg.com/goppy/random v0.1.1 // indirect
 	go.osspkg.com/goppy/sqlcommon v0.1.4 // indirect

examples/goppy/tcp-server/Makefile

@@ -0,0 +1,4 @@
+SHELL=/bin/bash
+
+run:
+	go run main.go --config=config.yaml

examples/goppy/tcp-server/config.yaml

@@ -0,0 +1,7 @@
+env: dev
+level: 4
+log: /dev/stdout
+
+tcp:
+    - address: 0.0.0.0:8080
+      timeout: 5s

examples/goppy/tcp-server/main.go

@@ -0,0 +1,46 @@
+/*
+ *  Copyright (c) 2022-2023 Mikhail Knyazhev <markus621@yandex.ru>. All rights reserved.
+ *  Use of this source code is governed by a BSD 3-Clause license that can be found in the LICENSE file.
+ */
+
+package main
+
+import (
+	"fmt"
+
+	"go.osspkg.com/goppy"
+	"go.osspkg.com/goppy/plugins"
+	"go.osspkg.com/goppy/tcp"
+)
+
+func main() {
+	app := goppy.New()
+	app.Plugins(
+		tcp.WithServer(),
+		plugins.Plugin{
+			Inject: func(server tcp.Server) error {
+				h := &Echo{}
+				server.HandleFunc(h)
+				return nil
+			},
+		},
+	)
+	app.Run()
+}
+
+type Echo struct{}
+
+func (e *Echo) HandlerTCP(w tcp.Response, r tcp.Request) {
+	for {
+		b, err := r.ReadLine()
+		if err != nil {
+			fmt.Println("ERR:", r.Addr().String(), err)
+			return
+		}
+
+		fmt.Println("GET:", r.Addr().String())
+		fmt.Println(string(b))
+
+		fmt.Fprintf(w, "ECHO\n")
+	}
+}

iosync/group.go

@@ -15,27 +15,19 @@ type (
 	}
 
 	_group struct {
-		wg   sync.WaitGroup
-		sync Switch
+		wg sync.WaitGroup
 	}
 )
 
 func NewGroup() Group {
-	return &_group{
-		sync: NewSwitch(),
-	}
+	return &_group{}
 }
 
 func (v *_group) Wait() {
-	v.sync.On()
 	v.wg.Wait()
-	v.sync.Off()
 }
 
 func (v *_group) Background(call func()) {
-	if v.sync.IsOn() {
-		return
-	}
 	v.wg.Add(1)
 	go func() {
 		call()
@@ -44,9 +36,6 @@ func (v *_group) Background(call func()) {
 }
 
 func (v *_group) Run(call func()) {
-	if v.sync.IsOn() {
-		return
-	}
 	v.wg.Add(1)
 	call()
 	v.wg.Done()

tcp/server/common.go → tcp/common.go

@@ -3,19 +3,24 @@
  *  Use of this source code is governed by a BSD 3-Clause license that can be found in the LICENSE file.
  */
 
-package server
+package tcp
 
 import (
 	"net"
 )
 
 type HandlerTCP interface {
-	HandlerTCP(p Processor)
+	HandlerTCP(w Response, r Request)
 }
 
-type Processor interface {
-	Write([]byte) (int, error)
-	ReadLine() ([]byte, error)
-	Read(b []byte) (int, error)
-	Addr() net.Addr
-}
+type (
+	Request interface {
+		ReadLine() ([]byte, error)
+		Read(b []byte) (int, error)
+		Addr() net.Addr
+	}
+
+	Response interface {
+		Write([]byte) (int, error)
+	}
+)

tcp/server/config.go → tcp/config.go

@@ -3,20 +3,16 @@
  *  Use of this source code is governed by a BSD 3-Clause license that can be found in the LICENSE file.
  */
 
-package server
+package tcp
 
 import "time"
 
 type ConfigItem struct {
-	Pools   []Pool        `yaml:"pool"`
+	Address string        `yaml:"address"`
+	Certs   []Cert        `yaml:"certs,omitempty"`
 	Timeout time.Duration `yaml:"timeout,omitempty"`
 }
 
-type Pool struct {
-	Port  int    `yaml:"port"`
-	Certs []Cert `yaml:"certs,omitempty"`
-}
-
 type Cert struct {
 	Public  string `yaml:"pub"`
 	Private string `yaml:"priv"`

tcp/go.mod

@@ -17,6 +17,7 @@ require (
 	go.osspkg.com/goppy/plugins v0.1.1
 	go.osspkg.com/goppy/xc v0.1.0
 	go.osspkg.com/goppy/xlog v0.1.4
+	go.osspkg.com/goppy/xnet v0.1.1
 )
 
 require (

tcp/go.sum

@@ -2,3 +2,5 @@ github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8Hm
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
+go.osspkg.com/goppy/xnet v0.1.1 h1:nysNyS5O7nHXIN/IjP9HGfa6Qh5BTTSYLULijk+Sv9M=
+go.osspkg.com/goppy/xnet v0.1.1/go.mod h1:eB5pFfZTCrcaIOHzt4RlTgBVF5dRUV/u52qz/2hY3qk=

tcp/server/listener.go → tcp/listener.go

@@ -3,24 +3,23 @@
  *  Use of this source code is governed by a BSD 3-Clause license that can be found in the LICENSE file.
  */
 
-package server
+package tcp
 
 import (
 	"crypto/rand"
 	"crypto/tls"
-	"fmt"
 	"net"
-)
 
-type (
-	Listen struct {
-		conn net.Listener
-		tls  bool
-	}
+	"go.osspkg.com/goppy/xnet"
 )
 
-func NewListen(port int, certs ...Cert) (*Listen, error) {
-	address := fmt.Sprintf("0.0.0.0:%d", port)
+type Listen struct {
+	conn net.Listener
+	tls  bool
+}
+
+func NewListen(address string, certs ...Cert) (*Listen, error) {
+	address = xnet.CheckHostPort(address)
 
 	if len(certs) == 0 {
 		l, err := net.Listen("tcp", address)
@@ -53,3 +52,7 @@ func (v *Listen) Close() error {
 func (v *Listen) Accept() (net.Conn, error) {
 	return v.conn.Accept()
 }
+
+func (v *Listen) IsTLS() bool {
+	return v.tls
+}

tcp/plugin_server.go

@@ -8,35 +8,93 @@ package tcp
 import (
 	"time"
 
+	"go.osspkg.com/goppy/errors"
+	"go.osspkg.com/goppy/iosync"
 	"go.osspkg.com/goppy/plugins"
-	"go.osspkg.com/goppy/tcp/server"
+	"go.osspkg.com/goppy/xc"
 	"go.osspkg.com/goppy/xlog"
 )
 
 type Config struct {
-	TCP server.ConfigItem `yaml:"tcp"`
+	TCP []ConfigItem `yaml:"tcp"`
 }
 
 func (v *Config) Default() {
-	if len(v.TCP.Pools) == 0 {
-		v.TCP.Pools = append(v.TCP.Pools, server.Pool{
-			Port: 8080,
-			Certs: []server.Cert{{
+	if len(v.TCP) == 0 {
+		v.TCP = append(v.TCP, ConfigItem{
+			Address: "0.0.0.0:8080",
+			Certs: []Cert{{
 				Public:  "./ssl/public.crt",
 				Private: "./ssl/private.key",
 			}},
+			Timeout: 5 * time.Second,
 		})
 	}
-	if v.TCP.Timeout == 0 {
-		v.TCP.Timeout = 10 * time.Second
-	}
 }
 
+type (
+	Server interface {
+		HandleFunc(h HandlerTCP)
+	}
+
+	serverProvider struct {
+		log   xlog.Logger
+		conf  []ConfigItem
+		servs []*ServerTCP
+		wg    iosync.Group
+	}
+)
+
 func WithServer() plugins.Plugin {
 	return plugins.Plugin{
 		Config: &Config{},
-		Inject: func(c *Config, l xlog.Logger) *server.Server {
-			return server.New(c.TCP, l)
+		Inject: func(c *Config, l xlog.Logger) (*serverProvider, Server) {
+			s := &serverProvider{
+				log:   l,
+				conf:  c.TCP,
+				servs: make([]*ServerTCP, 0, len(c.TCP)),
+				wg:    iosync.NewGroup(),
+			}
+			return s, s
 		},
 	}
 }
+
+func (v *serverProvider) HandleFunc(h HandlerTCP) {
+	for _, serv := range v.servs {
+		serv.HandleFunc(h)
+	}
+}
+
+func (v *serverProvider) Up(ctx xc.Context) error {
+	for _, conf := range v.conf {
+		conf := conf
+		serv := NewServerTCP(conf)
+		v.servs = append(v.servs, serv)
+		v.log.WithFields(xlog.Fields{
+			"addr": conf.Address,
+		}).Infof("TCP server started")
+		v.wg.Background(func() {
+			if err := serv.ListenAndServe(ctx.Context()); err != nil {
+				v.log.WithFields(xlog.Fields{
+					"err": err.Error(), "addr": conf.Address,
+				}).Errorf("TCP server stopped")
+				ctx.Close()
+				return
+			}
+			v.log.WithFields(xlog.Fields{
+				"addr": conf.Address,
+			}).Infof("TCP server stopped")
+		})
+	}
+	return nil
+}
+
+func (v *serverProvider) Down() error {
+	var err error
+	for _, serv := range v.servs {
+		err = errors.Wrap(err, serv.Close())
+	}
+	v.wg.Wait()
+	return err
+}