263 lines
9.2 KiB
Go
263 lines
9.2 KiB
Go
package transcode
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"log/slog"
|
||
"time"
|
||
)
|
||
|
||
// Engine 是 Transcoder 的默认实现：把「格式无关」的那一层（输入字节上限、
|
||
// 并发信号量、单次超时、日志）收在一处，把「格式相关」的那一层交给注册进来的
|
||
// Codec（design/25 §1.3.2 / §2.2）。
|
||
//
|
||
// 它是普通结构体，由组合根构造并注入；**没有包级全局变量**
|
||
// （AGENTS §10 红线）。它可安全并发使用。
|
||
type Engine struct {
|
||
limits Limits
|
||
reg *Registry
|
||
// sem 是全局并发信号量。容量 = Limits.Concurrency；槽满即 ErrBusy
|
||
// （快速失败、不入队）。
|
||
sem chan struct{}
|
||
log *slog.Logger
|
||
// codecs 是选项里攒下的待注册列表，在 NewEngine 末尾统一注册（见 WithCodecs）。
|
||
codecs []Codec
|
||
// regErr 记录注册期的第一个错误（不该发生：DefaultCodecs 是已知良好的）。
|
||
// 它不阻塞构造，但绝不静默——组合根可读 RegistrationError 并告警。
|
||
regErr error
|
||
}
|
||
|
||
// Compile-time assertion that *Engine satisfies the Transcoder interface.
|
||
var _ Transcoder = (*Engine)(nil)
|
||
|
||
// Option 配置 Engine。
|
||
type Option func(*Engine)
|
||
|
||
// WithCodecs 把 codecs 交给引擎（在 NewEngine 的所有选项都应用完之后统一注册，
|
||
// 这样即使调用方按 design/25 §6.1 的顺序先写 WithCodecs 再写 WithLogger，
|
||
// 注册失败的告警也能落到正确的日志器上）。注册失败（重复 / 不合法）是程序错误：
|
||
// 记 Error 日志并留在 RegistrationError 里，绝不静默吞掉。
|
||
func WithCodecs(codecs ...Codec) Option {
|
||
return func(e *Engine) {
|
||
e.codecs = append(e.codecs, codecs...)
|
||
}
|
||
}
|
||
|
||
// WithLogger 设置引擎日志。日志一律带 source=capability/transcode
|
||
// （AGENTS §10：日志必须标来源）。
|
||
func WithLogger(l *slog.Logger) Option {
|
||
return func(e *Engine) {
|
||
e.log = loggerFor(l)
|
||
}
|
||
}
|
||
|
||
// loggerFor 把 nil 日志收敛成一个丢弃日志：本包不因为「没人给日志」而崩，
|
||
// 也不因为「没人给日志」而少一次调用记录（默认丢弃，责任在组合根）。
|
||
func loggerFor(l *slog.Logger) *slog.Logger {
|
||
if l == nil {
|
||
return slog.New(slog.DiscardHandler)
|
||
}
|
||
return l
|
||
}
|
||
|
||
// NewEngine 构造一个引擎。limits 的零值字段由 DefaultLimits 补足；
|
||
// 未给 WithCodecs 时注册表为空 ⇒ Supports 恒 false、Transcode 恒
|
||
// ErrUnsupportedFormat（design/25 §8-T1 的空注册表行为）。
|
||
func NewEngine(limits Limits, opts ...Option) *Engine {
|
||
e := &Engine{
|
||
limits: limits.withDefaults(),
|
||
reg: NewRegistry(),
|
||
sem: make(chan struct{}, limits.withDefaults().Concurrency),
|
||
log: loggerFor(nil),
|
||
}
|
||
for _, opt := range opts {
|
||
if opt != nil {
|
||
opt(e)
|
||
}
|
||
}
|
||
e.registerCodecs()
|
||
return e
|
||
}
|
||
|
||
// registerCodecs 注册 WithCodecs 攒下的编解码器（在所有选项都生效之后）。
|
||
func (e *Engine) registerCodecs() {
|
||
for _, c := range e.codecs {
|
||
if err := e.reg.Register(c); err != nil {
|
||
if e.regErr == nil {
|
||
e.regErr = err
|
||
}
|
||
e.log.Error("transcode codec registration skipped", "error", err)
|
||
}
|
||
}
|
||
e.codecs = nil
|
||
}
|
||
|
||
// RegistrationError 返回构造期第一个编解码器注册错误（正常为 nil）。
|
||
func (e *Engine) RegistrationError() error { return e.regErr }
|
||
|
||
// Limits 返回生效的限额（零值已补默认）。
|
||
func (e *Engine) Limits() Limits { return e.limits }
|
||
|
||
// Supports 报告 src→dst 这条路是否可用（能力门）。
|
||
func (e *Engine) Supports(src, dst Format) bool {
|
||
_, ok := e.reg.Lookup(src, dst)
|
||
return ok
|
||
}
|
||
|
||
// Formats 报告当前注册的所有格式（发现）。
|
||
func (e *Engine) Formats() []Format { return e.reg.Formats() }
|
||
|
||
// Transcode 把 req.Data 从 req.Source 转成 req.Target（design/25 §2.1）。
|
||
//
|
||
// 失败一律返回 errors.go 的哨兵之一（或包裹它）——ctx 截止返回 ErrTimeout，
|
||
// 调用方主动取消返回 context.Canceled 的包裹（取消是调用方自己的决定，
|
||
// 与「转码失败」不是一回事）。进入编解码器之前的每一道闸都在**解码前**：
|
||
// 未知格式、越限、槽满都不会解出一个字节。
|
||
//
|
||
// ctx 为 nil 时按 context.Background() 处理；ctx 超时是硬闸，绝不无界等待，
|
||
// 也不泄漏 goroutine（本调用全程在调用方 goroutine 里跑完）。
|
||
func (e *Engine) Transcode(ctx context.Context, req Request) (Result, error) {
|
||
start := time.Now()
|
||
if ctx == nil {
|
||
ctx = context.Background()
|
||
}
|
||
src, err := e.resolveSource(req)
|
||
if err != nil {
|
||
return Result{}, e.reject(ctx, req, src, err, start)
|
||
}
|
||
if req.Target == "" {
|
||
return Result{}, e.reject(ctx, req, src,
|
||
fmt.Errorf("%w: empty target format", ErrUnsupportedFormat), start)
|
||
}
|
||
codec, ok := e.reg.Lookup(src, req.Target)
|
||
if !ok {
|
||
return Result{}, e.reject(ctx, req, src,
|
||
fmt.Errorf("%w: no codec registered", ErrUnsupportedFormat), start)
|
||
}
|
||
if len(req.Data) > e.limits.MaxInputBytes {
|
||
return Result{}, e.reject(ctx, req, src,
|
||
fmt.Errorf("%w: %d bytes > %d", ErrInputTooLarge, len(req.Data), e.limits.MaxInputBytes), start)
|
||
}
|
||
if err := ctx.Err(); err != nil {
|
||
return Result{}, e.reject(ctx, req, src, timeoutError(err), start)
|
||
}
|
||
if !e.acquire() {
|
||
return Result{}, e.reject(ctx, req, src,
|
||
fmt.Errorf("%w: %d slots busy", ErrBusy, e.limits.Concurrency), start)
|
||
}
|
||
defer e.release()
|
||
|
||
callCtx, cancel := context.WithTimeout(ctx, e.limits.Timeout)
|
||
defer cancel()
|
||
res, err := codec.Transcode(callCtx, req.Data, e.limits)
|
||
if err != nil {
|
||
return Result{}, e.reject(ctx, req, src, classify(callCtx, err), start)
|
||
}
|
||
if len(res.Data) == 0 {
|
||
return Result{}, e.reject(ctx, req, src,
|
||
fmt.Errorf("%w: codec returned empty output", ErrEncodeFailed), start)
|
||
}
|
||
e.log.Info("transcode ok",
|
||
"source_format", FormatName(res.Source),
|
||
"target_format", FormatName(res.Format),
|
||
"in_bytes", len(req.Data),
|
||
"out_bytes", len(res.Data),
|
||
"frames", res.Frames,
|
||
"duration_ms", res.DurationMs,
|
||
"elapsed_ms", time.Since(start).Milliseconds())
|
||
return res, nil
|
||
}
|
||
|
||
// resolveSource 定出这次请求的实际来源格式：显式给出就用它，留空则按魔数嗅探，
|
||
// 嗅不出即 ErrUnsupportedFormat（deny-by-default）。
|
||
func (e *Engine) resolveSource(req Request) (Format, error) {
|
||
if req.Source != "" {
|
||
return req.Source, nil
|
||
}
|
||
sniffed, ok := Sniff(req.Data)
|
||
if !ok {
|
||
return "", fmt.Errorf("%w: magic not recognized", ErrUnsupportedFormat)
|
||
}
|
||
return sniffed, nil
|
||
}
|
||
|
||
// acquire 非阻塞地取一个并发槽。拿不到即 false——绝不排队。
|
||
func (e *Engine) acquire() bool {
|
||
select {
|
||
case e.sem <- struct{}{}:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// release 归还一个并发槽。
|
||
func (e *Engine) release() { <-e.sem }
|
||
|
||
// reject 记一次失败日志并把错误包上**路由上下文**后返回（绝不静默、绝不裸返回）：
|
||
// 最终文案形如 `transcode: amr -> wav: unsupported format: no codec registered`。
|
||
// 可预期的输入问题（哨兵）记 Warn；不该发生的（非哨兵：编解码器违约）记 Error。
|
||
func (e *Engine) reject(ctx context.Context, req Request, src Format, err error, start time.Time) error {
|
||
level := slog.LevelWarn
|
||
if !isSentinelError(err) {
|
||
level = slog.LevelError
|
||
}
|
||
e.log.Log(ctx, level, "transcode failed",
|
||
"source_format", FormatName(src),
|
||
"target_format", FormatName(req.Target),
|
||
"in_bytes", len(req.Data),
|
||
"elapsed_ms", time.Since(start).Milliseconds(),
|
||
"error", err)
|
||
return fmt.Errorf("transcode: %s: %w", routeLabel(src, req.Target), err)
|
||
}
|
||
|
||
// routeLabel 是错误/日志里的路由可读名；格式未知（嗅探失败 / 目标留空）时用
|
||
// `<unknown>` 占位，绝不留一段空白让人猜。
|
||
func routeLabel(src, dst Format) string {
|
||
from, to := FormatName(src), FormatName(dst)
|
||
if from == "" {
|
||
from = "<unknown>"
|
||
}
|
||
if to == "" {
|
||
to = "<unknown>"
|
||
}
|
||
return from + " -> " + to
|
||
}
|
||
|
||
// classify 把编解码器返回的错误收敛到 design/25 §2.4 的口径：ctx 截止即
|
||
// ErrTimeout；调用方取消保持 context.Canceled 的包裹；其余补一层来源后上抛
|
||
// （编解码器自己用哨兵，engine 不替它猜，但也不裸返回）。
|
||
func classify(callCtx context.Context, err error) error {
|
||
switch {
|
||
case errors.Is(err, context.DeadlineExceeded), errors.Is(callCtx.Err(), context.DeadlineExceeded):
|
||
return timeoutError(context.DeadlineExceeded)
|
||
case errors.Is(err, context.Canceled), errors.Is(callCtx.Err(), context.Canceled):
|
||
return fmt.Errorf("canceled: %w", context.Canceled)
|
||
default:
|
||
return fmt.Errorf("codec failed: %w", err)
|
||
}
|
||
}
|
||
|
||
// timeoutError 把 ctx 截止统一成 ErrTimeout 的包裹，并保留原始原因。
|
||
func timeoutError(cause error) error {
|
||
return fmt.Errorf("%w: %w", ErrTimeout, cause)
|
||
}
|
||
|
||
// isSentinelError 报告 err 是否属于本包已登记的哨兵（含包内构造期错误）。
|
||
// 逐条列举而不是查一个包级切片：本包因此**没有任何包级状态**
|
||
// （AGENTS §10：禁包级全局变量，除 var ErrXxx）。
|
||
func isSentinelError(err error) bool {
|
||
return errors.Is(err, ErrUnsupportedFormat) ||
|
||
errors.Is(err, ErrInputTooLarge) ||
|
||
errors.Is(err, ErrOutputTooLarge) ||
|
||
errors.Is(err, ErrDecodeFailed) ||
|
||
errors.Is(err, ErrEncodeFailed) ||
|
||
errors.Is(err, ErrTimeout) ||
|
||
errors.Is(err, ErrBusy) ||
|
||
errors.Is(err, ErrNoBackend) ||
|
||
errors.Is(err, ErrDuplicate) ||
|
||
errors.Is(err, ErrInvalidCodec) ||
|
||
errors.Is(err, ErrBadArgs)
|
||
}
|