Add custom resource type registration

[mmatczuk]

Allow libraries to register custom resource types via
Environment.RegisterResource. Resources are declared in YAML under
"{name}", constructed per-stream by the manager, and accessible to
components via Resources.AccessResource. Values implementing
io.Closer are closed on shutdown. Context aware Close is supported.

[Jeffail]

Public API Surface Review

Having reviewed the public API additions in public/service/, I wished to raise a few observations for consideration.

1. AccessResource departs from the established Access* callback pattern

The existing resource accessors all follow a consistent shape:

AccessCache(ctx context.Context, name string, fn func(c Cache)) error
AccessInput(ctx context.Context, name string, fn func(i *ResourceInput)) error

The callback-with-lock pattern exists for good reason — it permits the manager to swap the underlying resource between calls (as occurs with the -w / watcher flag when configuration is reloaded). The new method returns the value directly:

AccessResource(typeName, label string) (any, bool)

Should custom resources need to support hot-reloading via -w in the future, this signature would be insufficient — callers holding a direct reference would not observe the swap. It may be worth adopting the callback pattern now to keep that door open, even if the initial implementation does not yet perform swaps.

2. RegisterResource accepts []*ConfigField rather than *ConfigSpec

Every other registration method accepts a *ConfigSpec:

RegisterCache(name string, spec *ConfigSpec, ctor CacheConstructor) error
RegisterInput(name string, spec *ConfigSpec, ctor InputConstructor) error

The new method takes a raw field slice instead:

RegisterResource(name string, fields []*ConfigField, ctor ResourceConstructor) error

ConfigSpec bundles fields together with a description and version metadata. Bypassing it means custom resources cannot carry a top-level description or version information, and it introduces an asymmetry in the registration API that library consumers will notice. Would it be preferable to accept a *ConfigSpec for consistency?

3. HasResource / AccessResource naming may be misleadingly broad

The names HasResource and AccessResource read as though they cover all resource types — a reasonable assumption given that caches, inputs, processors, and rate limits are all "resources." In practice, these methods only operate on custom resource types and will not return true for a cache named "foo", for example.

Consider naming them more specifically — perhaps HasCustomResource and AccessCustomResource — to make the scope immediately obvious and avoid confusion at the call site.

---

These are observations rather than blockers; happy to discuss further.

[claude]

Commits
LGTM

Review
Adds custom resource type registration via Environment.RegisterResource, YAML config parsing, manager construction with label validation, access via Resources.AccessCustomResource, and shutdown with Close(ctx)/io.Closer support. Reserved field name conflicts and duplicate type registration are properly guarded. Good test coverage including unit tests for registration, access, shutdown (both io.Closer and context-aware), duplicate labels, YAML round-trip, and a full end-to-end test via the service package.

LGTM

[claude]

Commits
LGTM

Review
This PR adds custom resource type registration to the Benthos environment, allowing libraries to declare custom YAML-configured resources that are constructed per-stream, accessible via Resources.AccessCustomResource, and properly closed on shutdown. The implementation follows established patterns (callback-based access, checkLabel for duplicate detection, ConfigSpec-based registration). Good test coverage across unit and e2e levels.

LGTM

[Jeffail]

LGTM, this is great