Logging-System/sink/async.go

package sink

import (
	"errors"
	"sync"
	"sync/atomic"
	"time"

	"logging-system/logsys"
)

// ErrClosed 表示出口已经关闭。
var ErrClosed = errors.New("sink: closed")

// Async 用有界队列把写日志从调用方 goroutine 挪走。
// 队列满时丢弃最新一条并计数（不阻塞调用方）。Close 会排空队列并限时等待。
type Async struct {
	Inner        logsys.Sink
	Queue        int
	FlushTimeout time.Duration

	ch        chan logsys.Entry
	syncCh    chan chan struct{}
	stop      chan struct{}
	done      chan struct{}
	once      sync.Once
	dropped   atomic.Int64
	accepted  atomic.Int64
	closeOnce sync.Once
	closeErr  error
}

// Start 启动后台写者。重复调用无效果。
func (a *Async) Start() {
	a.once.Do(func() {
		if a.Inner == nil {
			a.Inner = &Memory{}
		}
		if a.Queue < 1 {
			a.Queue = 1024
		}
		if a.FlushTimeout <= 0 {
			a.FlushTimeout = 5 * time.Second
		}
		a.ch = make(chan logsys.Entry, a.Queue)
		a.syncCh = make(chan chan struct{})
		a.stop = make(chan struct{})
		a.done = make(chan struct{})
		go a.loop()
	})
}

// Write 非阻塞入队。队列满时丢弃本条。
func (a *Async) Write(e logsys.Entry) error {
	a.Start()
	select {
	case <-a.stop:
		return ErrClosed
	default:
	}
	select {
	case a.ch <- e:
		a.accepted.Add(1)
		return nil
	case <-a.stop:
		return ErrClosed
	default:
		a.dropped.Add(1)
		return nil
	}
}

// Dropped 返回因队列满被丢弃的条数。
func (a *Async) Dropped() int64 { return a.dropped.Load() }

// Accepted 返回成功入队的条数。
func (a *Async) Accepted() int64 { return a.accepted.Load() }

// Sync 等待已入队的日志写完并刷出底层出口。
func (a *Async) Sync() error {
	a.Start()
	req := make(chan struct{})
	select {
	case <-a.stop:
		return ErrClosed
	case a.syncCh <- req:
	case <-time.After(a.FlushTimeout):
		return errors.New("sink: async sync enqueue timeout")
	}
	select {
	case <-req:
		return nil
	case <-time.After(a.FlushTimeout):
		return errors.New("sink: async sync timeout")
	}
}

// Close 通知写者排空队列、刷盘并关闭底层出口。重复调用返回同一次结果。
func (a *Async) Close() error {
	a.Start()
	a.closeOnce.Do(func() {
		close(a.stop)
		select {
		case <-a.done:
			a.closeErr = a.Inner.Close()
		case <-time.After(a.FlushTimeout):
			a.closeErr = errors.New("sink: async close timeout")
		}
	})
	return a.closeErr
}

func (a *Async) loop() {
	defer close(a.done)
	for {
		select {
		case <-a.stop:
			a.drain()
			_ = a.Inner.Sync()
			return
		case e := <-a.ch:
			_ = a.Inner.Write(e)
		case req := <-a.syncCh:
			a.drain()
			_ = a.Inner.Sync()
			close(req)
		}
	}
}

func (a *Async) drain() {
	for {
		select {
		case e := <-a.ch:
			_ = a.Inner.Write(e)
		default:
			return
		}
	}
}