hedl-xml 2.0.0

HEDL to/from XML conversion
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
// Dweve HEDL - Hierarchical Entity Data Language
//
// Copyright (c) 2025 Dweve IP B.V. and individual contributors.
//
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License in the LICENSE file at the
// root of this repository or at: http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Async API for XML conversion with Tokio
//!
//! This module provides asynchronous versions of all XML conversion functions,
//! enabling high-performance concurrent I/O operations with Tokio.
//!
//! # Features
//!
//! - **Async File I/O**: Read/write XML files without blocking
//! - **Async Streaming**: Stream large XML files incrementally
//! - **Concurrency**: Process multiple files concurrently
//! - **Backpressure**: Built-in flow control for streaming
//!
//! # Examples
//!
//! ## Async file conversion
//!
//! ```no_run
//! use hedl_xml::async_api::{from_xml_file_async, to_xml_file_async};
//! use hedl_xml::{FromXmlConfig, ToXmlConfig};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // Read XML file asynchronously
//!     let doc = from_xml_file_async("input.xml", &FromXmlConfig::default()).await?;
//!
//!     // Process document...
//!
//!     // Write XML file asynchronously
//!     to_xml_file_async(&doc, "output.xml", &ToXmlConfig::default()).await?;
//!
//!     Ok(())
//! }
//! ```
//!
//! ## Async streaming for large files
//!
//! ```no_run
//! use hedl_xml::async_api::from_xml_stream_async;
//! use hedl_xml::streaming::StreamConfig;
//! use tokio::fs::File;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     let file = File::open("large.xml").await?;
//!     let config = StreamConfig::default();
//!
//!     let mut stream = from_xml_stream_async(file, &config).await?;
//!
//!     while let Some(result) = stream.next().await {
//!         match result {
//!             Ok(item) => println!("Processed: {}", item.key),
//!             Err(e) => eprintln!("Error: {}", e),
//!         }
//!     }
//!
//!     Ok(())
//! }
//! ```
//!
//! ## Concurrent processing
//!
//! ```no_run
//! use hedl_xml::async_api::from_xml_file_async;
//! use hedl_xml::FromXmlConfig;
//! use tokio::task;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     let config = FromXmlConfig::default();
//!
//!     // Process multiple files concurrently
//!     let handles: Vec<_> = vec!["file1.xml", "file2.xml", "file3.xml"]
//!         .into_iter()
//!         .map(|path| {
//!             let config = config.clone();
//!             task::spawn(async move {
//!                 from_xml_file_async(path, &config).await
//!             })
//!         })
//!         .collect();
//!
//!     // Wait for all to complete
//!     for handle in handles {
//!         let doc = handle.await??;
//!         // Process document...
//!     }
//!
//!     Ok(())
//! }
//! ```

use crate::streaming::{StreamConfig, StreamItem, StreamPosition};
use crate::{from_xml, to_xml, FromXmlConfig, ToXmlConfig};
use hedl_core::Document;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

// ============================================================================
// Async File I/O Functions
// ============================================================================

/// Read and parse an XML file asynchronously
///
/// This function reads the entire file into memory and parses it. For large files,
/// consider using `from_xml_stream_async` instead.
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::from_xml_file_async;
/// use hedl_xml::FromXmlConfig;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let config = FromXmlConfig::default();
/// let doc = from_xml_file_async("data.xml", &config).await?;
/// println!("Parsed document with {} root items", doc.root.len());
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns an error if the file cannot be read or if XML parsing fails.
pub async fn from_xml_file_async(
    path: impl AsRef<std::path::Path>,
    config: &FromXmlConfig,
) -> Result<Document, String> {
    let contents = tokio::fs::read_to_string(path)
        .await
        .map_err(|e| format!("Failed to read file: {}", e))?;

    from_xml(&contents, config)
}

