Amqp source add

[stan-dot]

Fixes #806

My first contribution to the repo, I assume it has a couple of mistakes at this point, so draft for now.

I'd welcome an initial review.

[stan-dot]

looking for help with this error:

error[E0277]: the trait bound `AmqpState: arroyo_types::Data` is not satisfied
   --> crates/arroyo-connectors/src/amqp/operator.rs:163:55
    |
163 | ...                   let s = ctx.table_manager.get_global_keyed_state("k").await
    |                                                 ^^^^^^^^^^^^^^^^^^^^^^ the trait `bincode::Encode` is not implemented for `AmqpState`
    |
    = help: the following other types implement trait `bincode::Encode`:
              &'a CStr
              &'a T
              &std::path::Path
              ()
              (A, B)
              (A, B, C)
              (A, B, C, D)
              (A, B, C, D, E)
            and 141 others
    = note: required for `AmqpState` to implement `arroyo_types::Data`
note: required by a bound in `TableManager::get_global_keyed_state`
   --> /home/runner/work/arroyo/arroyo/crates/arroyo-state/src/tables/table_manager.rs:387:52
    |
387 |     pub async fn get_global_keyed_state<K: Key, V: Data>(
    |                                                    ^^^^ required by this bound in `TableManager::get_global_keyed_state`

[stan-dot]

@ecarrara , @mwylde I saw your commits in the arguably similar standard Rabbit MQ connector (https://github.com/ArroyoSystems/arroyo/blob/master/crates/arroyo-connectors/src/rabbitmq/source.rs), I'm a bit confused about the offsets and GlobalKeyedView, could you please take a quick look?