use async_io::Timer;
use futures_lite::future;
use mediasoup::data_structures::{AppData, TransportListenIp};
use mediasoup::producer::{ProducerOptions, ProducerTraceEventType, ProducerType};
use mediasoup::router::{Router, RouterOptions};
use mediasoup::rtp_parameters::{
    MediaKind, MimeTypeAudio, MimeTypeVideo, RtcpFeedback, RtcpParameters, RtpCodecCapability,
    RtpCodecParameters, RtpCodecParametersParameters, RtpEncodingParameters,
    RtpEncodingParametersRtx, RtpHeaderExtensionParameters, RtpHeaderExtensionUri, RtpParameters,
};
use mediasoup::transport::{ProduceError, Transport, TransportGeneric};
use mediasoup::webrtc_transport::{TransportListenIps, WebRtcTransport, WebRtcTransportOptions};
use mediasoup::worker::{Worker, WorkerSettings};
use mediasoup::worker_manager::WorkerManager;
use std::collections::{HashMap, HashSet};
use std::env;
use std::num::{NonZeroU32, NonZeroU8};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

struct ProducerAppData {
    foo: i32,
    bar: &'static str,
}

fn media_codecs() -> Vec<RtpCodecCapability> {
    vec![
        RtpCodecCapability::Audio {
            mime_type: MimeTypeAudio::Opus,
            preferred_payload_type: None,
            clock_rate: NonZeroU32::new(48000).unwrap(),
            channels: NonZeroU8::new(2).unwrap(),
            parameters: RtpCodecParametersParameters::from([("foo", "111".into())]),
            rtcp_feedback: vec![],
        },
        RtpCodecCapability::Video {
            mime_type: MimeTypeVideo::Vp8,
            preferred_payload_type: None,
            clock_rate: NonZeroU32::new(90000).unwrap(),
            parameters: RtpCodecParametersParameters::new(),
            rtcp_feedback: vec![],
        },
        RtpCodecCapability::Video {
            mime_type: MimeTypeVideo::H264,
            preferred_payload_type: None,
            clock_rate: NonZeroU32::new(90000).unwrap(),
            parameters: RtpCodecParametersParameters::from([
                ("level-asymmetry-allowed", 1u32.into()),
                ("packetization-mode", 1u32.into()),
                ("profile-level-id", "4d0032".into()),
                ("foo", "bar".into()),
            ]),
            rtcp_feedback: vec![],
        },
    ]
}

fn audio_producer_options() -> ProducerOptions {
    let mut options = ProducerOptions::new(
        MediaKind::Audio,
        RtpParameters {
            mid: Some("AUDIO".to_string()),
            codecs: vec![RtpCodecParameters::Audio {
                mime_type: MimeTypeAudio::Opus,
                payload_type: 0,
                clock_rate: NonZeroU32::new(48000).unwrap(),
                channels: NonZeroU8::new(2).unwrap(),
                parameters: RtpCodecParametersParameters::from([
                    ("useinbandfec", 1u32.into()),
                    ("usedtx", 1u32.into()),
                    ("foo", "222.222".into()),
                    ("bar", "333".into()),
                ]),
                rtcp_feedback: vec![],
            }],
            header_extensions: vec![
                RtpHeaderExtensionParameters {
                    uri: RtpHeaderExtensionUri::Mid,
                    id: 10,
                    encrypt: false,
                },
                RtpHeaderExtensionParameters {
                    uri: RtpHeaderExtensionUri::AudioLevel,
                    id: 12,
                    encrypt: false,
                },
            ],
            // Missing encodings on purpose.
            encodings: vec![],
            rtcp: RtcpParameters {
                cname: Some("audio-1".to_string()),
                ..RtcpParameters::default()
            },
        },
    );

    options.app_data = AppData::new(ProducerAppData { foo: 1, bar: "2" });

    options
}