/// Write a HEDL document to an XML file asynchronously
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::to_xml_file_async;
/// use hedl_xml::ToXmlConfig;
/// use hedl_core::Document;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let doc = Document::new((2, 0));
/// let config = ToXmlConfig::default();
/// to_xml_file_async(&doc, "output.xml", &config).await?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns an error if XML generation fails or if the file cannot be written.
pub async fn to_xml_file_async(
    doc: &Document,
    path: impl AsRef<std::path::Path>,
    config: &ToXmlConfig,
) -> Result<(), String> {
    let xml = to_xml(doc, config)?;

    tokio::fs::write(path, xml)
        .await
        .map_err(|e| format!("Failed to write file: {}", e))?;

    Ok(())
}

// ============================================================================
// Async Reader/Writer Functions
// ============================================================================

/// Parse XML from an async reader
///
/// This function reads the entire content into memory before parsing. For streaming
/// large files, use `from_xml_stream_async` instead.
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::from_xml_reader_async;
/// use hedl_xml::FromXmlConfig;
/// use tokio::fs::File;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let file = File::open("data.xml").await?;
/// let config = FromXmlConfig::default();
/// let doc = from_xml_reader_async(file, &config).await?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns an error if reading fails or if XML parsing fails.
pub async fn from_xml_reader_async<R: AsyncRead + Unpin>(
    mut reader: R,
    config: &FromXmlConfig,
) -> Result<Document, String> {
    let mut contents = String::new();
    reader
        .read_to_string(&mut contents)
        .await
        .map_err(|e| format!("Failed to read XML: {}", e))?;

    from_xml(&contents, config)
}

/// Write XML to an async writer
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::to_xml_writer_async;
/// use hedl_xml::ToXmlConfig;
/// use hedl_core::Document;
/// use tokio::fs::File;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let doc = Document::new((2, 0));
/// let file = File::create("output.xml").await?;
/// let config = ToXmlConfig::default();
/// to_xml_writer_async(&doc, file, &config).await?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns an error if XML generation fails or if writing fails.
pub async fn to_xml_writer_async<W: AsyncWrite + Unpin>(
    doc: &Document,
    mut writer: W,
    config: &ToXmlConfig,
) -> Result<(), String> {
    let xml = to_xml(doc, config)?;

    writer
        .write_all(xml.as_bytes())
        .await
        .map_err(|e| format!("Failed to write XML: {}", e))?;

    writer
        .flush()
        .await
        .map_err(|e| format!("Failed to flush writer: {}", e))?;

    Ok(())
}

// ============================================================================
// Async Streaming Functions
// ============================================================================

/// Create an async streaming XML parser
///
/// This function returns a stream that yields items incrementally, allowing
/// processing of files larger than available RAM.
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::from_xml_stream_async;
/// use hedl_xml::streaming::StreamConfig;
/// use tokio::fs::File;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let file = File::open("large.xml").await?;
/// let config = StreamConfig::default();
///
/// let mut stream = from_xml_stream_async(file, &config).await?;
///
/// let mut count = 0;
/// while let Some(result) = stream.next().await {
///     match result {
///         Ok(_item) => count += 1,
///         Err(e) => eprintln!("Parse error: {}", e),
///     }
/// }
/// println!("Processed {} items", count);
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns an error if the stream cannot be initialized.
pub async fn from_xml_stream_async<R: AsyncRead + Unpin + Send + 'static>(
    reader: R,
    config: &StreamConfig,
) -> Result<AsyncXmlStream<R>, String> {
    AsyncXmlStream::new(reader, config.clone())
}

/// Async streaming XML parser
///
/// This stream yields `Result<StreamItem, String>` as items are parsed.
/// It uses Tokio's async I/O for non-blocking operations.
///
/// # Implementation Note
///
/// This implementation reads all data into memory first, then parses it
/// synchronously and yields items one at a time. For true streaming with
/// backpressure, a more sophisticated implementation using quick-xml's
/// async features would be needed.
pub struct AsyncXmlStream<R: AsyncRead + Unpin> {
    reader: R,
    config: StreamConfig,
    buffer: Vec<u8>,
    /// Byte position in the stream
    byte_position: usize,
    chunk_size: usize,
    /// Parsed items ready to be yielded
    parsed_items: std::collections::VecDeque<Result<StreamItem, String>>,
    /// Whether we've read and parsed all data
    fully_parsed: bool,
    /// Stream position tracking for progress reporting
    stream_position: StreamPosition,
}

