<commit_before>// Copyright 2015-2018 HenryLee. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package tp

import (
	"context"
	"crypto/tls"
	"net"
	"strings"
	"sync"
	"time"

	"github.com/henrylee2cn/teleport/quic"

	"github.com/henrylee2cn/goutil"
	"github.com/henrylee2cn/goutil/coarsetime"
	"github.com/henrylee2cn/goutil/errors"
	"github.com/henrylee2cn/teleport/codec"
)

type (
	// BasePeer peer with the common method set
	BasePeer interface {
		// Close closes peer.
		Close() (err error)
		// CountSession returns the number of sessions.
		CountSession() int
		// GetSession gets the session by id.
		GetSession(sessionID string) (Session, bool)
		// RangeSession ranges all sessions. If fn returns false, stop traversing.
		RangeSession(fn func(sess Session) bool)
		// SetTLSConfig sets the TLS config.
		SetTLSConfig(tlsConfig *tls.Config)
		// SetTLSConfigFromFile sets the TLS config from file.
		SetTLSConfigFromFile(tlsCertFile, tlsKeyFile string, insecureSkipVerifyForClient ...bool) error
		// TLSConfig returns the TLS config.
		TLSConfig() *tls.Config
		// PluginContainer returns the global plugin container.
		PluginContainer() *PluginContainer
	}
	// EarlyPeer the communication peer that has just been created
	EarlyPeer interface {
		BasePeer
		// Router returns the root router of call or push handlers.
		Router() *Router
		// SubRoute adds handler group.
		SubRoute(pathPrefix string, plugin ...Plugin) *SubRouter
		// RouteCall registers CALL handlers, and returns the paths.
		RouteCall(ctrlStruct interface{}, plugin ...Plugin) []string
		// RouteCallFunc registers CALL handler, and returns the path.
		RouteCallFunc(callHandleFunc interface{}, plugin ...Plugin) string
		// RoutePush registers PUSH handlers, and returns the paths.
		RoutePush(ctrlStruct interface{}, plugin ...Plugin) []string
		// RoutePushFunc registers PUSH handler, and returns the path.
		RoutePushFunc(pushHandleFunc interface{}, plugin ...Plugin) string
		// SetUnknownCall sets the default handler, which is called when no handler for CALL is found.
		SetUnknownCall(fn func(UnknownCallCtx) (interface{}, *Status), plugin ...Plugin)
		// SetUnknownPush sets the default handler, which is called when no handler for PUSH is found.
		SetUnknownPush(fn func(UnknownPushCtx) *Status, plugin ...Plugin)
	}
	// Peer the communication peer which is server or client role
	Peer interface {
		EarlyPeer
		// ListenAndServe turns on the listening service.
		ListenAndServe(protoFunc ...ProtoFunc) error
		// Dial connects with the peer of the destination address.
		Dial(addr string, protoFunc ...ProtoFunc) (Session, *Status)
		// ServeConn serves the connection and returns a session.
		// NOTE:
		//  Not support automatically redials after disconnection;
		//  Execute the PostAcceptPlugin plugins.
		ServeConn(conn net.Conn, protoFunc ...ProtoFunc) (Session, *Status)
	}
)

var (
	_ BasePeer  = new(peer)
	_ EarlyPeer = new(peer)
	_ Peer      = new(peer)
)

type peer struct {
	router            *Router
	pluginContainer   *PluginContainer
	sessHub           *SessionHub
	closeCh           chan struct{}
	defaultSessionAge time.Duration // Default session max age, if less than or equal to 0, no time limit
	defaultContextAge time.Duration // Default CALL or PUSH context max age, if less than or equal to 0, no time limit
	tlsConfig         *tls.Config
	slowCometDuration time.Duration
	timeNow           func() int64
	mu                sync.Mutex
	network           string
	defaultBodyCodec  byte
	printDetail       bool
	countTime         bool

	// only for server role
	listenAddr string
	listeners  map[net.Listener]struct{}

	// only for client role
	defaultDialTimeout time.Duration
	redialInterval     time.Duration
	localAddr          net.Addr
	redialTimes        int32
}

