notify/server.go

package notify

import (
	"b612.me/stario"
	"context"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type ServerCommon struct {
	msgID                          uint64
	alive                          atomic.Value
	status                         Status
	sessionOwnerState              atomic.Int32
	sessionRuntime                 atomic.Pointer[serverSessionRuntime]
	listener                       net.Listener
	udpListener                    *net.UDPConn
	queue                          *stario.StarQueue
	stopFn                         context.CancelFunc
	stopCtx                        context.Context
	maxReadTimeout                 time.Duration
	maxWriteTimeout                time.Duration
	replyWriteTimeout              atomic.Int64
	parallelNum                    int
	wg                             stario.WaitGroup
	peerRegistry                   *serverPeerRegistry
	mu                             sync.RWMutex
	handshakeRsaKey                []byte
	SecretKey                      []byte
	defaultMsgEn                   func([]byte, []byte) []byte
	defaultMsgDe                   func([]byte, []byte) []byte
	defaultFastStreamEncode        transportFastStreamEncoder
	defaultFastBulkEncode          transportFastBulkEncoder
	defaultFastPlainEncode         transportFastPlainEncoder
	defaultModernPSKRuntime        *modernPSKCodecRuntime
	peerAttachSecurity             atomic.Pointer[peerAttachSecurityState]
	securityBootstrap              transportProtectionProfile
	securitySteady                 transportProtectionProfile
	securityAuthMode               AuthMode
	securityProtectionMode         ProtectionMode
	securityRequireForwardSecrecy  bool
	securityConfigured             bool
	peerAttachReplay               peerAttachReplayCache
	peerAttachExplicitCount        atomic.Int64
	peerAttachAuthFallbackCount    atomic.Int64
	peerAttachAuthRejectCount      atomic.Int64
	peerAttachDowngradeRejectCount atomic.Int64
	peerAttachBindingRejectCount   atomic.Int64
	linkMu                         sync.RWMutex
	linkFns                        map[string]func(message *Message)
	defaultFns                     func(message *Message)
	noFinSyncMsgMaxKeepSeconds     int64
	maxHeartbeatLostSeconds        int64
	sequenceDe                     func([]byte) (interface{}, error)
	sequenceEn                     func(interface{}) ([]byte, error)
	logicalSession                 *logicalSessionState
	onFileEvent                    func(FileEvent)
	fileEventObserver              func(FileEvent)
	fileTransferCfg                fileTransferConfig
	signalReliableCfg              signalReliabilityConfig
	streamRuntime                  *streamRuntime
	recordRuntime                  *recordRuntime
	bulkRuntime                    *bulkRuntime
	bulkRecovery                   *bulkRecoveryQueue
	bulkRecoveryMu                 sync.Mutex
	bulkOpenTuning                 BulkOpenTuning
	udpWriteGateOnce               sync.Once
	udpWriteGate                   chan struct{}
	bulkDedicatedSidecarMu         sync.Mutex
	bulkDedicatedSidecars          map[*LogicalConn]map[uint32]*bulkDedicatedSidecar
	connectionRetryState           *connectionRetryState
	detachedClientKeepSeconds      int64
	securityReadyCheck             bool
	showError                      bool
	debugMode                      bool
}

func NewServer() Server {
	transport := defaultModernPSKTransportBundle()
	var server ServerCommon
	server.wg = stario.NewWaitGroup(0)
	server.parallelNum = 0
	server.noFinSyncMsgMaxKeepSeconds = 0
	server.maxHeartbeatLostSeconds = 300
	server.replyWriteTimeout.Store(int64(defaultMessageReplyWriteTimeout))
	server.stopCtx, server.stopFn = context.WithCancel(context.Background())
	server.SecretKey = nil
	server.handshakeRsaKey = defaultRsaKey
	server.peerRegistry = newServerPeerRegistry()
	server.defaultMsgEn = transport.msgEn
	server.defaultMsgDe = transport.msgDe
	server.defaultFastStreamEncode = transport.fastStreamEncode
	server.defaultFastBulkEncode = transport.fastBulkEncode
	server.defaultFastPlainEncode = transport.fastPlainEncode
	server.securityReadyCheck = true
	server.sequenceEn = encode
	server.sequenceDe = Decode
	server.alive.Store(false)
	server.linkFns = make(map[string]func(*Message))
	server.fileTransferCfg = defaultFileTransferConfig()
	server.signalReliableCfg = defaultSignalReliabilityConfig()
	server.logicalSession = newLogicalSessionState(server.fileTransferCfg, server.signalReliableCfg)
	server.streamRuntime = newStreamRuntime("sstrm")
	server.recordRuntime = newRecordRuntime()
	server.bulkRuntime = newBulkRuntime("sblk")
	server.bulkRecovery = newBulkRecoveryQueue(server.reportBulkRecoveryError)
	server.bulkOpenTuning = defaultBulkOpenTuning()
	server.bulkDedicatedSidecars = make(map[*LogicalConn]map[uint32]*bulkDedicatedSidecar)
	server.connectionRetryState = newConnectionRetryState()
	server.onFileEvent = normalizeFileEventCallback(nil)
	server.fileEventObserver = normalizeFileEventCallback(nil)
	server.defaultFns = func(message *Message) {
		return
	}
	server.setServerDefaultTransportProtectionProfile(defaultTransportProtectionProfile())
	server.peerAttachSecurity.Store(defaultPeerAttachSecurityState())
	server.sessionRuntime.Store(newServerSessionRuntimeBase(server.stopCtx, server.stopFn))
	bindServerStreamControl(&server)
	bindServerBulkControl(&server)
	server.getTransferState().setBuiltinHandler(server.builtinFileTransferHandler)
	return &server
}