impl<R: AsyncRead + Unpin> AsyncXmlStream<R> {
    /// Create a new async XML stream
    pub fn new(reader: R, config: StreamConfig) -> Result<Self, String> {
        let chunk_size = config.buffer_size;
        Ok(AsyncXmlStream {
            reader,
            config,
            buffer: Vec::new(),
            byte_position: 0,
            chunk_size,
            parsed_items: std::collections::VecDeque::new(),
            fully_parsed: false,
            stream_position: StreamPosition::default(),
        })
    }

    /// Get the current stream position for progress tracking
    ///
    /// Returns information about bytes processed and items parsed.
    /// Useful for progress bars and error reporting.
    #[inline]
    pub fn position(&self) -> StreamPosition {
        StreamPosition {
            byte_offset: self.byte_position as u64,
            items_parsed: self.stream_position.items_parsed,
        }
    }

    /// Read the next chunk of data
    async fn read_chunk(&mut self) -> Result<usize, String> {
        let mut chunk = vec![0u8; self.chunk_size];
        let n = self
            .reader
            .read(&mut chunk)
            .await
            .map_err(|e| format!("Failed to read chunk: {}", e))?;

        if n > 0 {
            self.buffer.extend_from_slice(&chunk[..n]);
            self.byte_position += n;
        }

        Ok(n)
    }

    /// Read all data and parse into items
    async fn ensure_parsed(&mut self) -> Result<(), String> {
        if self.fully_parsed {
            return Ok(());
        }

        // Read all remaining data
        loop {
            match self.read_chunk().await {
                Ok(0) => break, // EOF
                Ok(_) => continue,
                Err(e) => return Err(e),
            }
        }

        // Parse complete buffer using sync streaming parser
        if !self.buffer.is_empty() {
            use crate::streaming::from_xml_stream;
            use std::io::Cursor;

            let cursor = Cursor::new(&self.buffer);
            match from_xml_stream(cursor, &self.config) {
                Ok(parser) => {
                    // Collect all items from the parser
                    for result in parser {
                        self.parsed_items
                            .push_back(result.map_err(|e| e.to_string()));
                    }
                }
                Err(e) => {
                    self.parsed_items.push_back(Err(e));
                }
            }
        }

        self.fully_parsed = true;
        Ok(())
    }

    /// Async version of next() - yields the next parsed item
    pub async fn next(&mut self) -> Option<Result<StreamItem, String>> {
        // Ensure we've read and parsed all data
        if let Err(e) = self.ensure_parsed().await {
            return Some(Err(e));
        }

        // Yield the next item from our queue
        let result = self.parsed_items.pop_front();

        // Update items_parsed counter on successful item
        if result.as_ref().is_some_and(|r| r.is_ok()) {
            self.stream_position.items_parsed += 1;
        }

        result
    }
}

// Implement Stream trait for async iteration (requires futures crate)
// For simplicity, we provide next() method instead of implementing Stream

// ============================================================================
// Utility Functions
// ============================================================================

/// Parse XML string asynchronously (runs on tokio threadpool)
///
/// This is useful for CPU-bound parsing that shouldn't block the async runtime.
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::from_xml_async;
/// use hedl_xml::FromXmlConfig;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let xml = r#"<?xml version="1.0"?><hedl><name>test</name></hedl>"#;
/// let config = FromXmlConfig::default();
/// let doc = from_xml_async(xml, &config).await?;
/// # Ok(())
/// # }
/// ```
pub async fn from_xml_async(xml: &str, config: &FromXmlConfig) -> Result<Document, String> {
    let xml = xml.to_string();
    let config = config.clone();

    tokio::task::spawn_blocking(move || from_xml(&xml, &config))
        .await
        .map_err(|e| format!("Task join error: {}", e))?
}