// NewPeer creates a new peer.
func NewPeer(cfg PeerConfig, globalLeftPlugin ...Plugin) Peer {
	doPrintPid()
	pluginContainer := newPluginContainer()
	pluginContainer.AppendLeft(globalLeftPlugin...)
	pluginContainer.preNewPeer(&cfg)
	if err := cfg.check(); err != nil {
		Fatalf("%v", err)
	}

	var p = &peer{
		router:             newRouter(pluginContainer),
		pluginContainer:    pluginContainer,
		sessHub:            newSessionHub(),
		defaultSessionAge:  cfg.DefaultSessionAge,
		defaultContextAge:  cfg.DefaultContextAge,
		closeCh:            make(chan struct{}),
		slowCometDuration:  cfg.slowCometDuration,
		defaultDialTimeout: cfg.DefaultDialTimeout,
		redialInterval:     cfg.RedialInterval,
		network:            cfg.Network,
		listenAddr:         cfg.listenAddrStr,
		localAddr:          cfg.localAddr,
		printDetail:        cfg.PrintDetail,
		countTime:          cfg.CountTime,
		redialTimes:        cfg.RedialTimes,
		listeners:          make(map[net.Listener]struct{}),
	}

	if c, err := codec.GetByName(cfg.DefaultBodyCodec); err != nil {
		Fatalf("%v", err)
	} else {
		p.defaultBodyCodec = c.ID()
	}
	if p.countTime {
		p.timeNow = func() int64 { return time.Now().UnixNano() }
	} else {
		p.timeNow = func() int64 { return 0 }
	}
	addPeer(p)
	p.pluginContainer.postNewPeer(p)
	return p
}

// PluginContainer returns the global plugin container.
func (p *peer) PluginContainer() *PluginContainer {
	return p.pluginContainer
}

// TLSConfig returns the TLS config.
func (p *peer) TLSConfig() *tls.Config {
	return p.tlsConfig
}

// SetTLSConfig sets the TLS config.
func (p *peer) SetTLSConfig(tlsConfig *tls.Config) {
	p.tlsConfig = tlsConfig
}

// SetTLSConfigFromFile sets the TLS config from file.
func (p *peer) SetTLSConfigFromFile(tlsCertFile, tlsKeyFile string, insecureSkipVerifyForClient ...bool) error {
	var err error
	p.tlsConfig, err = NewTLSConfigFromFile(tlsCertFile, tlsKeyFile, insecureSkipVerifyForClient...)
	return err
}

// GetSession gets the session by id.
func (p *peer) GetSession(sessionID string) (Session, bool) {
	return p.sessHub.get(sessionID)
}

// RangeSession ranges all sessions.
// If fn returns false, stop traversing.
func (p *peer) RangeSession(fn func(sess Session) bool) {
	p.sessHub.sessions.Range(func(key, value interface{}) bool {
		return fn(value.(*session))
	})
}

// CountSession returns the number of sessions.
func (p *peer) CountSession() int {
	return p.sessHub.len()
}

// Dial connects with the peer of the destination address.
func (p *peer) Dial(addr string, protoFunc ...ProtoFunc) (Session, *Status) {
	return p.newSessionForClient(func() (net.Conn, error) {
		if p.network == "quic" {
			ctx := context.Background()
			if p.defaultDialTimeout > 0 {
				ctx, _ = context.WithTimeout(ctx, p.defaultDialTimeout)
			}
			if p.tlsConfig == nil {
				return quic.DialAddrContext(ctx, addr, GenerateTLSConfigForClient(), nil)
			}
			return quic.DialAddrContext(ctx, addr, p.tlsConfig, nil)
		}
		d := &net.Dialer{
			LocalAddr: p.localAddr,
			Timeout:   p.defaultDialTimeout,
		}
		if p.tlsConfig != nil {
			return tls.DialWithDialer(d, p.network, addr, p.tlsConfig)
		}
		return d.Dial(p.network, addr)
	}, addr, protoFunc)
}

type redialTimes int32

func (p *peer) newRedialTimes() *redialTimes {
	r := redialTimes(p.redialTimes)
	return &r
}

func (r *redialTimes) next() bool {
	t := *r
	if t == 0 {
		return false
	}
	if t > 0 {
		*r--
	}
	return true
}

