package main

import (
    "fmt"
    "net"
    "io"
    "log"
    "bufio"
)

func ListenAndServe(address string) {
    // 綁定監(jiān)聽(tīng)地址
    listener, err := net.Listen("tcp", address)
    if err != nil {
        log.Fatal(fmt.Sprintf("listen err: %v", err))
    }
    defer listener.Close()
    log.Println(fmt.Sprintf("bind: %s, start listening...", address))

    for {
        // Accept 會(huì)一直阻塞直到有新的連接建立或者listen中斷才會(huì)返回
        conn, err := listener.Accept()
        if err != nil {
            // 通常是由于listener被關(guān)閉無(wú)法繼續(xù)監(jiān)聽(tīng)導(dǎo)致的錯(cuò)誤
            log.Fatal(fmt.Sprintf("accept err: %v", err))
        }
        // 開(kāi)啟新的 goroutine 處理該連接
        go Handle(conn)
    }
}

func Handle(conn net.Conn) {
    // 使用 bufio 標(biāo)準(zhǔn)庫(kù)提供的緩沖區(qū)功能
    reader := bufio.NewReader(conn)
    for {
        // ReadString 會(huì)一直阻塞直到遇到分隔符 '\n'
        // 遇到分隔符后會(huì)返回上次遇到分隔符或連接建立后收到的所有數(shù)據(jù), 包括分隔符本身
        // 若在遇到分隔符之前遇到異常, ReadString 會(huì)返回已收到的數(shù)據(jù)和錯(cuò)誤信息
        msg, err := reader.ReadString('\n')
        if err != nil {
            // 通常遇到的錯(cuò)誤是連接中斷或被關(guān)閉，用io.EOF表示
            if err == io.EOF {
                log.Println("connection close")
            } else {
                log.Println(err)
            }
            return
        }
        b := []byte(msg)
        // 將收到的信息發(fā)送給客戶端
        conn.Write(b)
    }
}

func main() {
    ListenAndServe(":8000")
}

$ telnet 127.0.0.1 8000
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
> a
a
> b
b
Connection closed by foreign host.

// handler 是應(yīng)用層服務(wù)器的抽象
type Handler interface {
    Handle(ctx context.Context, conn net.Conn)
    Close()error
}

// 監(jiān)聽(tīng)并提供服務(wù)，并在收到 closeChan 發(fā)來(lái)的關(guān)閉通知后關(guān)閉
func ListenAndServe(listener net.Listener, handler tcp.Handler, closeChan <-chan struct{}) {
    // 監(jiān)聽(tīng)關(guān)閉通知
    go func() {
        <-closeChan
        logger.Info("shutting down...")
        // 停止監(jiān)聽(tīng)，listener.Accept()會(huì)立即返回 io.EOF
        _ = listener.Close()
        // 關(guān)閉應(yīng)用層服務(wù)器 
        _ = handler.Close()
    }()

    // 在異常退出后釋放資源
    defer func() {
        // close during unexpected error
        _ = listener.Close()
        _ = handler.Close()
    }()
    ctx := context.Background()
    var waitDone sync.WaitGroup
    for {
        // 監(jiān)聽(tīng)端口, 阻塞直到收到新連接或者出現(xiàn)錯(cuò)誤
        conn, err := listener.Accept()
        if err != nil {
            break
        }
        // 開(kāi)啟 goroutine 來(lái)處理新連接
        logger.Info("accept link")
        waitDone.Add(1)
        go func() {
            defer func() {
                waitDone.Done()
            }()
            handler.Handle(ctx, conn)
        }()
    }
    waitDone.Wait()
}

// ListenAndServeWithSignal 監(jiān)聽(tīng)中斷信號(hào)并通過(guò) closeChan 通知服務(wù)器關(guān)閉
func ListenAndServeWithSignal(cfg *Config, handler tcp.Handler) error {
    closeChan := make(chan struct{})
    sigCh := make(chan os.Signal)
    signal.Notify(sigCh, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
    go func() {
        sig := <-sigCh
        switch sig {
        case syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT:
            closeChan <- struct{}{}
        }
    }()
    listener, err := net.Listen("tcp", cfg.Address)
    if err != nil {
        return err
    }
    logger.Info(fmt.Sprintf("bind: %s, start listening...", cfg.Address))
    ListenAndServe(listener, handler, closeChan)
    return nil
}

// 客戶端連接的抽象
type Client struct {
    // tcp 連接
    Conn net.Conn
    // 當(dāng)服務(wù)端開(kāi)始發(fā)送數(shù)據(jù)時(shí)進(jìn)入waiting, 阻止其它goroutine關(guān)閉連接
    // wait.Wait是作者編寫(xiě)的帶有最大等待時(shí)間的封裝: 
    // https://github.com/HDT3213/godis/blob/master/src/lib/sync/wait/wait.go
    Waiting wait.Wait
}

type EchoHandler struct {
    
    // 保存所有工作狀態(tài)client的集合(把map當(dāng)set用)
    // 需使用并發(fā)安全的容器
    activeConn sync.Map 

    // 關(guān)閉狀態(tài)標(biāo)識(shí)位
    closing atomic.AtomicBool
}

func MakeEchoHandler()(*EchoHandler) {
    return &EchoHandler{}
}

func (h *EchoHandler)Handle(ctx context.Context, conn net.Conn) {
    // 關(guān)閉中的 handler 不會(huì)處理新連接
    if h.closing.Get() {
        conn.Close()
        return 
    }

    client := &Client {
        Conn: conn,
    }
    h.activeConn.Store(client, struct{}{}) // 記住仍然存活的連接

    reader := bufio.NewReader(conn)
    for {
        msg, err := reader.ReadString('\n')
        if err != nil {
            if err == io.EOF {
                logger.Info("connection close")
                h.activeConn.Delete(client)
            } else {
                logger.Warn(err)
            }
            return
        }
        // 發(fā)送數(shù)據(jù)前先置為waiting狀態(tài)，阻止連接被關(guān)閉
        client.Waiting.Add(1)

        // 模擬關(guān)閉時(shí)未完成發(fā)送的情況
        //logger.Info("sleeping")
        //time.Sleep(10 * time.Second)

        b := []byte(msg)
        conn.Write(b)
        // 發(fā)送完畢, 結(jié)束waiting
        client.Waiting.Done()
    }
}

// 關(guān)閉客戶端連接
func (c *Client)Close()error {
    // 等待數(shù)據(jù)發(fā)送完成或超時(shí)
    c.Waiting.WaitWithTimeout(10 * time.Second)
    c.Conn.Close()
    return nil
}

// 關(guān)閉服務(wù)器
func (h *EchoHandler)Close()error {
    logger.Info("handler shutting down...")
    h.closing.Set(true)
    // 逐個(gè)關(guān)閉連接
    h.activeConn.Range(func(key interface{}, val interface{})bool {
        client := key.(*Client)
        client.Close()
        return true
    })
    return nil
}