/// Convert HEDL to XML string asynchronously (runs on tokio threadpool)
///
/// This is useful for CPU-bound conversion that shouldn't block the async runtime.
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::to_xml_async;
/// use hedl_xml::ToXmlConfig;
/// use hedl_core::Document;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let doc = Document::new((2, 0));
/// let config = ToXmlConfig::default();
/// let xml = to_xml_async(&doc, &config).await?;
/// # Ok(())
/// # }
/// ```
pub async fn to_xml_async(doc: &Document, config: &ToXmlConfig) -> Result<String, String> {
    let doc = doc.clone();
    let config = config.clone();

    tokio::task::spawn_blocking(move || to_xml(&doc, &config))
        .await
        .map_err(|e| format!("Task join error: {}", e))?
}

// ============================================================================
// Batch Processing Functions
// ============================================================================

/// Process multiple XML files concurrently
///
/// Returns results in the same order as input paths.
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::from_xml_files_concurrent;
/// use hedl_xml::FromXmlConfig;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let paths = vec!["file1.xml", "file2.xml", "file3.xml"];
/// let config = FromXmlConfig::default();
///
/// let results = from_xml_files_concurrent(&paths, &config, 4).await;
///
/// for (path, result) in paths.iter().zip(results.iter()) {
///     match result {
///         Ok(doc) => println!("{}: {} items", path, doc.root.len()),
///         Err(e) => eprintln!("{}: error - {}", path, e),
///     }
/// }
/// # Ok(())
/// # }
/// ```
///
/// # Arguments
///
/// * `paths` - Iterator of file paths to process
/// * `config` - Configuration for XML parsing
/// * `concurrency` - Maximum number of concurrent operations
pub async fn from_xml_files_concurrent<'a, I, P>(
    paths: I,
    config: &FromXmlConfig,
    concurrency: usize,
) -> Vec<Result<Document, String>>
where
    I: IntoIterator<Item = P>,
    P: AsRef<std::path::Path> + Send + 'a,
{
    use tokio::task::JoinSet;

    let mut set: JoinSet<(usize, Result<Document, String>)> = JoinSet::new();
    let config = config.clone();
    let paths_vec: Vec<_> = paths.into_iter().collect();
    let total = paths_vec.len();

    // Pre-allocate results with placeholders
    let mut results: Vec<Option<Result<Document, String>>> = vec![None; total];
    let mut paths_iter = paths_vec.into_iter().enumerate();
    let mut pending = 0;

    // Fill initial batch up to concurrency limit
    for _ in 0..concurrency {
        if let Some((idx, path)) = paths_iter.next() {
            let path = path.as_ref().to_path_buf();
            let config = config.clone();
            set.spawn(async move { (idx, from_xml_file_async(&path, &config).await) });
            pending += 1;
        } else {
            break;
        }
    }

    // Process remaining items, maintaining concurrency level
    while pending > 0 {
        if let Some(join_result) = set.join_next().await {
            match join_result {
                Ok((idx, doc_result)) => {
                    results[idx] = Some(doc_result);
                }
                Err(e) => {
                    // Task panicked - find first None slot and fill it
                    if let Some(slot) = results.iter_mut().find(|r| r.is_none()) {
                        *slot = Some(Err(format!("Task error: {}", e)));
                    }
                }
            }
            pending -= 1;

            // Add next item if available
            if let Some((idx, path)) = paths_iter.next() {
                let path = path.as_ref().to_path_buf();
                let config = config.clone();
                set.spawn(async move { (idx, from_xml_file_async(&path, &config).await) });
                pending += 1;
            }
        }
    }

    // Convert to final results, unwrapping Options (all should be Some now)
    results
        .into_iter()
        .map(|opt| opt.unwrap_or_else(|| Err("Missing result".to_string())))
        .collect()
}