func (p *peer) newSessionForClient(dialFunc func() (net.Conn, error), addr string, protoFuncs []ProtoFunc) (*session, *Status) {
	conn, dialErr := dialFunc()
	if dialErr != nil {
		redialTimes := p.newRedialTimes()
		for redialTimes.next() {
			time.Sleep(p.redialInterval)
			Debugf("trying to redial... (network:%s, addr:%s)", p.network, addr)
			conn, dialErr = dialFunc()
			if dialErr == nil {
				break
			}
		}
	}
	if dialErr != nil {
		return nil, statDialFailed.Copy(dialErr)
	}
	var sess = newSession(p, conn, protoFuncs)

	// create redial func
	if p.redialTimes != 0 {
		sess.redialForClientLocked = func() bool {
			var stat *Status
			redialTimes := p.newRedialTimes()
			for redialTimes.next() {
				time.Sleep(p.redialInterval)
				Debugf("trying to redial... (network:%s, addr:%s, id:%s)", p.network, sess.RemoteAddr().String(), sess.ID())
				stat = p.renewSessionForClientLocked(sess, dialFunc, addr, protoFuncs)
				if stat.OK() {
					Infof("redial ok (network:%s, addr:%s, id:%s)", p.network, sess.RemoteAddr().String(), sess.ID())
					return true
				}
			}
			sess.tryChangeStatus(statusRedialFailed, statusRedialing)
			if !stat.OK() {
				Errorf("redial fail (network:%s, addr:%s, id:%s): %s", p.network, sess.RemoteAddr().String(), sess.ID(), stat.String())
			}
			return false
		}
	}

	sess.socket.SetID(sess.LocalAddr().String())
	if stat := p.pluginContainer.postDial(sess, false); !stat.OK() {
		sess.Close()
		return nil, stat
	}
	Infof("dial ok (network:%s, addr:%s, id:%s)", p.network, sess.RemoteAddr().String(), sess.ID())
	sess.changeStatus(statusOk)
	AnywayGo(sess.startReadAndHandle)
	p.sessHub.set(sess)
	return sess, nil
}

func (p *peer) renewSessionForClientLocked(sess *session, dialFunc func() (net.Conn, error), addr string, protoFuncs []ProtoFunc) *Status {
	var conn, dialErr = dialFunc()
	if dialErr != nil {
		return statDialFailed.Copy(dialErr)
	}
	if p.tlsConfig != nil {
		conn = tls.Client(conn, p.tlsConfig)
	}
	oldIP := sess.LocalAddr().String()
	oldID := sess.ID()
	oldConn := sess.getConn()
	if oldConn != nil {
		oldConn.Close()
	}
	sess.socket.Reset(conn, protoFuncs...)
	if oldIP == oldID {
		sess.socket.SetID(sess.LocalAddr().String())
	} else {
		sess.socket.SetID(oldID)
	}
	sess.changeStatus(statusPreparing)
	if stat := p.pluginContainer.postDial(sess, true); !stat.OK() {
		sess.closeLocked()
		return stat
	}
	sess.changeStatus(statusOk)
	AnywayGo(sess.startReadAndHandle)
	p.sessHub.set(sess)
	return nil
}

// ServeConn serves the connection and returns a session.
// NOTE:
//  Not support automatically redials after disconnection;
//  Execute the PostAcceptPlugin plugins.
func (p *peer) ServeConn(conn net.Conn, protoFunc ...ProtoFunc) (Session, *Status) {
	network := conn.LocalAddr().Network()
	if strings.Contains(network, "udp") {
		if _, ok := conn.(*quic.Conn); !ok {
			return nil, NewStatus(CodeWrongConn, "Not support UDP", "network must be one of the following: tcp, tcp4, tcp6, unix, unixpacket or quic")
		}
		network = "quic"
	} else if p.tlsConfig != nil {
		// Make sure to use TLS
		if _, ok := conn.(*tls.Conn); !ok {
			conn = tls.Server(conn, p.tlsConfig)
		}
	}
	var sess = newSession(p, conn, protoFunc)
	if stat := p.pluginContainer.postAccept(sess); !stat.OK() {
		sess.Close()
		return nil, stat
	}
	Infof("serve ok (network:%s, addr:%s, id:%s)", network, sess.RemoteAddr().String(), sess.ID())
	sess.changeStatus(statusOk)
	AnywayGo(sess.startReadAndHandle)
	p.sessHub.set(sess)
	return sess, nil
}

// ErrListenClosed listener is closed error.
var ErrListenClosed = errors.New("listener is closed")

