processor.go

package log

import (
	"fmt"
	"sync"
	"sync/atomic"
)

const buffSize = 1024

// processor handles message processing and writing in either blocking or non-blocking mode.
// In non-blocking mode, it uses a buffered channel and background goroutine.
type processor struct {
	mode         WriteMode
	buf          chan *message
	wg           sync.WaitGroup
	closed       atomic.Bool
	droppedCount atomic.Uint64
}

// NewProcessor creates a new processor with the specified write mode.
// In non-blocking mode, starts a background goroutine for processing.
func NewProcessor(mode WriteMode) *processor {
	p := &processor{
		mode: mode,
	}

	if mode == ModeNonBlocking {
		p.run()
	}

	return p
}

func (p *processor) run() {
	p.buf = make(chan *message, buffSize)
	p.wg.Add(1)

	go func() {
		defer p.wg.Done()
		for msg := range p.buf {
			if msg != nil {
				p.write(msg)
			}
		}
	}()
}

func (p *processor) Do(m *message) {
	switch p.mode {
	case ModeBlocking:
		p.write(m)
	case ModeNonBlocking:
		p.addToBuf(m)
	}

	if m.level == Fatal {
		panic(m.text)
	}
}

func (p *processor) addToBuf(m *message) {
	if m == nil {
		return
	}

	defer func() {
		if recover() != nil {
			fmt.Println("log: error writing to buffer")
		}
	}()

	select {
	case p.buf <- m:
		// Successfully added to buffer
	default:
		// Buffer is full, drop the message and increment counter
		dropped := p.droppedCount.Add(1)
		// Log warning every 100 dropped messages to avoid spam
		if dropped%100 == 1 {
			// Safely write warning to first available writer
			if len(m.super.global.writers) > 0 {
				fmt.Fprintf(m.super.global.writers[0].writer,
					"log: WARNING - buffer overflow, %d messages dropped\n", dropped)
			}
		}
	}
}

func (p *processor) write(m *message) {
	v, ok := levels[m.level]
	if !ok {
		return
	}

	logger := m.super

	if v < int(logger.local.level.Load()) {
		return
	}

	if len(m.text) == 0 {
		return
	}

	for _, w := range logger.global.writers {
		str, err := formatter.Get(&formatterProps{m, w.format})
		if err != nil {
			continue
		}

		p.writeByMode(w, str)
	}

	for _, w := range logger.local.writers {
		str, err := formatter.Get(&formatterProps{m, w.format})
		if err != nil {
			continue
		}

		p.writeByMode(w, str)
	}
}

func (p *processor) writeByMode(w *writer, msg string) {
	switch p.mode {
	case ModeBlocking:
		p.writeSync(w, msg)
	case ModeNonBlocking:
		p.writeAsync(w, msg)
	}
}

func (p *processor) writeAsync(w *writer, msg string) {
	w.mu.Lock()
	fmt.Fprintln(w.writer, msg)
	w.mu.Unlock()
}

func (p *processor) writeSync(w *writer, msg string) {
	w.mu.Lock()
	fmt.Fprintln(w.writer, msg)
	w.mu.Unlock()
}

// Shutdown gracefully shuts down the processor, draining the buffer
func (p *processor) Shutdown() error {
	if p.mode != ModeNonBlocking {
		// Nothing to do for blocking mode
		return nil
	}

	if !p.closed.CompareAndSwap(false, true) {
		return fmt.Errorf("processor already closed")
	}

	close(p.buf)
	p.wg.Wait()
	return nil
}

// DroppedCount returns the number of messages dropped due to buffer overflow
func (p *processor) DroppedCount() uint64 {
	return p.droppedCount.Load()
}