/// Write multiple documents to XML files concurrently
///
/// # Examples
///
/// ```no_run
/// use hedl_xml::async_api::to_xml_files_concurrent;
/// use hedl_xml::ToXmlConfig;
/// use hedl_core::Document;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let docs = vec![
///     Document::new((2, 0)),
///     Document::new((2, 0)),
/// ];
/// let paths = vec!["out1.xml", "out2.xml"];
/// let config = ToXmlConfig::default();
///
/// let results = to_xml_files_concurrent(
///     docs.iter().zip(paths.iter()),
///     &config,
///     4
/// ).await;
///
/// for (i, result) in results.iter().enumerate() {
///     match result {
///         Ok(_) => println!("File {} written successfully", i),
///         Err(e) => eprintln!("File {} error: {}", i, e),
///     }
/// }
/// # Ok(())
/// # }
/// ```
pub async fn to_xml_files_concurrent<'a, I, P>(
    docs_and_paths: I,
    config: &ToXmlConfig,
    concurrency: usize,
) -> Vec<Result<(), String>>
where
    I: IntoIterator<Item = (&'a Document, P)>,
    P: AsRef<std::path::Path> + Send + 'a,
{
    use tokio::task::JoinSet;

    let mut set: JoinSet<(usize, Result<(), String>)> = JoinSet::new();
    let config = config.clone();
    let docs_and_paths_vec: Vec<_> = docs_and_paths.into_iter().collect();
    let total = docs_and_paths_vec.len();

    // Pre-allocate results with placeholders
    let mut results: Vec<Option<Result<(), String>>> = vec![None; total];
    let mut iter = docs_and_paths_vec.into_iter().enumerate();
    let mut pending = 0;

    // Fill initial batch up to concurrency limit
    for _ in 0..concurrency {
        if let Some((idx, (doc, path))) = iter.next() {
            let doc = doc.clone();
            let path = path.as_ref().to_path_buf();
            let config = config.clone();
            set.spawn(async move { (idx, to_xml_file_async(&doc, &path, &config).await) });
            pending += 1;
        } else {
            break;
        }
    }

    // Process remaining items, maintaining concurrency level
    while pending > 0 {
        if let Some(join_result) = set.join_next().await {
            match join_result {
                Ok((idx, write_result)) => {
                    results[idx] = Some(write_result);
                }
                Err(e) => {
                    // Task panicked - find first None slot and fill it
                    if let Some(slot) = results.iter_mut().find(|r| r.is_none()) {
                        *slot = Some(Err(format!("Task error: {}", e)));
                    }
                }
            }
            pending -= 1;

            // Add next item if available
            if let Some((idx, (doc, path))) = iter.next() {
                let doc = doc.clone();
                let path = path.as_ref().to_path_buf();
                let config = config.clone();
                set.spawn(async move { (idx, to_xml_file_async(&doc, &path, &config).await) });
                pending += 1;
            }
        }
    }

    // Convert to final results, unwrapping Options (all should be Some now)
    results
        .into_iter()
        .map(|opt| opt.unwrap_or_else(|| Err("Missing result".to_string())))
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use hedl_core::{Item, Value};
    use std::io::Cursor;

    // ==================== File I/O tests ====================

    #[tokio::test]
    async fn test_from_xml_file_async_not_found() {
        let config = FromXmlConfig::default();
        let result = from_xml_file_async("/nonexistent/file.xml", &config).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("Failed to read file"));
    }

    #[tokio::test]
    async fn test_to_xml_file_async_invalid_path() {
        let doc = Document::new((2, 0));
        let config = ToXmlConfig::default();
        let result = to_xml_file_async(&doc, "/invalid/\0/path.xml", &config).await;
        assert!(result.is_err());
    }

    // ==================== Reader/Writer tests ====================

    #[tokio::test]
    async fn test_from_xml_reader_async_valid() {
        let xml = r#"<?xml version="1.0"?><hedl><val>42</val></hedl>"#;
        let cursor = Cursor::new(xml.as_bytes());
        let config = FromXmlConfig::default();

        let doc = from_xml_reader_async(cursor, &config).await.unwrap();
        assert_eq!(
            doc.root.get("val").and_then(|i| i.as_scalar()),
            Some(&Value::Int(42))
        );
    }

    #[tokio::test]
    async fn test_from_xml_reader_async_empty() {
        let xml = "";
        let cursor = Cursor::new(xml.as_bytes());
        let config = FromXmlConfig::default();

        let doc = from_xml_reader_async(cursor, &config).await.unwrap();
        assert!(doc.root.is_empty());
    }

    #[tokio::test]
    async fn test_to_xml_writer_async_valid() {
        let mut doc = Document::new((2, 0));
        doc.root
            .insert("val".to_string(), Item::Scalar(Value::Int(42)));

        let mut buffer = Vec::new();
        let config = ToXmlConfig::default();

        to_xml_writer_async(&doc, &mut buffer, &config)
            .await
            .unwrap();

        let xml = String::from_utf8(buffer).unwrap();
        assert!(xml.contains("<val>42</val>"));
    }

    #[tokio::test]
    async fn test_to_xml_writer_async_empty() {
        let doc = Document::new((2, 0));
        let mut buffer = Vec::new();
        let config = ToXmlConfig::default();

        to_xml_writer_async(&doc, &mut buffer, &config)
            .await
            .unwrap();

        let xml = String::from_utf8(buffer).unwrap();
        assert!(xml.contains("<?xml"));
        assert!(xml.contains("<hedl"));
    }

    // ==================== Async string parsing tests ====================

    #[tokio::test]
    async fn test_from_xml_async_valid() {
        let xml = r#"<?xml version="1.0"?><hedl><name>test</name></hedl>"#;
        let config = FromXmlConfig::default();

        let doc = from_xml_async(xml, &config).await.unwrap();
        assert_eq!(
            doc.root.get("name").and_then(|i| i.as_scalar()),
            Some(&Value::String("test".to_string().into()))
        );
    }

    #[tokio::test]
    async fn test_from_xml_async_invalid() {
        let xml = r#"</invalid>"#;
        let config = FromXmlConfig::default();

        let result = from_xml_async(xml, &config).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_to_xml_async_valid() {
        let mut doc = Document::new((2, 0));
        doc.root
            .insert("val".to_string(), Item::Scalar(Value::Int(123)));

        let config = ToXmlConfig::default();
        let xml = to_xml_async(&doc, &config).await.unwrap();

        assert!(xml.contains("<val>123</val>"));
    }

    #[tokio::test]
    async fn test_to_xml_async_empty() {
        let doc = Document::new((2, 0));
        let config = ToXmlConfig::default();

        let xml = to_xml_async(&doc, &config).await.unwrap();
        assert!(xml.contains("<?xml"));
    }

    // ==================== Concurrent processing tests ====================

    #[tokio::test]
    async fn test_from_xml_files_concurrent_empty() {
        let paths: Vec<&str> = vec![];
        let config = FromXmlConfig::default();

        let results = from_xml_files_concurrent(&paths, &config, 4).await;
        assert!(results.is_empty());
    }

    #[tokio::test]
    async fn test_to_xml_files_concurrent_empty() {
        let docs_and_paths: Vec<(&Document, &str)> = vec![];
        let config = ToXmlConfig::default();

        let results = to_xml_files_concurrent(docs_and_paths, &config, 4).await;
        assert!(results.is_empty());
    }

    // ==================== Edge cases ====================

    #[tokio::test]
    async fn test_from_xml_reader_async_unicode() {
        let xml = r#"<?xml version="1.0" encoding="UTF-8"?>
            <hedl><name>héllo 世界</name></hedl>"#;
        let cursor = Cursor::new(xml.as_bytes());
        let config = FromXmlConfig::default();

        let doc = from_xml_reader_async(cursor, &config).await.unwrap();
        assert_eq!(
            doc.root.get("name").and_then(|i| i.as_scalar()),
            Some(&Value::String("héllo 世界".to_string().into()))
        );
    }

    #[tokio::test]
    async fn test_from_xml_reader_async_large_value() {
        let large_string = "x".repeat(10000);
        let xml = format!(
            r#"<?xml version="1.0"?><hedl><val>{}</val></hedl>"#,
            large_string
        );
        let cursor = Cursor::new(xml.as_bytes());
        let config = FromXmlConfig::default();

        let doc = from_xml_reader_async(cursor, &config).await.unwrap();
        assert_eq!(
            doc.root.get("val").and_then(|i| i.as_scalar()),
            Some(&Value::String(large_string.into()))
        );
    }

    #[tokio::test]
    async fn test_round_trip_async() {
        let mut doc = Document::new((2, 0));
        doc.root
            .insert("bool_val".to_string(), Item::Scalar(Value::Bool(true)));
        doc.root
            .insert("int_val".to_string(), Item::Scalar(Value::Int(42)));
        doc.root.insert(
            "string_val".to_string(),
            Item::Scalar(Value::String("hello".to_string().into())),
        );

        let config_to = ToXmlConfig::default();
        let xml = to_xml_async(&doc, &config_to).await.unwrap();

        let config_from = FromXmlConfig::default();
        let doc2 = from_xml_async(&xml, &config_from).await.unwrap();

        assert_eq!(
            doc2.root.get("bool_val").and_then(|i| i.as_scalar()),
            Some(&Value::Bool(true))
        );
        assert_eq!(
            doc2.root.get("int_val").and_then(|i| i.as_scalar()),
            Some(&Value::Int(42))
        );
        assert_eq!(
            doc2.root.get("string_val").and_then(|i| i.as_scalar()),
            Some(&Value::String("hello".to_string().into()))
        );
    }

    // ==================== Concurrency and parallelism tests ====================

    #[tokio::test]
    async fn test_concurrent_parsing() {
        let xml1 = r#"<?xml version="1.0"?><hedl><id>1</id></hedl>"#;
        let xml2 = r#"<?xml version="1.0"?><hedl><id>2</id></hedl>"#;
        let xml3 = r#"<?xml version="1.0"?><hedl><id>3</id></hedl>"#;

        let config = FromXmlConfig::default();

        let (r1, r2, r3) = tokio::join!(
            from_xml_async(xml1, &config),
            from_xml_async(xml2, &config),
            from_xml_async(xml3, &config)
        );

        assert!(r1.is_ok());
        assert!(r2.is_ok());
        assert!(r3.is_ok());

        assert_eq!(
            r1.unwrap().root.get("id").and_then(|i| i.as_scalar()),
            Some(&Value::Int(1))
        );
        assert_eq!(
            r2.unwrap().root.get("id").and_then(|i| i.as_scalar()),
            Some(&Value::Int(2))
        );
        assert_eq!(
            r3.unwrap().root.get("id").and_then(|i| i.as_scalar()),
            Some(&Value::Int(3))
        );
    }

    #[tokio::test]
    async fn test_concurrent_generation() {
        let mut doc1 = Document::new((2, 0));
        doc1.root
            .insert("id".to_string(), Item::Scalar(Value::Int(1)));

        let mut doc2 = Document::new((2, 0));
        doc2.root
            .insert("id".to_string(), Item::Scalar(Value::Int(2)));

        let config = ToXmlConfig::default();

        let (r1, r2) = tokio::join!(to_xml_async(&doc1, &config), to_xml_async(&doc2, &config));

        assert!(r1.is_ok());
        assert!(r2.is_ok());

        assert!(r1.unwrap().contains("<id>1</id>"));
        assert!(r2.unwrap().contains("<id>2</id>"));
    }
}