// serveListener serves the listener.
// NOTE: The caller ensures that the listener supports graceful shutdown.
func (p *peer) serveListener(lis net.Listener, protoFunc ...ProtoFunc) error {
	defer lis.Close()
	p.listeners[lis] = struct{}{}

	network := lis.Addr().Network()
	if _, ok := lis.(*quic.Listener); ok {
		network = "quic"
	}
	addr := lis.Addr().String()
	Printf("listen and serve (network:%s, addr:%s)", network, addr)
<commit_msg>chore: peer.ServeConn do not check TLS<commit_after>// Copyright 2015-2018 HenryLee. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package tp

import (
	"context"
	"crypto/tls"
	"net"
	"strings"
	"sync"
	"time"

	"github.com/henrylee2cn/teleport/quic"

	"github.com/henrylee2cn/goutil"
	"github.com/henrylee2cn/goutil/coarsetime"
	"github.com/henrylee2cn/goutil/errors"
	"github.com/henrylee2cn/teleport/codec"
)

type (
	// BasePeer peer with the common method set
	BasePeer interface {
		// Close closes peer.
		Close() (err error)
		// CountSession returns the number of sessions.
		CountSession() int
		// GetSession gets the session by id.
		GetSession(sessionID string) (Session, bool)
		// RangeSession ranges all sessions. If fn returns false, stop traversing.
		RangeSession(fn func(sess Session) bool)
		// SetTLSConfig sets the TLS config.
		SetTLSConfig(tlsConfig *tls.Config)
		// SetTLSConfigFromFile sets the TLS config from file.
		SetTLSConfigFromFile(tlsCertFile, tlsKeyFile string, insecureSkipVerifyForClient ...bool) error
		// TLSConfig returns the TLS config.
		TLSConfig() *tls.Config
		// PluginContainer returns the global plugin container.
		PluginContainer() *PluginContainer
	}
	// EarlyPeer the communication peer that has just been created
	EarlyPeer interface {
		BasePeer
		// Router returns the root router of call or push handlers.
		Router() *Router
		// SubRoute adds handler group.
		SubRoute(pathPrefix string, plugin ...Plugin) *SubRouter
		// RouteCall registers CALL handlers, and returns the paths.
		RouteCall(ctrlStruct interface{}, plugin ...Plugin) []string
		// RouteCallFunc registers CALL handler, and returns the path.
		RouteCallFunc(callHandleFunc interface{}, plugin ...Plugin) string
		// RoutePush registers PUSH handlers, and returns the paths.
		RoutePush(ctrlStruct interface{}, plugin ...Plugin) []string
		// RoutePushFunc registers PUSH handler, and returns the path.
		RoutePushFunc(pushHandleFunc interface{}, plugin ...Plugin) string
		// SetUnknownCall sets the default handler, which is called when no handler for CALL is found.
		SetUnknownCall(fn func(UnknownCallCtx) (interface{}, *Status), plugin ...Plugin)
		// SetUnknownPush sets the default handler, which is called when no handler for PUSH is found.
		SetUnknownPush(fn func(UnknownPushCtx) *Status, plugin ...Plugin)
	}
	// Peer the communication peer which is server or client role
	Peer interface {
		EarlyPeer
		// ListenAndServe turns on the listening service.
		ListenAndServe(protoFunc ...ProtoFunc) error
		// Dial connects with the peer of the destination address.
		Dial(addr string, protoFunc ...ProtoFunc) (Session, *Status)
		// ServeConn serves the connection and returns a session.
		// NOTE:
		//  Not support automatically redials after disconnection;
		//  Not check TLS;
		//  Execute the PostAcceptPlugin plugins.
		ServeConn(conn net.Conn, protoFunc ...ProtoFunc) (Session, *Status)
	}
)

var (
	_ BasePeer  = new(peer)
	_ EarlyPeer = new(peer)
	_ Peer      = new(peer)
)

type peer struct {
	router            *Router
	pluginContainer   *PluginContainer
	sessHub           *SessionHub
	closeCh           chan struct{}
	defaultSessionAge time.Duration // Default session max age, if less than or equal to 0, no time limit
	defaultContextAge time.Duration // Default CALL or PUSH context max age, if less than or equal to 0, no time limit
	tlsConfig         *tls.Config
	slowCometDuration time.Duration
	timeNow           func() int64
	mu                sync.Mutex
	network           string
	defaultBodyCodec  byte
	printDetail       bool
	countTime         bool

	// only for server role
	listenAddr string
	listeners  map[net.Listener]struct{}

	// only for client role
	defaultDialTimeout time.Duration
	redialInterval     time.Duration
	localAddr          net.Addr
	redialTimes        int32
}

// NewPeer creates a new peer.
func NewPeer(cfg PeerConfig, globalLeftPlugin ...Plugin) Peer {
	doPrintPid()
	pluginContainer := newPluginContainer()
	pluginContainer.AppendLeft(globalLeftPlugin...)
	pluginContainer.preNewPeer(&cfg)
	if err := cfg.check(); err != nil {
		Fatalf("%v", err)
	}

	var p = &peer{
		router:             newRouter(pluginContainer),
		pluginContainer:    pluginContainer,
		sessHub:            newSessionHub(),
		defaultSessionAge:  cfg.DefaultSessionAge,
		defaultContextAge:  cfg.DefaultContextAge,
		closeCh:            make(chan struct{}),
		slowCometDuration:  cfg.slowCometDuration,
		defaultDialTimeout: cfg.DefaultDialTimeout,
		redialInterval:     cfg.RedialInterval,
		network:            cfg.Network,
		listenAddr:         cfg.listenAddrStr,
		localAddr:          cfg.localAddr,
		printDetail:        cfg.PrintDetail,
		countTime:          cfg.CountTime,
		redialTimes:        cfg.RedialTimes,
		listeners:          make(map[net.Listener]struct{}),
	}

	if c, err := codec.GetByName(cfg.DefaultBodyCodec); err != nil {
		Fatalf("%v", err)
	} else {
		p.defaultBodyCodec = c.ID()
	}
	if p.countTime {
		p.timeNow = func() int64 { return time.Now().UnixNano() }
	} else {
		p.timeNow = func() int64 { return 0 }
	}
	addPeer(p)
	p.pluginContainer.postNewPeer(p)
	return p
}

// PluginContainer returns the global plugin container.
func (p *peer) PluginContainer() *PluginContainer {
	return p.pluginContainer
}

// TLSConfig returns the TLS config.
func (p *peer) TLSConfig() *tls.Config {
	return p.tlsConfig
}

// SetTLSConfig sets the TLS config.
func (p *peer) SetTLSConfig(tlsConfig *tls.Config) {
	p.tlsConfig = tlsConfig
}

// SetTLSConfigFromFile sets the TLS config from file.
func (p *peer) SetTLSConfigFromFile(tlsCertFile, tlsKeyFile string, insecureSkipVerifyForClient ...bool) error {
	var err error
	p.tlsConfig, err = NewTLSConfigFromFile(tlsCertFile, tlsKeyFile, insecureSkipVerifyForClient...)
	return err
}

// GetSession gets the session by id.
func (p *peer) GetSession(sessionID string) (Session, bool) {
	return p.sessHub.get(sessionID)
}

// RangeSession ranges all sessions.
// If fn returns false, stop traversing.
func (p *peer) RangeSession(fn func(sess Session) bool) {
	p.sessHub.sessions.Range(func(key, value interface{}) bool {
		return fn(value.(*session))
	})
}

// CountSession returns the number of sessions.
func (p *peer) CountSession() int {
	return p.sessHub.len()
}

// Dial connects with the peer of the destination address.
func (p *peer) Dial(addr string, protoFunc ...ProtoFunc) (Session, *Status) {
	return p.newSessionForClient(func() (net.Conn, error) {
		if p.network == "quic" {
			ctx := context.Background()
			if p.defaultDialTimeout > 0 {
				ctx, _ = context.WithTimeout(ctx, p.defaultDialTimeout)
			}
			if p.tlsConfig == nil {
				return quic.DialAddrContext(ctx, addr, GenerateTLSConfigForClient(), nil)
			}
			return quic.DialAddrContext(ctx, addr, p.tlsConfig, nil)
		}
		d := &net.Dialer{
			LocalAddr: p.localAddr,
			Timeout:   p.defaultDialTimeout,
		}
		if p.tlsConfig != nil {
			return tls.DialWithDialer(d, p.network, addr, p.tlsConfig)
		}
		return d.Dial(p.network, addr)
	}, addr, protoFunc)
}

type redialTimes int32

func (p *peer) newRedialTimes() *redialTimes {
	r := redialTimes(p.redialTimes)
	return &r
}

func (r *redialTimes) next() bool {
	t := *r
	if t == 0 {
		return false
	}
	if t > 0 {
		*r--
	}
	return true
}

func (p *peer) newSessionForClient(dialFunc func() (net.Conn, error), addr string, protoFuncs []ProtoFunc) (*session, *Status) {
	conn, dialErr := dialFunc()
	if dialErr != nil {
		redialTimes := p.newRedialTimes()
		for redialTimes.next() {
			time.Sleep(p.redialInterval)
			Debugf("trying to redial... (network:%s, addr:%s)", p.network, addr)
			conn, dialErr = dialFunc()
			if dialErr == nil {
				break
			}
		}
	}
	if dialErr != nil {
		return nil, statDialFailed.Copy(dialErr)
	}
	var sess = newSession(p, conn, protoFuncs)

	// create redial func
	if p.redialTimes != 0 {
		sess.redialForClientLocked = func() bool {
			var stat *Status
			redialTimes := p.newRedialTimes()
			for redialTimes.next() {
				time.Sleep(p.redialInterval)
				Debugf("trying to redial... (network:%s, addr:%s, id:%s)", p.network, sess.RemoteAddr().String(), sess.ID())
				stat = p.renewSessionForClientLocked(sess, dialFunc, addr, protoFuncs)
				if stat.OK() {
					Infof("redial ok (network:%s, addr:%s, id:%s)", p.network, sess.RemoteAddr().String(), sess.ID())
					return true
				}
			}
			sess.tryChangeStatus(statusRedialFailed, statusRedialing)
			if !stat.OK() {
				Errorf("redial fail (network:%s, addr:%s, id:%s): %s", p.network, sess.RemoteAddr().String(), sess.ID(), stat.String())
			}
			return false
		}
	}

	sess.socket.SetID(sess.LocalAddr().String())
	if stat := p.pluginContainer.postDial(sess, false); !stat.OK() {
		sess.Close()
		return nil, stat
	}
	Infof("dial ok (network:%s, addr:%s, id:%s)", p.network, sess.RemoteAddr().String(), sess.ID())
	sess.changeStatus(statusOk)
	AnywayGo(sess.startReadAndHandle)
	p.sessHub.set(sess)
	return sess, nil
}

func (p *peer) renewSessionForClientLocked(sess *session, dialFunc func() (net.Conn, error), addr string, protoFuncs []ProtoFunc) *Status {
	var conn, dialErr = dialFunc()
	if dialErr != nil {
		return statDialFailed.Copy(dialErr)
	}
	if p.tlsConfig != nil {
		conn = tls.Client(conn, p.tlsConfig)
	}
	oldIP := sess.LocalAddr().String()
	oldID := sess.ID()
	oldConn := sess.getConn()
	if oldConn != nil {
		oldConn.Close()
	}
	sess.socket.Reset(conn, protoFuncs...)
	if oldIP == oldID {
		sess.socket.SetID(sess.LocalAddr().String())
	} else {
		sess.socket.SetID(oldID)
	}
	sess.changeStatus(statusPreparing)
	if stat := p.pluginContainer.postDial(sess, true); !stat.OK() {
		sess.closeLocked()
		return stat
	}
	sess.changeStatus(statusOk)
	AnywayGo(sess.startReadAndHandle)
	p.sessHub.set(sess)
	return nil
}

// ServeConn serves the connection and returns a session.
// NOTE:
//  Not support automatically redials after disconnection;
//  Not check TLS;
//  Execute the PostAcceptPlugin plugins.
func (p *peer) ServeConn(conn net.Conn, protoFunc ...ProtoFunc) (Session, *Status) {
	network := conn.LocalAddr().Network()
	if strings.Contains(network, "udp") {
		if _, ok := conn.(*quic.Conn); !ok {
			return nil, NewStatus(CodeWrongConn, "Not support UDP", "network must be one of the following: tcp, tcp4, tcp6, unix, unixpacket or quic")
		}
		network = "quic"
	}
	var sess = newSession(p, conn, protoFunc)
	if stat := p.pluginContainer.postAccept(sess); !stat.OK() {
		sess.Close()
		return nil, stat
	}
	Infof("serve ok (network:%s, addr:%s, id:%s)", network, sess.RemoteAddr().String(), sess.ID())
	sess.changeStatus(statusOk)
	AnywayGo(sess.startReadAndHandle)
	p.sessHub.set(sess)
	return sess, nil
}

// ErrListenClosed listener is closed error.
var ErrListenClosed = errors.New("listener is closed")

// serveListener serves the listener.
// NOTE: The caller ensures that the listener supports graceful shutdown.
func (p *peer) serveListener(lis net.Listener, protoFunc ...ProtoFunc) error {
	defer lis.Close()
	p.listeners[lis] = struct{}{}

	network := lis.Addr().Network()
	if _, ok := lis.(*quic.Listener); ok {
		network = "quic"
	}
	addr := lis.Addr().String()
	Printf("listen and serve (network:%s, addr:%s)", network, addr)

{'fraction_non_alphanumeric': 0.12167962939351439, 'fraction_numerical': 0.0030600535509371415, 'mean_word_length': 3.7144860749348827, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '25835', 'n_tokens_mistral': 8144, 'n_tokens_neox': 7581, 'n_words': 2205}