Patterns for Building Custom Asset Watchers

### Description

Less of a feature request and more of a conversation around how we'll start to build custom Asset Watchers.

Is the general pattern going to be to build a Trigger, and leverage the out-of-the-box `airflow.sdk.AssetWatcher` when defining an `Asset`? Or instead, would a user/contributor define a custom Asset Watcher that has a baked-in Trigger.

How will we pass information from an `AssetWatcher` object to downstream Tasks in a DAG? Will this always be via `context['triggering_asset_events'] ...`? The most obvious example would be a file lands in an object store, and we want to process it.

Here's an example that's available in Astronomer's documentation, using SQS.

```
# Define an asset that watches for messages on the queue
sqs_queue_asset = Asset(
"sqs_queue_asset", watchers=[AssetWatcher(name="sqs_watcher", trigger=trigger)]
)

# Schedule the DAG to run when the asset is triggered
@dag(schedule=[sqs_queue_asset])
def event_driven_dag():
@task
def process_message(**context):
# Extract the triggering asset events from the context
triggering_asset_events = context["triggering_asset_events"]
for event in triggering_asset_events[sqs_queue_asset]:
# Get the message from the TriggerEvent payload
print(
f"Processing message: {event.extra["payload"]["message_batch"][0]["Body"]}"
)

process_message()

event_driven_dag()
```

### Use case/motivation

Event-driven (Asset-aware) scheduling is (IMO) going to be one of the most powerful features that Airflow users will leverage. I'm hoping to lay the groundwork for this as both users create custom Asset Watchers and contributors add to providers.

### Related issues

_No response_

### Are you willing to submit a PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)