fn video_producer_options() -> ProducerOptions {
    let mut options = ProducerOptions::new(
        MediaKind::Video,
        RtpParameters {
            mid: Some("VIDEO".to_string()),
            codecs: vec![
                RtpCodecParameters::Video {
                    mime_type: MimeTypeVideo::H264,
                    payload_type: 112,
                    clock_rate: NonZeroU32::new(90000).unwrap(),
                    parameters: RtpCodecParametersParameters::from([
                        ("packetization-mode", 1u32.into()),
                        ("profile-level-id", "4d0032".into()),
                    ]),
                    rtcp_feedback: vec![
                        RtcpFeedback::Nack,
                        RtcpFeedback::NackPli,
                        RtcpFeedback::GoogRemb,
                    ],
                },
                RtpCodecParameters::Video {
                    mime_type: MimeTypeVideo::Rtx,
                    payload_type: 113,
                    clock_rate: NonZeroU32::new(90000).unwrap(),
                    parameters: RtpCodecParametersParameters::from([("apt", 112u32.into())]),
                    rtcp_feedback: vec![],
                },
            ],
            header_extensions: vec![
                RtpHeaderExtensionParameters {
                    uri: RtpHeaderExtensionUri::Mid,
                    id: 10,
                    encrypt: false,
                },
                RtpHeaderExtensionParameters {
                    uri: RtpHeaderExtensionUri::VideoOrientation,
                    id: 13,
                    encrypt: false,
                },
            ],
            encodings: vec![
                RtpEncodingParameters {
                    ssrc: Some(22222222),
                    rtx: Some(RtpEncodingParametersRtx { ssrc: 22222223 }),
                    scalability_mode: Some("L1T3".to_string()),
                    ..RtpEncodingParameters::default()
                },
                RtpEncodingParameters {
                    ssrc: Some(22222224),
                    rtx: Some(RtpEncodingParametersRtx { ssrc: 22222225 }),
                    ..RtpEncodingParameters::default()
                },
                RtpEncodingParameters {
                    ssrc: Some(22222226),
                    rtx: Some(RtpEncodingParametersRtx { ssrc: 22222227 }),
                    ..RtpEncodingParameters::default()
                },
                RtpEncodingParameters {
                    ssrc: Some(22222228),
                    rtx: Some(RtpEncodingParametersRtx { ssrc: 22222229 }),
                    ..RtpEncodingParameters::default()
                },
            ],
            rtcp: RtcpParameters {
                cname: Some("video-1".to_string()),
                ..RtcpParameters::default()
            },
        },
    );

    options.app_data = AppData::new(ProducerAppData { foo: 1, bar: "2" });

    options
}

async fn init() -> (Worker, Router, WebRtcTransport, WebRtcTransport) {
    {
        let mut builder = env_logger::builder();
        if env::var(env_logger::DEFAULT_FILTER_ENV).is_err() {
            builder.filter_level(log::LevelFilter::Off);
        }
        let _ = builder.is_test(true).try_init();
    }

    let worker_manager = WorkerManager::new(
        env::var("MEDIASOUP_WORKER_BIN")
            .map(|path| path.into())
            .unwrap_or_else(|_| "../worker/out/Release/mediasoup-worker".into()),
    );

    let worker = worker_manager
        .create_worker(WorkerSettings::default())
        .await
        .expect("Failed to create worker");

    let router = worker
        .create_router(RouterOptions::new(media_codecs()))
        .await
        .expect("Failed to create router");

    let transport_options =
        WebRtcTransportOptions::new(TransportListenIps::new(TransportListenIp {
            ip: "127.0.0.1".parse().unwrap(),
            announced_ip: None,
        }));

    let transport_1 = router
        .create_webrtc_transport(transport_options.clone())
        .await
        .expect("Failed to create transport1");

    let transport_2 = router
        .create_webrtc_transport(transport_options)
        .await
        .expect("Failed to create transport2");

    (worker, router, transport_1, transport_2)
}

