use futures::compat::Future01CompatExt;
use log::{error, info};
use runtime;
use rusoto_core::Region;
use serde_derive::Serialize;
use sqsquatch::{PubFut, Result, SubFut};
use std::collections::HashMap;

#[derive(Clone, Debug, Serialize)]
struct Order<'a> {
    order_id: &'a str,
}

#[runtime::test(runtime_tokio::Tokio)]
async fn publish() -> Result<()> {
    pretty_env_logger::try_init_timed()?;
    let mut publisher = PubFut::initialize(Region::UsEast2, "Orders")?;
    let mut attributes = HashMap::new();
    let _ = attributes.insert("kind", "create");
    let attr = attributes
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect();
    if let Some(message_id) = publisher
        .publish_fut(Order { order_id: "123456" }, Some(attr))
        .compat()
        .await?
        .message_id
    {
        info!("Message ID: {}", message_id);
    } else {
        error!("No message id!");
    }

    let mut subscriber = SubFut::initialize(Region::UsEast2, "Orders")?;
    let result = subscriber
        .receive_message_fut(Some(vec!["All".to_string()]), Some(30), Some(20))
        .compat()
        .await?;
    info!("Message Received: {:?}", result);

    if let Some(messages) = result.messages {
        let mut touched = false;
        for message in messages {
            if let Some(receipt_handle) = message.receipt_handle {
                assert!(subscriber
                    .delete_message_fut(receipt_handle)
                    .compat()
                    .await
                    .is_ok());
                touched = true;
            }
        }
        if touched {
            Ok(())
        } else {
            Err("no messages deleted".into())
        }
    } else {
        Err("no messages".into())
    }
}