define source error categories

internal/source/errors.go

@@ -0,0 +1,51 @@
+package source
+
+import (
+	"errors"
+	"fmt"
+)
+
+type ErrorKind uint8
+
+const (
+	ErrorKindUnknown       ErrorKind = iota
+	ErrorKindTemporary               // timeout or temporarily early/late data i.e. wait or resync
+	ErrorKindUnavailable             // producer/flow disappeared
+	ErrorKindInvalidConfig           // wrong media type, invalid rate, etc.
+)
+
+type SourceError struct {
+	Op   string
+	Kind ErrorKind
+	Err  error
+}
+
+func (e *SourceError) Error() string {
+	if e.Op == "" {
+		return e.Err.Error()
+	}
+	return fmt.Sprintf("%s: %v", e.Op, e.Err)
+}
+
+func (e *SourceError) Unwrap() error {
+	return e.Err
+}
+
+func errorKind(err error) ErrorKind {
+	var sourceErr *SourceError
+	if errors.As(err, &sourceErr) {
+		return sourceErr.Kind
+	}
+	return ErrorKindUnknown
+}
+
+func wrapError(op string, kind ErrorKind, err error) error {
+	if err == nil {
+		return nil
+	}
+	return &SourceError{
+		Op:   op,
+		Kind: kind,
+		Err:  err,
+	}
+}