#[test]
fn produce_succeeds() {
    future::block_on(async move {
        let (_worker, router, transport_1, transport_2) = init().await;

        {
            let new_producers_count = Arc::new(AtomicUsize::new(0));

            transport_1
                .on_new_producer({
                    let new_producers_count = Arc::clone(&new_producers_count);

                    Box::new(move |_producer| {
                        new_producers_count.fetch_add(1, Ordering::SeqCst);
                    })
                })
                .detach();

            let audio_producer = transport_1
                .produce(audio_producer_options())
                .await
                .expect("Failed to produce audio");

            assert_eq!(new_producers_count.load(Ordering::SeqCst), 1);
            assert_eq!(audio_producer.closed(), false);
            assert_eq!(audio_producer.kind(), MediaKind::Audio);
            assert_eq!(audio_producer.r#type(), ProducerType::Simple);
            assert_eq!(audio_producer.paused(), false);
            assert_eq!(audio_producer.score(), vec![]);
            assert_eq!(
                audio_producer
                    .app_data()
                    .downcast_ref::<ProducerAppData>()
                    .unwrap()
                    .foo,
                1
            );
            assert_eq!(
                audio_producer
                    .app_data()
                    .downcast_ref::<ProducerAppData>()
                    .unwrap()
                    .bar,
                "2"
            );

            let router_dump = router.dump().await.expect("Failed to get router dump");

            assert_eq!(router_dump.map_producer_id_consumer_ids, {
                let mut map = HashMap::new();
                map.insert(audio_producer.id(), HashSet::new());
                map
            });
            assert_eq!(router_dump.map_consumer_id_producer_id, HashMap::new());

            let transport_1_dump = transport_1
                .dump()
                .await
                .expect("Failed to get transport 1 dump");

            assert_eq!(transport_1_dump.producer_ids, vec![audio_producer.id()]);
            assert_eq!(transport_1_dump.consumer_ids, vec![]);
        }

        {
            let new_producers_count = Arc::new(AtomicUsize::new(0));

            transport_2
                .on_new_producer({
                    let new_producers_count = Arc::clone(&new_producers_count);

                    Box::new(move |_producer| {
                        new_producers_count.fetch_add(1, Ordering::SeqCst);
                    })
                })
                .detach();

            let video_producer = transport_2
                .produce(video_producer_options())
                .await
                .expect("Failed to produce video");

            assert_eq!(new_producers_count.load(Ordering::SeqCst), 1);
            assert_eq!(video_producer.closed(), false);
            assert_eq!(video_producer.kind(), MediaKind::Video);
            assert_eq!(video_producer.r#type(), ProducerType::Simulcast);
            assert_eq!(video_producer.paused(), false);
            assert_eq!(video_producer.score(), vec![]);
            assert_eq!(
                video_producer
                    .app_data()
                    .downcast_ref::<ProducerAppData>()
                    .unwrap()
                    .foo,
                1
            );
            assert_eq!(
                video_producer
                    .app_data()
                    .downcast_ref::<ProducerAppData>()
                    .unwrap()
                    .bar,
                "2"
            );

            let router_dump = router.dump().await.expect("Failed to get router dump");

            assert_eq!(router_dump.map_producer_id_consumer_ids, {
                let mut map = HashMap::new();
                map.insert(video_producer.id(), HashSet::new());
                map
            });
            assert_eq!(router_dump.map_consumer_id_producer_id, HashMap::new());

            let transport_2_dump = transport_2
                .dump()
                .await
                .expect("Failed to get transport 2 dump");

            assert_eq!(transport_2_dump.producer_ids, vec![video_producer.id()]);
            assert_eq!(transport_2_dump.consumer_ids, vec![]);
        }
    });
}

#[test]
fn weak() {
    future::block_on(async move {
        let (_worker, _router, transport_1, _transport_2) = init().await;

        let producer = transport_1
            .produce(audio_producer_options())
            .await
            .expect("Failed to produce audio");

        let weak_producer = producer.downgrade();

        assert!(weak_producer.upgrade().is_some());

        drop(producer);

        assert!(weak_producer.upgrade().is_none());
    });
}

#[test]
fn produce_wrong_arguments() {
    future::block_on(async move {
        let (_worker, _router, transport_1, _transport_2) = init().await;

        // Empty rtp_parameters.codecs.
        assert!(matches!(
            transport_1
                .produce(ProducerOptions::new(
                    MediaKind::Audio,
                    RtpParameters::default()
                ))
                .await,
            Err(ProduceError::Request(_)),
        ));

        {
            // Empty rtp_parameters.encodings.
            let produce_result = transport_1
                .produce(ProducerOptions::new(MediaKind::Video, {
                    let mut parameters = RtpParameters::default();
                    parameters.codecs = vec![
                        RtpCodecParameters::Video {
                            mime_type: MimeTypeVideo::H264,
                            payload_type: 112,
                            clock_rate: NonZeroU32::new(90000).unwrap(),
                            parameters: RtpCodecParametersParameters::from([
                                ("packetization-mode", 1u32.into()),
                                ("profile-level-id", "4d0032".into()),
                            ]),
                            rtcp_feedback: vec![],
                        },
                        RtpCodecParameters::Video {
                            mime_type: MimeTypeVideo::Rtx,
                            payload_type: 113,
                            clock_rate: NonZeroU32::new(90000).unwrap(),
                            parameters: RtpCodecParametersParameters::from([(
                                "apt",
                                112u32.into(),
                            )]),
                            rtcp_feedback: vec![],
                        },
                    ];
                    parameters.rtcp = RtcpParameters {
                        cname: Some("qwerty".to_string()),
                        ..RtcpParameters::default()
                    };
                    parameters
                }))
                .await;

            assert!(matches!(produce_result, Err(ProduceError::Request(_))));
        }

        {
            // Wrong apt in RTX codec.
            let produce_result = transport_1
                .produce(ProducerOptions::new(MediaKind::Video, {
                    let mut parameters = RtpParameters::default();
                    parameters.codecs = vec![
                        RtpCodecParameters::Video {
                            mime_type: MimeTypeVideo::H264,
                            payload_type: 112,
                            clock_rate: NonZeroU32::new(90000).unwrap(),
                            parameters: RtpCodecParametersParameters::from([
                                ("packetization-mode", 1u32.into()),
                                ("profile-level-id", "4d0032".into()),
                            ]),
                            rtcp_feedback: vec![],
                        },
                        RtpCodecParameters::Video {
                            mime_type: MimeTypeVideo::Rtx,
                            payload_type: 113,
                            clock_rate: NonZeroU32::new(90000).unwrap(),
                            parameters: RtpCodecParametersParameters::from([(
                                "apt",
                                111u32.into(),
                            )]),
                            rtcp_feedback: vec![],
                        },
                    ];
                    parameters.encodings = vec![RtpEncodingParameters {
                        ssrc: Some(6666),
                        rtx: Some(RtpEncodingParametersRtx { ssrc: 6667 }),
                        ..RtpEncodingParameters::default()
                    }];
                    parameters.rtcp = RtcpParameters {
                        cname: Some("video-1".to_string()),
                        ..RtcpParameters::default()
                    };
                    parameters
                }))
                .await;

            assert!(matches!(
                produce_result,
                Err(ProduceError::FailedRtpParametersMapping(_))
            ));
        }
    });
}

#[test]
fn produce_unsupported_codecs() {
    future::block_on(async move {
        let (_worker, _router, transport_1, _transport_2) = init().await;

        // Empty rtp_parameters.codecs.
        assert!(matches!(
            transport_1
                .produce(ProducerOptions::new(MediaKind::Audio, {
                    let mut parameters = RtpParameters::default();
                    parameters.codecs = vec![RtpCodecParameters::Audio {
                        mime_type: MimeTypeAudio::Isac,
                        payload_type: 108,
                        clock_rate: NonZeroU32::new(32000).unwrap(),
                        channels: NonZeroU8::new(1).unwrap(),
                        parameters: RtpCodecParametersParameters::new(),
                        rtcp_feedback: vec![],
                    }];
                    parameters.header_extensions = vec![];
                    parameters.encodings = vec![RtpEncodingParameters {
                        ssrc: Some(1111),
                        ..RtpEncodingParameters::default()
                    }];
                    parameters.rtcp = RtcpParameters {
                        cname: Some("audio".to_string()),
                        ..RtcpParameters::default()
                    };
                    parameters
                }))
                .await,
            Err(ProduceError::FailedRtpParametersMapping(_)),
        ));

        {
            // Invalid H264 profile-level-id.
            let produce_result = transport_1
                .produce(ProducerOptions::new(MediaKind::Video, {
                    let mut parameters = RtpParameters::default();
                    parameters.codecs = vec![
                        RtpCodecParameters::Video {
                            mime_type: MimeTypeVideo::H264,
                            payload_type: 112,
                            clock_rate: NonZeroU32::new(90000).unwrap(),
                            parameters: RtpCodecParametersParameters::from([
                                ("packetization-mode", 1u32.into()),
                                ("profile-level-id", "CHICKEN".into()),
                            ]),
                            rtcp_feedback: vec![],
                        },
                        RtpCodecParameters::Video {
                            mime_type: MimeTypeVideo::Rtx,
                            payload_type: 113,
                            clock_rate: NonZeroU32::new(90000).unwrap(),
                            parameters: RtpCodecParametersParameters::from([(
                                "apt",
                                112u32.into(),
                            )]),
                            rtcp_feedback: vec![],
                        },
                    ];
                    parameters.encodings = vec![RtpEncodingParameters {
                        ssrc: Some(6666),
                        rtx: Some(RtpEncodingParametersRtx { ssrc: 6667 }),
                        ..RtpEncodingParameters::default()
                    }];
                    parameters
                }))
                .await;

            assert!(matches!(
                produce_result,
                Err(ProduceError::FailedRtpParametersMapping(_))
            ));
        }
    });
}

#[test]
fn produce_already_used_mid_ssrc() {
    future::block_on(async move {
        let (_worker, _router, transport_1, transport_2) = init().await;

        {
            let _first_producer = transport_1
                .produce(audio_producer_options())
                .await
                .expect("Failed to produce audio");

            let produce_result = transport_1
                .produce(ProducerOptions::new(MediaKind::Audio, {
                    let mut parameters = RtpParameters::default();
                    parameters.mid = Some("AUDIO".to_string());
                    parameters.codecs = vec![RtpCodecParameters::Audio {
                        mime_type: MimeTypeAudio::Opus,
                        payload_type: 0,
                        clock_rate: NonZeroU32::new(48000).unwrap(),
                        channels: NonZeroU8::new(2).unwrap(),
                        parameters: RtpCodecParametersParameters::new(),
                        rtcp_feedback: vec![],
                    }];
                    parameters.header_extensions = vec![];
                    parameters.encodings = vec![RtpEncodingParameters {
                        ssrc: Some(33333333),
                        ..RtpEncodingParameters::default()
                    }];
                    parameters.rtcp = RtcpParameters {
                        cname: Some("audio-2".to_string()),
                        ..RtcpParameters::default()
                    };
                    parameters
                }))
                .await;

            assert!(matches!(produce_result, Err(ProduceError::Request(_)),));
        }

        {
            let _first_producer = transport_2
                .produce(video_producer_options())
                .await
                .expect("Failed to produce video");

            // Invalid H264 profile-level-id.
            let produce_result = transport_2
                .produce(ProducerOptions::new(MediaKind::Video, {
                    let mut parameters = RtpParameters::default();
                    parameters.mid = Some("VIDEO2".to_string());
                    parameters.codecs = vec![RtpCodecParameters::Video {
                        mime_type: MimeTypeVideo::Vp8,
                        payload_type: 112,
                        clock_rate: NonZeroU32::new(90000).unwrap(),
                        parameters: RtpCodecParametersParameters::new(),
                        rtcp_feedback: vec![],
                    }];
                    parameters.encodings = vec![RtpEncodingParameters {
                        ssrc: Some(22222222),
                        rtx: Some(RtpEncodingParametersRtx { ssrc: 6667 }),
                        ..RtpEncodingParameters::default()
                    }];
                    parameters
                }))
                .await;

            assert!(matches!(produce_result, Err(ProduceError::Request(_))));
        }
    });
}

#[test]
fn produce_no_mid_single_encoding_without_dir_or_ssrc() {
    future::block_on(async move {
        let (_worker, _router, transport_1, _transport_2) = init().await;

        let produce_result = transport_1
            .produce(ProducerOptions::new(MediaKind::Audio, {
                let mut parameters = RtpParameters::default();
                parameters.codecs = vec![RtpCodecParameters::Audio {
                    mime_type: MimeTypeAudio::Opus,
                    payload_type: 111,
                    clock_rate: NonZeroU32::new(48000).unwrap(),
                    channels: NonZeroU8::new(2).unwrap(),
                    parameters: RtpCodecParametersParameters::new(),
                    rtcp_feedback: vec![],
                }];
                parameters.header_extensions = vec![];
                parameters.encodings = vec![RtpEncodingParameters::default()];
                parameters.rtcp = RtcpParameters {
                    cname: Some("audio-2".to_string()),
                    ..RtcpParameters::default()
                };
                parameters
            }))
            .await;

        assert!(matches!(produce_result, Err(ProduceError::Request(_)),));
    });
}

#[test]
fn dump_succeeds() {
    future::block_on(async move {
        let (_worker, _router, transport_1, transport_2) = init().await;

        {
            let audio_producer = transport_1
                .produce(audio_producer_options())
                .await
                .expect("Failed to produce audio");

            let dump = audio_producer
                .dump()
                .await
                .expect("Failed to dump audio producer");

            assert_eq!(dump.id, audio_producer.id());
            assert_eq!(dump.kind, audio_producer.kind());
            assert_eq!(
                dump.rtp_parameters.codecs,
                audio_producer_options().rtp_parameters.codecs,
            );
            assert_eq!(
                dump.rtp_parameters.header_extensions,
                audio_producer_options().rtp_parameters.header_extensions,
            );
            assert_eq!(
                dump.rtp_parameters.encodings,
                vec![RtpEncodingParameters {
                    ssrc: None,
                    rid: None,
                    codec_payload_type: Some(0),
                    rtx: None,
                    dtx: None,
                    scalability_mode: None,
                    scale_resolution_down_by: None,
                    max_bitrate: None
                }],
            );
            assert_eq!(dump.r#type, ProducerType::Simple);
        }

        {
            let video_producer = transport_2
                .produce(video_producer_options())
                .await
                .expect("Failed to produce video");

            let dump = video_producer
                .dump()
                .await
                .expect("Failed to dump video producer");

            assert_eq!(dump.id, video_producer.id());
            assert_eq!(dump.kind, video_producer.kind());
            assert_eq!(
                dump.rtp_parameters.codecs,
                video_producer_options().rtp_parameters.codecs,
            );
            assert_eq!(
                dump.rtp_parameters.header_extensions,
                video_producer_options().rtp_parameters.header_extensions,
            );
            assert_eq!(
                dump.rtp_parameters.encodings,
                vec![
                    RtpEncodingParameters {
                        ssrc: Some(22222222),
                        rid: None,
                        codec_payload_type: Some(112),
                        rtx: Some(RtpEncodingParametersRtx { ssrc: 22222223 }),
                        dtx: None,
                        scalability_mode: Some("L1T3".to_string()),
                        scale_resolution_down_by: None,
                        max_bitrate: None
                    },
                    RtpEncodingParameters {
                        ssrc: Some(22222224),
                        rid: None,
                        codec_payload_type: Some(112),
                        rtx: Some(RtpEncodingParametersRtx { ssrc: 22222225 }),
                        dtx: None,
                        scalability_mode: None,
                        scale_resolution_down_by: None,
                        max_bitrate: None
                    },
                    RtpEncodingParameters {
                        ssrc: Some(22222226),
                        rid: None,
                        codec_payload_type: Some(112),
                        rtx: Some(RtpEncodingParametersRtx { ssrc: 22222227 }),
                        dtx: None,
                        scalability_mode: None,
                        scale_resolution_down_by: None,
                        max_bitrate: None
                    },
                    RtpEncodingParameters {
                        ssrc: Some(22222228),
                        rid: None,
                        codec_payload_type: Some(112),
                        rtx: Some(RtpEncodingParametersRtx { ssrc: 22222229 }),
                        dtx: None,
                        scalability_mode: None,
                        scale_resolution_down_by: None,
                        max_bitrate: None
                    },
                ],
            );
            assert_eq!(dump.r#type, ProducerType::Simulcast);
        }
    });
}

#[test]
fn get_stats_succeeds() {
    future::block_on(async move {
        let (_worker, _router, transport_1, transport_2) = init().await;

        {
            let audio_producer = transport_1
                .produce(audio_producer_options())
                .await
                .expect("Failed to produce audio");

            let stats = audio_producer
                .get_stats()
                .await
                .expect("Failed to get stats on audio producer");

            assert_eq!(stats, vec![]);
        }

        {
            let video_producer = transport_2
                .produce(video_producer_options())
                .await
                .expect("Failed to produce video");

            let stats = video_producer
                .get_stats()
                .await
                .expect("Failed to get stats on video producer");

            assert_eq!(stats, vec![]);
        }
    });
}

#[test]
fn pause_resume_succeeds() {
    future::block_on(async move {
        let (_worker, _router, transport_1, _transport_2) = init().await;

        let audio_producer = transport_1
            .produce(audio_producer_options())
            .await
            .expect("Failed to produce audio");

        {
            audio_producer
                .pause()
                .await
                .expect("Failed to pause audio producer");

            assert_eq!(audio_producer.paused(), true);

            let dump = audio_producer
                .dump()
                .await
                .expect("Failed to dump audio producer");

            assert_eq!(dump.paused, true);
        }

        {
            audio_producer
                .resume()
                .await
                .expect("Failed to resume audio producer");

            assert_eq!(audio_producer.paused(), false);

            let dump = audio_producer
                .dump()
                .await
                .expect("Failed to dump audio producer");

            assert_eq!(dump.paused, false);
        }
    });
}

#[test]
fn enable_trace_event_succeeds() {
    future::block_on(async move {
        let (_worker, _router, transport_1, _transport_2) = init().await;

        let audio_producer = transport_1
            .produce(audio_producer_options())
            .await
            .expect("Failed to produce audio");

        {
            audio_producer
                .enable_trace_event(vec![
                    ProducerTraceEventType::Rtp,
                    ProducerTraceEventType::Pli,
                ])
                .await
                .expect("Failed to enable trace event");

            let dump = audio_producer
                .dump()
                .await
                .expect("Failed to dump audio producer");

            assert_eq!(dump.trace_event_types.as_str(), "rtp,pli");
        }

        {
            audio_producer
                .enable_trace_event(vec![])
                .await
                .expect("Failed to enable trace event");

            let dump = audio_producer
                .dump()
                .await
                .expect("Failed to dump audio producer");

            assert_eq!(dump.trace_event_types.as_str(), "");
        }
    });
}

#[test]
fn close_event() {
    future::block_on(async move {
        let (_worker, router, transport_1, _transport_2) = init().await;

        let audio_producer = transport_1
            .produce(audio_producer_options())
            .await
            .expect("Failed to produce audio");

        {
            let (tx, rx) = async_oneshot::oneshot::<()>();
            let _handler = audio_producer.on_close(move || {
                let _ = tx.send(());
            });
            drop(audio_producer);

            rx.await.expect("Failed to receive close event");
        }

        // Drop is async, give producer a bit of time to finish
        Timer::after(Duration::from_millis(200)).await;

        {
            let dump = router.dump().await.expect("Failed to dump router");

            assert_eq!(dump.map_producer_id_consumer_ids, HashMap::new());
            assert_eq!(dump.map_consumer_id_producer_id, HashMap::new());
        }

        {
            let dump = transport_1.dump().await.expect("Failed to dump transport");

            assert_eq!(dump.producer_ids, vec![]);
            assert_eq!(dump.consumer_ids, vec![]);
        }
    });
}

#[test]
fn transport_close_event() {
    future::block_on(async move {
        let (worker, _router, transport_1, _transport_2) = init().await;

        let audio_producer = transport_1
            .produce(audio_producer_options())
            .await
            .expect("Failed to produce audio");

        let (close_tx, close_rx) = async_oneshot::oneshot::<()>();
        let _handler = audio_producer.on_close(move || {
            let _ = close_tx.send(());
        });

        let (transport_close_tx, transport_close_rx) = async_oneshot::oneshot::<()>();
        let _handler = audio_producer.on_transport_close(move || {
            let _ = transport_close_tx.send(());
        });

        unsafe {
            libc::kill(worker.pid() as i32, libc::SIGINT);
        }

        transport_close_rx
            .await
            .expect("Failed to receive transport_close event");
        close_rx.await.expect("Failed to receive close event");

        assert_eq!(audio_producer.closed(), true);
    });
}
