mod direct_transport {
    use bytes::Bytes;
    use futures_lite::future;
    use mediasoup::data_consumer::DataConsumerOptions;
    use mediasoup::data_producer::{DataProducer, DataProducerOptions};
    use mediasoup::data_structures::{AppData, WebRtcMessage};
    use mediasoup::direct_transport::{DirectTransport, DirectTransportOptions};
    use mediasoup::router::{Router, RouterOptions};
    use mediasoup::transport::{Transport, TransportGeneric};
    use mediasoup::worker::{Worker, WorkerSettings};
    use mediasoup::worker_manager::WorkerManager;
    use parking_lot::Mutex;
    use std::collections::HashSet;
    use std::env;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    struct CustomAppData {
        foo: &'static str,
    }

    async fn init() -> (Worker, Router, DirectTransport) {
        {
            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(vec![]))
            .await
            .expect("Failed to create router");

        let transport = router
            .create_direct_transport(DirectTransportOptions::default())
            .await
            .expect("Failed to create transport1");

        (worker, router, transport)
    }

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

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

                assert_eq!(dump.transport_ids, {
                    let mut set = HashSet::new();
                    set.insert(transport.id());
                    set
                });
            }

            let new_transports_count = Arc::new(AtomicUsize::new(0));

            router
                .on_new_transport({
                    let new_transports_count = Arc::clone(&new_transports_count);

                    move |_transport| {
                        new_transports_count.fetch_add(1, Ordering::SeqCst);
                    }
                })
                .detach();

            let transport1 = router
                .create_direct_transport({
                    let mut direct_transport_options = DirectTransportOptions::default();
                    direct_transport_options.max_message_size = 1024;
                    direct_transport_options.app_data = AppData::new(CustomAppData { foo: "bar" });

                    direct_transport_options
                })
                .await
                .expect("Failed to create Direct transport");

            assert_eq!(new_transports_count.load(Ordering::SeqCst), 1);
            assert_eq!(transport1.closed(), false);
            assert_eq!(
                transport1
                    .app_data()
                    .downcast_ref::<CustomAppData>()
                    .unwrap()
                    .foo,
                "bar",
            );

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

                assert_eq!(dump.id, transport1.id());
                assert_eq!(dump.direct, true);
                assert_eq!(dump.producer_ids, vec![]);
                assert_eq!(dump.consumer_ids, vec![]);
                assert_eq!(dump.data_producer_ids, vec![]);
                assert_eq!(dump.data_consumer_ids, vec![]);
            }
        });
    }

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

            let stats = transport
                .get_stats()
                .await
                .expect("Failed to get stats on Direct transport");

            assert_eq!(stats.len(), 1);
            assert_eq!(stats[0].transport_id, transport.id());
            assert_eq!(stats[0].bytes_received, 0);
            assert_eq!(stats[0].recv_bitrate, 0);
            assert_eq!(stats[0].bytes_sent, 0);
            assert_eq!(stats[0].send_bitrate, 0);
            assert_eq!(stats[0].rtp_bytes_received, 0);
            assert_eq!(stats[0].rtp_recv_bitrate, 0);
            assert_eq!(stats[0].rtp_bytes_sent, 0);
            assert_eq!(stats[0].rtp_send_bitrate, 0);
            assert_eq!(stats[0].rtx_bytes_received, 0);
            assert_eq!(stats[0].rtx_recv_bitrate, 0);
            assert_eq!(stats[0].rtx_bytes_sent, 0);
            assert_eq!(stats[0].rtx_send_bitrate, 0);
            assert_eq!(stats[0].probation_bytes_sent, 0);
            assert_eq!(stats[0].probation_send_bitrate, 0);
        });
    }

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

            let data_producer = transport
                .produce_data({
                    let mut options = DataProducerOptions::new_direct();

                    options.label = "foo".to_string();
                    options.protocol = "bar".to_string();
                    options.app_data = AppData::new(CustomAppData { foo: "bar" });

                    options
                })
                .await
                .expect("Failed to produce data");

            let data_consumer = transport
                .consume_data(DataConsumerOptions::new_direct(data_producer.id()))
                .await
                .expect("Failed to consume data");

            let num_messages = 200_usize;
            let mut sent_message_bytes = 0_usize;
            let recv_message_bytes = Arc::new(AtomicUsize::new(0));
            let mut last_sent_message_id = 0_usize;
            let last_recv_message_id = Arc::new(AtomicUsize::new(0));

            let (received_messages_tx, received_messages_rx) = async_oneshot::oneshot::<()>();
            let _handler = data_consumer.on_message({
                let received_messages_tx = Mutex::new(Some(received_messages_tx));
                let recv_message_bytes = Arc::clone(&recv_message_bytes);
                let last_recv_message_id = Arc::clone(&last_recv_message_id);

                move |message| {
                    let id: usize;
                    match message {
                        WebRtcMessage::String(string) => {
                            recv_message_bytes.fetch_add(string.as_bytes().len(), Ordering::SeqCst);
                            id = string.parse().unwrap();
                        }
                        WebRtcMessage::Binary(binary) => {
                            recv_message_bytes.fetch_add(binary.len(), Ordering::SeqCst);
                            id = String::from_utf8(binary.to_vec()).unwrap().parse().unwrap();
                        }
                        WebRtcMessage::EmptyString => {
                            panic!("Unexpected empty messages!");
                        }
                        WebRtcMessage::EmptyBinary => {
                            panic!("Unexpected empty messages!");
                        }
                    };

                    if id == num_messages {
                        let _ = received_messages_tx.lock().take().unwrap().send(());
                    }

                    if id < num_messages / 2 {
                        assert!(matches!(message, &WebRtcMessage::String(_)));
                    } else {
                        assert!(matches!(message, &WebRtcMessage::Binary(_)));
                    }

                    last_recv_message_id.fetch_add(1, Ordering::SeqCst);
                    assert_eq!(id, last_recv_message_id.load(Ordering::SeqCst));
                }
            });

            let direct_data_producer = match &data_producer {
                DataProducer::Direct(direct_data_producer) => direct_data_producer,
                _ => {
                    panic!("Expected direct data producer")
                }
            };

            loop {
                last_sent_message_id += 1;
                let id = last_sent_message_id;

                let message = if id < num_messages / 2 {
                    let content = id.to_string();
                    sent_message_bytes += content.len();
                    WebRtcMessage::String(content)
                } else {
                    let content = Bytes::from(id.to_string().into_bytes());
                    sent_message_bytes += content.len();
                    WebRtcMessage::Binary(content)
                };

                direct_data_producer
                    .send(message)
                    .await
                    .expect("Failed to send message");

                if id == num_messages {
                    break;
                }
            }

            received_messages_rx
                .await
                .expect("Failed tor receive all messages");

            assert_eq!(last_sent_message_id, num_messages);
            assert_eq!(last_recv_message_id.load(Ordering::SeqCst), num_messages);
            assert_eq!(
                recv_message_bytes.load(Ordering::SeqCst),
                sent_message_bytes,
            );

            {
                let stats = data_producer
                    .get_stats()
                    .await
                    .expect("Failed to get stats on data producer");

                assert_eq!(stats.len(), 1);
                assert_eq!(&stats[0].label, data_producer.label());
                assert_eq!(&stats[0].protocol, data_producer.protocol());
                assert_eq!(stats[0].messages_received, num_messages);
                assert_eq!(stats[0].bytes_received, sent_message_bytes);
            }

            {
                let stats = data_consumer
                    .get_stats()
                    .await
                    .expect("Failed to get stats on data consumer");

                assert_eq!(stats.len(), 1);
                assert_eq!(&stats[0].label, data_consumer.label());
                assert_eq!(&stats[0].protocol, data_consumer.protocol());
                assert_eq!(stats[0].messages_sent, num_messages);
                assert_eq!(
                    stats[0].bytes_sent,
                    recv_message_bytes.load(Ordering::SeqCst),
                );
            }
        });
    }

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

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

            drop(transport);

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

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

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

            let (router_close_tx, router_close_rx) = async_oneshot::oneshot::<()>();
            let _handler = transport.on_router_close(move || {
                let _ = router_close_tx.send(());
            });

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

            router_close_rx
                .await
                .expect("Failed to receive router_close event");
            close_rx.await.expect("Failed to receive close event");

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