1use crate::{GeneratorError, Result, analysis::SchemaAnalysis, streaming::StreamingConfig};
2use proc_macro2::TokenStream;
3use quote::{format_ident, quote};
4use std::collections::BTreeMap;
5use std::path::PathBuf;
6
7#[derive(Clone)]
9struct DiscriminatedVariantInfo {
10 discriminator_field: String,
12 discriminator_value: String,
14 is_parent_untagged: bool,
16}
17
18#[derive(Debug, Clone)]
19pub struct GeneratorConfig {
20 pub spec_path: PathBuf,
22 pub output_dir: PathBuf,
24 pub module_name: String,
26 pub enable_sse_client: bool,
28 pub enable_async_client: bool,
30 pub enable_specta: bool,
32 pub type_mappings: BTreeMap<String, String>,
34 pub streaming_config: Option<StreamingConfig>,
36 pub nullable_field_overrides: BTreeMap<String, bool>,
39 pub schema_extensions: Vec<PathBuf>,
42 pub http_client_config: Option<crate::http_config::HttpClientConfig>,
44 pub retry_config: Option<crate::http_config::RetryConfig>,
46 pub tracing_enabled: bool,
48 pub auth_config: Option<crate::http_config::AuthConfig>,
50 pub enable_registry: bool,
52 pub registry_only: bool,
54}
55
56impl Default for GeneratorConfig {
57 fn default() -> Self {
58 Self {
59 spec_path: "openapi.json".into(),
60 output_dir: "src/gen".into(),
61 module_name: "api_types".to_string(),
62 enable_sse_client: true,
63 enable_async_client: true,
64 enable_specta: false,
65 type_mappings: default_type_mappings(),
66 streaming_config: None,
67 nullable_field_overrides: BTreeMap::new(),
68 schema_extensions: Vec::new(),
69 http_client_config: None,
70 retry_config: None,
71 tracing_enabled: true,
72 auth_config: None,
73 enable_registry: false,
74 registry_only: false,
75 }
76 }
77}
78
79pub fn default_type_mappings() -> BTreeMap<String, String> {
80 let mut mappings = BTreeMap::new();
81 mappings.insert("integer".to_string(), "i64".to_string());
82 mappings.insert("number".to_string(), "f64".to_string());
83 mappings.insert("string".to_string(), "String".to_string());
84 mappings.insert("boolean".to_string(), "bool".to_string());
85 mappings
86}
87
88#[derive(Debug, Clone)]
90pub struct GeneratedFile {
91 pub path: PathBuf,
93 pub content: String,
95}
96
97#[derive(Debug, Clone)]
99pub struct GenerationResult {
100 pub files: Vec<GeneratedFile>,
102 pub mod_file: GeneratedFile,
104}
105
106pub struct CodeGenerator {
107 config: GeneratorConfig,
108}
109
110impl CodeGenerator {
111 pub fn new(config: GeneratorConfig) -> Self {
112 Self { config }
113 }
114
115 pub fn config(&self) -> &GeneratorConfig {
117 &self.config
118 }
119
120 pub fn generate_all(&self, analysis: &mut SchemaAnalysis) -> Result<GenerationResult> {
122 let mut files = Vec::new();
123
124 if !self.config.registry_only {
125 let types_content = self.generate_types(analysis)?;
127 files.push(GeneratedFile {
128 path: "types.rs".into(),
129 content: types_content,
130 });
131
132 if let Some(ref streaming_config) = self.config.streaming_config {
134 let streaming_content =
135 self.generate_streaming_client(streaming_config, analysis)?;
136 files.push(GeneratedFile {
137 path: "streaming.rs".into(),
138 content: streaming_content,
139 });
140 }
141
142 if self.config.enable_async_client {
144 let http_content = self.generate_http_client(analysis)?;
145 files.push(GeneratedFile {
146 path: "client.rs".into(),
147 content: http_content,
148 });
149 }
150 }
151
152 if self.config.enable_registry || self.config.registry_only {
154 let registry_content = self.generate_registry(analysis)?;
155 files.push(GeneratedFile {
156 path: "registry.rs".into(),
157 content: registry_content,
158 });
159 }
160
161 let mod_content = self.generate_mod_file(&files)?;
163 let mod_file = GeneratedFile {
164 path: "mod.rs".into(),
165 content: mod_content,
166 };
167
168 Ok(GenerationResult { files, mod_file })
169 }
170
171 pub fn generate(&self, analysis: &mut SchemaAnalysis) -> Result<String> {
173 self.generate_types(analysis)
174 }
175
176 fn generate_types(&self, analysis: &mut SchemaAnalysis) -> Result<String> {
178 let mut type_definitions = TokenStream::new();
179
180 let mut discriminated_variant_info: BTreeMap<String, DiscriminatedVariantInfo> =
183 BTreeMap::new();
184
185 let mut sorted_schemas: Vec<_> = analysis.schemas.iter().collect();
187 sorted_schemas.sort_by_key(|(name, _)| name.as_str());
188
189 for (_parent_name, schema) in sorted_schemas {
190 if let crate::analysis::SchemaType::DiscriminatedUnion {
191 variants,
192 discriminator_field,
193 } = &schema.schema_type
194 {
195 let is_parent_untagged =
197 self.should_use_untagged_discriminated_union(schema, analysis);
198
199 for variant in variants {
200 if let Some(variant_schema) = analysis.schemas.get(&variant.type_name) {
203 if let crate::analysis::SchemaType::Object { properties, .. } =
204 &variant_schema.schema_type
205 {
206 if properties.contains_key(discriminator_field) {
207 discriminated_variant_info.insert(
208 variant.type_name.clone(),
209 DiscriminatedVariantInfo {
210 discriminator_field: discriminator_field.clone(),
211 discriminator_value: variant.discriminator_value.clone(),
212 is_parent_untagged,
213 },
214 );
215 }
216 }
217 }
218 }
219 }
220 }
221
222 let generation_order = analysis.dependencies.topological_sort()?;
224
225 let mut processed = std::collections::HashSet::new();
227
228 for schema_name in generation_order {
230 if let Some(schema) = analysis.schemas.get(&schema_name) {
231 let type_def =
232 self.generate_type_definition(schema, analysis, &discriminated_variant_info)?;
233 if !type_def.is_empty() {
234 type_definitions.extend(type_def);
235 }
236 processed.insert(schema_name);
237 }
238 }
239
240 let mut remaining_schemas: Vec<_> = analysis
243 .schemas
244 .iter()
245 .filter(|(name, _)| !processed.contains(*name))
246 .collect();
247 remaining_schemas.sort_by_key(|(name, _)| name.as_str());
248
249 for (_schema_name, schema) in remaining_schemas {
250 let type_def =
251 self.generate_type_definition(schema, analysis, &discriminated_variant_info)?;
252 if !type_def.is_empty() {
253 type_definitions.extend(type_def);
254 }
255 }
256
257 let generated = quote! {
259 #![allow(clippy::large_enum_variant)]
265 #![allow(clippy::format_in_format_args)]
266 #![allow(clippy::let_unit_value)]
267 #![allow(unreachable_patterns)]
268
269 use serde::{Deserialize, Serialize};
270
271 #type_definitions
272 };
273
274 let syntax_tree = syn::parse2::<syn::File>(generated).map_err(|e| {
276 GeneratorError::CodeGenError(format!("Failed to parse generated code: {e}"))
277 })?;
278
279 let formatted = prettyplease::unparse(&syntax_tree);
280
281 Ok(formatted)
282 }
283
284 fn generate_streaming_client(
286 &self,
287 streaming_config: &StreamingConfig,
288 analysis: &SchemaAnalysis,
289 ) -> Result<String> {
290 let mut client_code = TokenStream::new();
291
292 let imports = quote! {
294 #![allow(clippy::format_in_format_args)]
299 #![allow(clippy::let_unit_value)]
300 #![allow(unused_mut)]
301
302 use super::types::*;
303 use async_trait::async_trait;
304 use futures_util::{Stream, StreamExt};
305 use std::pin::Pin;
306 use std::time::Duration;
307 use reqwest::header::{HeaderMap, HeaderValue};
308 use tracing::{debug, error, info, warn, instrument};
309 };
310 client_code.extend(imports);
311
312 if streaming_config.generate_client {
314 let error_types = self.generate_streaming_error_types()?;
315 client_code.extend(error_types);
316 }
317
318 for endpoint in &streaming_config.endpoints {
320 let trait_code = self.generate_endpoint_trait(endpoint, analysis)?;
321 client_code.extend(trait_code);
322 }
323
324 if streaming_config.generate_client {
326 let client_impl = self.generate_streaming_client_impl(streaming_config, analysis)?;
327 client_code.extend(client_impl);
328 }
329
330 if streaming_config.event_parser_helpers {
332 let parser_code = self.generate_sse_parser_utilities(streaming_config)?;
333 client_code.extend(parser_code);
334 }
335
336 if let Some(reconnect_config) = &streaming_config.reconnection_config {
338 let reconnect_code = self.generate_reconnection_utilities(reconnect_config)?;
339 client_code.extend(reconnect_code);
340 }
341
342 let syntax_tree = syn::parse2::<syn::File>(client_code).map_err(|e| {
343 GeneratorError::CodeGenError(format!("Failed to parse streaming client code: {e}"))
344 })?;
345
346 Ok(prettyplease::unparse(&syntax_tree))
347 }
348
349 pub fn generate_http_client(&self, analysis: &SchemaAnalysis) -> Result<String> {
351 let error_types = self.generate_http_error_types();
352 let client_struct = self.generate_http_client_struct();
353 let operation_methods = self.generate_operation_methods(analysis);
354
355 let generated = quote! {
356 #![allow(clippy::format_in_format_args)]
361 #![allow(clippy::let_unit_value)]
362
363 use super::types::*;
364
365 #error_types
366
367 #client_struct
368
369 #operation_methods
370 };
371
372 let syntax_tree = syn::parse2::<syn::File>(generated).map_err(|e| {
373 GeneratorError::CodeGenError(format!("Failed to parse HTTP client code: {e}"))
374 })?;
375
376 Ok(prettyplease::unparse(&syntax_tree))
377 }
378
379 fn generate_http_error_types(&self) -> TokenStream {
381 quote! {
382 use thiserror::Error;
383
384 #[derive(Error, Debug)]
386 pub enum HttpError {
387 #[error("Network error: {0}")]
389 Network(#[from] reqwest::Error),
390
391 #[error("Middleware error: {0}")]
393 Middleware(#[from] reqwest_middleware::Error),
394
395 #[error("Failed to serialize request: {0}")]
397 Serialization(String),
398
399 #[error("Failed to deserialize response: {0}")]
401 Deserialization(String),
402
403 #[error("HTTP error {status}: {message}")]
405 Http {
406 status: u16,
407 message: String,
408 body: Option<String>,
409 },
410
411 #[error("Authentication error: {0}")]
413 Auth(String),
414
415 #[error("Request timeout")]
417 Timeout,
418
419 #[error("Configuration error: {0}")]
421 Config(String),
422
423 #[error("{0}")]
425 Other(String),
426 }
427
428 impl HttpError {
429 pub fn from_status(status: u16, message: impl Into<String>, body: Option<String>) -> Self {
431 Self::Http {
432 status,
433 message: message.into(),
434 body,
435 }
436 }
437
438 pub fn serialization_error(error: impl std::fmt::Display) -> Self {
440 Self::Serialization(error.to_string())
441 }
442
443 pub fn deserialization_error(error: impl std::fmt::Display) -> Self {
445 Self::Deserialization(error.to_string())
446 }
447
448 pub fn is_client_error(&self) -> bool {
450 matches!(self, Self::Http { status, .. } if *status >= 400 && *status < 500)
451 }
452
453 pub fn is_server_error(&self) -> bool {
455 matches!(self, Self::Http { status, .. } if *status >= 500 && *status < 600)
456 }
457
458 pub fn is_retryable(&self) -> bool {
460 match self {
461 Self::Network(_) => true,
462 Self::Middleware(_) => true,
463 Self::Timeout => true,
464 Self::Http { status, .. } => {
465 matches!(status, 429 | 500 | 502 | 503 | 504)
467 }
468 _ => false,
469 }
470 }
471 }
472
473 pub type HttpResult<T> = Result<T, HttpError>;
475 }
476 }
477
478 fn generate_mod_file(&self, files: &[GeneratedFile]) -> Result<String> {
480 let mut module_declarations = Vec::new();
481 let mut pub_uses = Vec::new();
482
483 for file in files {
484 if let Some(module_name) = file.path.file_stem().and_then(|s| s.to_str()) {
485 if module_name != "mod" {
486 module_declarations.push(format!("pub mod {module_name};"));
487 pub_uses.push(format!("pub use {module_name}::*;"));
488 }
489 }
490 }
491
492 let content = format!(
493 r#"//! Generated API modules
494//!
495//! This module exports all generated API types and clients.
496//! Do not edit manually - regenerate using the appropriate script.
497
498#![allow(unused_imports)]
499
500{}
501
502{}
503"#,
504 module_declarations.join("\n"),
505 pub_uses.join("\n")
506 );
507
508 Ok(content)
509 }
510
511 pub fn write_files(&self, result: &GenerationResult) -> Result<()> {
513 use std::fs;
514
515 fs::create_dir_all(&self.config.output_dir)?;
517
518 for file in &result.files {
520 let file_path = self.config.output_dir.join(&file.path);
521 fs::write(&file_path, &file.content)?;
522 }
523
524 let mod_path = self.config.output_dir.join(&result.mod_file.path);
526 fs::write(&mod_path, &result.mod_file.content)?;
527
528 Ok(())
529 }
530
531 fn generate_type_definition(
532 &self,
533 schema: &crate::analysis::AnalyzedSchema,
534 analysis: &crate::analysis::SchemaAnalysis,
535 discriminated_variant_info: &BTreeMap<String, DiscriminatedVariantInfo>,
536 ) -> Result<TokenStream> {
537 use crate::analysis::SchemaType;
538
539 match &schema.schema_type {
540 SchemaType::Primitive { rust_type } => {
541 self.generate_type_alias(schema, rust_type)
543 }
544 SchemaType::StringEnum { values } => self.generate_string_enum(schema, values),
545 SchemaType::ExtensibleEnum { known_values } => {
546 self.generate_extensible_enum(schema, known_values)
547 }
548 SchemaType::Object {
549 properties,
550 required,
551 additional_properties,
552 } => self.generate_struct(
553 schema,
554 properties,
555 required,
556 *additional_properties,
557 analysis,
558 discriminated_variant_info.get(&schema.name),
559 ),
560 SchemaType::DiscriminatedUnion {
561 discriminator_field,
562 variants,
563 } => {
564 if self.should_use_untagged_discriminated_union(schema, analysis) {
566 let schema_refs: Vec<crate::analysis::SchemaRef> = variants
568 .iter()
569 .map(|v| crate::analysis::SchemaRef {
570 target: v.type_name.clone(),
571 nullable: false,
572 })
573 .collect();
574 self.generate_union_enum(schema, &schema_refs)
575 } else {
576 self.generate_discriminated_enum(
577 schema,
578 discriminator_field,
579 variants,
580 analysis,
581 )
582 }
583 }
584 SchemaType::Union { variants } => self.generate_union_enum(schema, variants),
585 SchemaType::Reference { target } => {
586 if schema.name != *target {
589 let alias_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
591 let target_type = format_ident!("{}", self.to_rust_type_name(target));
592
593 let doc_comment = if let Some(desc) = &schema.description {
594 quote! { #[doc = #desc] }
595 } else {
596 TokenStream::new()
597 };
598
599 Ok(quote! {
600 #doc_comment
601 pub type #alias_name = #target_type;
602 })
603 } else {
604 Ok(TokenStream::new())
606 }
607 }
608 SchemaType::Array { item_type } => {
609 let array_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
617
618 if let SchemaType::Reference { target } = item_type.as_ref() {
620 if let Some(info) = discriminated_variant_info.get(target) {
621 if !info.is_parent_untagged {
622 let wrapper_name =
624 format_ident!("{}Item", self.to_rust_type_name(&schema.name));
625 let variant_type = format_ident!("{}", self.to_rust_type_name(target));
626 let disc_field = &info.discriminator_field;
627 let disc_value = &info.discriminator_value;
628
629 let doc_comment = if let Some(desc) = &schema.description {
630 quote! { #[doc = #desc] }
631 } else {
632 TokenStream::new()
633 };
634
635 return Ok(quote! {
636 #[derive(Debug, Clone, Deserialize, Serialize)]
640 #[serde(tag = #disc_field)]
641 pub enum #wrapper_name {
642 #[serde(rename = #disc_value)]
643 #variant_type(#variant_type),
644 }
645 #doc_comment
646 pub type #array_name = Vec<#wrapper_name>;
647 });
648 }
649 }
650 }
651
652 let inner_type = self.generate_array_item_type(item_type, analysis);
653
654 let doc_comment = if let Some(desc) = &schema.description {
655 quote! { #[doc = #desc] }
656 } else {
657 TokenStream::new()
658 };
659
660 Ok(quote! {
661 #doc_comment
662 pub type #array_name = Vec<#inner_type>;
663 })
664 }
665 SchemaType::Composition { schemas } => {
666 self.generate_composition_struct(schema, schemas)
667 }
668 }
669 }
670
671 fn generate_type_alias(
672 &self,
673 schema: &crate::analysis::AnalyzedSchema,
674 rust_type: &str,
675 ) -> Result<TokenStream> {
676 let type_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
677
678 let base_type = if rust_type.contains("::") {
680 let parts: Vec<&str> = rust_type.split("::").collect();
681 if parts.len() == 2 {
682 let module = format_ident!("{}", parts[0]);
683 let type_name_part = format_ident!("{}", parts[1]);
684 quote! { #module::#type_name_part }
685 } else {
686 let path_parts: Vec<_> = parts.iter().map(|p| format_ident!("{}", p)).collect();
688 quote! { #(#path_parts)::* }
689 }
690 } else {
691 let simple_type = format_ident!("{}", rust_type);
692 quote! { #simple_type }
693 };
694
695 let doc_comment = if let Some(desc) = &schema.description {
696 let sanitized_desc = self.sanitize_doc_comment(desc);
697 quote! { #[doc = #sanitized_desc] }
698 } else {
699 TokenStream::new()
700 };
701
702 Ok(quote! {
703 #doc_comment
704 pub type #type_name = #base_type;
705 })
706 }
707
708 fn generate_extensible_enum(
709 &self,
710 schema: &crate::analysis::AnalyzedSchema,
711 known_values: &[String],
712 ) -> Result<TokenStream> {
713 let enum_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
714
715 let doc_comment = if let Some(desc) = &schema.description {
716 quote! { #[doc = #desc] }
717 } else {
718 TokenStream::new()
719 };
720
721 let known_variants = known_values.iter().map(|value| {
726 let variant_name = self.to_rust_enum_variant(value);
727 let variant_ident = format_ident!("{}", variant_name);
728 quote! {
729 #variant_ident,
730 }
731 });
732
733 let match_arms_de = known_values.iter().map(|value| {
734 let variant_name = self.to_rust_enum_variant(value);
735 let variant_ident = format_ident!("{}", variant_name);
736 quote! {
737 #value => Ok(#enum_name::#variant_ident),
738 }
739 });
740
741 let match_arms_ser = known_values.iter().map(|value| {
742 let variant_name = self.to_rust_enum_variant(value);
743 let variant_ident = format_ident!("{}", variant_name);
744 quote! {
745 #enum_name::#variant_ident => #value,
746 }
747 });
748
749 let derives = if self.config.enable_specta {
750 quote! {
751 #[derive(Debug, Clone, PartialEq, Eq)]
752 #[cfg_attr(feature = "specta", derive(specta::Type))]
753 }
754 } else {
755 quote! {
756 #[derive(Debug, Clone, PartialEq, Eq)]
757 }
758 };
759
760 Ok(quote! {
761 #doc_comment
762 #derives
763 pub enum #enum_name {
764 #(#known_variants)*
765 Custom(String),
767 }
768
769 impl<'de> serde::Deserialize<'de> for #enum_name {
770 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
771 where
772 D: serde::Deserializer<'de>,
773 {
774 let value = String::deserialize(deserializer)?;
775 match value.as_str() {
776 #(#match_arms_de)*
777 _ => Ok(#enum_name::Custom(value)),
778 }
779 }
780 }
781
782 impl serde::Serialize for #enum_name {
783 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
784 where
785 S: serde::Serializer,
786 {
787 let value = match self {
788 #(#match_arms_ser)*
789 #enum_name::Custom(s) => s.as_str(),
790 };
791 serializer.serialize_str(value)
792 }
793 }
794 })
795 }
796
797 fn generate_string_enum(
798 &self,
799 schema: &crate::analysis::AnalyzedSchema,
800 values: &[String],
801 ) -> Result<TokenStream> {
802 let enum_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
803
804 let default_value = schema
806 .default
807 .as_ref()
808 .and_then(|v| v.as_str())
809 .map(|s| s.to_string());
810
811 let variants = values.iter().enumerate().map(|(i, value)| {
812 let variant_name = self.to_rust_enum_variant(value);
814 let variant_ident = format_ident!("{}", variant_name);
815
816 let is_default = if let Some(ref default) = default_value {
818 value == default
819 } else {
820 i == 0 };
822
823 if is_default {
824 quote! {
825 #[default]
826 #[serde(rename = #value)]
827 #variant_ident,
828 }
829 } else {
830 quote! {
831 #[serde(rename = #value)]
832 #variant_ident,
833 }
834 }
835 });
836
837 let doc_comment = if let Some(desc) = &schema.description {
838 quote! { #[doc = #desc] }
839 } else {
840 TokenStream::new()
841 };
842
843 let derives = if self.config.enable_specta {
845 quote! {
846 #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, Default)]
847 #[cfg_attr(feature = "specta", derive(specta::Type))]
848 }
849 } else {
850 quote! {
851 #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, Default)]
852 }
853 };
854
855 Ok(quote! {
856 #doc_comment
857 #derives
858 pub enum #enum_name {
859 #(#variants)*
860 }
861 })
862 }
863
864 fn generate_struct(
865 &self,
866 schema: &crate::analysis::AnalyzedSchema,
867 properties: &BTreeMap<String, crate::analysis::PropertyInfo>,
868 required: &std::collections::HashSet<String>,
869 additional_properties: bool,
870 analysis: &crate::analysis::SchemaAnalysis,
871 discriminator_info: Option<&DiscriminatedVariantInfo>,
872 ) -> Result<TokenStream> {
873 let struct_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
874
875 let mut sorted_properties: Vec<_> = properties.iter().collect();
877 sorted_properties.sort_by_key(|(name, _)| name.as_str());
878
879 let mut fields: Vec<TokenStream> = sorted_properties
880 .into_iter()
881 .filter(|(field_name, _)| {
882 if let Some(info) = discriminator_info {
886 if !info.is_parent_untagged
887 && field_name.as_str() == info.discriminator_field.as_str()
888 {
889 false } else {
891 true }
893 } else {
894 true }
896 })
897 .map(|(field_name, prop)| {
898 let field_ident = Self::to_field_ident(&self.to_rust_field_name(field_name));
899 let is_required = required.contains(field_name);
900 let field_type =
901 self.generate_field_type(&schema.name, field_name, prop, is_required, analysis);
902
903 let serde_attrs =
904 self.generate_serde_field_attrs(field_name, prop, is_required, analysis);
905 let specta_attrs = self.generate_specta_field_attrs(field_name);
906
907 let doc_comment = if let Some(desc) = &prop.description {
908 let sanitized_desc = self.sanitize_doc_comment(desc);
909 quote! { #[doc = #sanitized_desc] }
910 } else {
911 TokenStream::new()
912 };
913
914 quote! {
915 #doc_comment
916 #serde_attrs
917 #specta_attrs
918 pub #field_ident: #field_type,
919 }
920 })
921 .collect();
922
923 if additional_properties {
925 fields.push(quote! {
926 #[serde(flatten)]
928 pub additional_properties: std::collections::BTreeMap<String, serde_json::Value>,
929 });
930 }
931
932 let doc_comment = if let Some(desc) = &schema.description {
933 quote! { #[doc = #desc] }
934 } else {
935 TokenStream::new()
936 };
937
938 let derives = if self.config.enable_specta {
942 quote! {
943 #[derive(Debug, Clone, Deserialize, Serialize)]
944 #[cfg_attr(feature = "specta", derive(specta::Type))]
945 }
946 } else {
947 quote! {
948 #[derive(Debug, Clone, Deserialize, Serialize)]
949 }
950 };
951
952 Ok(quote! {
953 #doc_comment
954 #derives
955 pub struct #struct_name {
956 #(#fields)*
957 }
958 })
959 }
960
961 fn generate_discriminated_enum(
962 &self,
963 schema: &crate::analysis::AnalyzedSchema,
964 discriminator_field: &str,
965 variants: &[crate::analysis::UnionVariant],
966 analysis: &crate::analysis::SchemaAnalysis,
967 ) -> Result<TokenStream> {
968 let enum_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
969
970 let has_nested_discriminated_union = variants.iter().any(|variant| {
972 if let Some(variant_schema) = analysis.schemas.get(&variant.type_name) {
973 matches!(
974 variant_schema.schema_type,
975 crate::analysis::SchemaType::DiscriminatedUnion { .. }
976 )
977 } else {
978 false
979 }
980 });
981
982 if has_nested_discriminated_union {
984 let schema_refs: Vec<crate::analysis::SchemaRef> = variants
986 .iter()
987 .map(|v| crate::analysis::SchemaRef {
988 target: v.type_name.clone(),
989 nullable: false,
990 })
991 .collect();
992 return self.generate_union_enum(schema, &schema_refs);
993 }
994
995 let enum_variants = variants.iter().map(|variant| {
996 let variant_name = format_ident!("{}", variant.rust_name);
997 let variant_value = &variant.discriminator_value;
998
999 let variant_type = format_ident!("{}", self.to_rust_type_name(&variant.type_name));
1002 quote! {
1003 #[serde(rename = #variant_value)]
1004 #variant_name(#variant_type),
1005 }
1006 });
1007
1008 let doc_comment = if let Some(desc) = &schema.description {
1009 quote! { #[doc = #desc] }
1010 } else {
1011 TokenStream::new()
1012 };
1013
1014 let derives = if self.config.enable_specta {
1016 quote! {
1017 #[derive(Debug, Clone, Deserialize, Serialize)]
1018 #[cfg_attr(feature = "specta", derive(specta::Type))]
1019 #[serde(tag = #discriminator_field)]
1020 }
1021 } else {
1022 quote! {
1023 #[derive(Debug, Clone, Deserialize, Serialize)]
1024 #[serde(tag = #discriminator_field)]
1025 }
1026 };
1027
1028 Ok(quote! {
1029 #doc_comment
1030 #derives
1031 pub enum #enum_name {
1032 #(#enum_variants)*
1033 }
1034 })
1035 }
1036
1037 fn should_use_untagged_discriminated_union(
1039 &self,
1040 schema: &crate::analysis::AnalyzedSchema,
1041 analysis: &crate::analysis::SchemaAnalysis,
1042 ) -> bool {
1043 for other_schema in analysis.schemas.values() {
1048 if let crate::analysis::SchemaType::DiscriminatedUnion {
1049 variants,
1050 discriminator_field: _,
1051 } = &other_schema.schema_type
1052 {
1053 for variant in variants {
1054 if variant.type_name == schema.name {
1055 if let crate::analysis::SchemaType::DiscriminatedUnion {
1060 discriminator_field: current_discriminator,
1061 variants: current_variants,
1062 ..
1063 } = &schema.schema_type
1064 {
1065 for current_variant in current_variants {
1067 if let Some(variant_schema) =
1068 analysis.schemas.get(¤t_variant.type_name)
1069 {
1070 if let crate::analysis::SchemaType::Object {
1071 properties, ..
1072 } = &variant_schema.schema_type
1073 {
1074 if properties.contains_key(current_discriminator) {
1075 return false;
1078 }
1079 }
1080 }
1081 }
1082 }
1083
1084 return true;
1086 }
1087 }
1088 }
1089 }
1090 false
1091 }
1092
1093 fn generate_union_enum(
1094 &self,
1095 schema: &crate::analysis::AnalyzedSchema,
1096 variants: &[crate::analysis::SchemaRef],
1097 ) -> Result<TokenStream> {
1098 let enum_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
1099
1100 let mut used_variant_names = std::collections::HashSet::new();
1102 let enum_variants = variants.iter().enumerate().map(|(i, variant)| {
1103 let base_variant_name = self.type_name_to_variant_name(&variant.target);
1105 let variant_name = self.ensure_unique_variant_name_generator(
1106 base_variant_name,
1107 &mut used_variant_names,
1108 i,
1109 );
1110 let variant_name_ident = format_ident!("{}", variant_name);
1111
1112 let variant_type_tokens = if matches!(
1114 variant.target.as_str(),
1115 "bool"
1116 | "i8"
1117 | "i16"
1118 | "i32"
1119 | "i64"
1120 | "i128"
1121 | "u8"
1122 | "u16"
1123 | "u32"
1124 | "u64"
1125 | "u128"
1126 | "f32"
1127 | "f64"
1128 | "String"
1129 ) {
1130 let type_ident = format_ident!("{}", variant.target);
1131 quote! { #type_ident }
1132 } else if variant.target.starts_with("Vec<") && variant.target.ends_with(">") {
1133 let inner = &variant.target[4..variant.target.len() - 1];
1135
1136 if inner.starts_with("Vec<") && inner.ends_with(">") {
1138 let inner_inner = &inner[4..inner.len() - 1];
1139 let inner_inner_type = if matches!(
1140 inner_inner,
1141 "bool"
1142 | "i8"
1143 | "i16"
1144 | "i32"
1145 | "i64"
1146 | "i128"
1147 | "u8"
1148 | "u16"
1149 | "u32"
1150 | "u64"
1151 | "u128"
1152 | "f32"
1153 | "f64"
1154 | "String"
1155 ) {
1156 format_ident!("{}", inner_inner)
1157 } else {
1158 format_ident!("{}", self.to_rust_type_name(inner_inner))
1159 };
1160 quote! { Vec<Vec<#inner_inner_type>> }
1161 } else {
1162 let inner_type = if matches!(
1163 inner,
1164 "bool"
1165 | "i8"
1166 | "i16"
1167 | "i32"
1168 | "i64"
1169 | "i128"
1170 | "u8"
1171 | "u16"
1172 | "u32"
1173 | "u64"
1174 | "u128"
1175 | "f32"
1176 | "f64"
1177 | "String"
1178 ) {
1179 format_ident!("{}", inner)
1180 } else {
1181 format_ident!("{}", self.to_rust_type_name(inner))
1182 };
1183 quote! { Vec<#inner_type> }
1184 }
1185 } else {
1186 let type_ident = format_ident!("{}", self.to_rust_type_name(&variant.target));
1187 quote! { #type_ident }
1188 };
1189
1190 quote! {
1191 #variant_name_ident(#variant_type_tokens),
1192 }
1193 });
1194
1195 let doc_comment = if let Some(desc) = &schema.description {
1196 quote! { #[doc = #desc] }
1197 } else {
1198 TokenStream::new()
1199 };
1200
1201 let derives = if self.config.enable_specta {
1203 quote! {
1204 #[derive(Debug, Clone, Deserialize, Serialize)]
1205 #[cfg_attr(feature = "specta", derive(specta::Type))]
1206 #[serde(untagged)]
1207 }
1208 } else {
1209 quote! {
1210 #[derive(Debug, Clone, Deserialize, Serialize)]
1211 #[serde(untagged)]
1212 }
1213 };
1214
1215 Ok(quote! {
1216 #doc_comment
1217 #derives
1218 pub enum #enum_name {
1219 #(#enum_variants)*
1220 }
1221 })
1222 }
1223
1224 fn generate_field_type(
1225 &self,
1226 schema_name: &str,
1227 field_name: &str,
1228 prop: &crate::analysis::PropertyInfo,
1229 is_required: bool,
1230 analysis: &crate::analysis::SchemaAnalysis,
1231 ) -> TokenStream {
1232 use crate::analysis::SchemaType;
1233
1234 let base_type = match &prop.schema_type {
1235 SchemaType::Primitive { rust_type } => {
1236 if rust_type.contains("::") {
1238 let parts: Vec<&str> = rust_type.split("::").collect();
1239 if parts.len() == 2 {
1240 let module = format_ident!("{}", parts[0]);
1241 let type_name = format_ident!("{}", parts[1]);
1242 quote! { #module::#type_name }
1243 } else {
1244 let path_parts: Vec<_> =
1246 parts.iter().map(|p| format_ident!("{}", p)).collect();
1247 quote! { #(#path_parts)::* }
1248 }
1249 } else {
1250 let type_ident = format_ident!("{}", rust_type);
1251 quote! { #type_ident }
1252 }
1253 }
1254 SchemaType::Reference { target } => {
1255 let target_type = format_ident!("{}", self.to_rust_type_name(target));
1256 if analysis.dependencies.recursive_schemas.contains(target) {
1258 quote! { Box<#target_type> }
1259 } else {
1260 quote! { #target_type }
1261 }
1262 }
1263 SchemaType::Array { item_type } => {
1264 let inner_type = self.generate_array_item_type(item_type, analysis);
1265 quote! { Vec<#inner_type> }
1266 }
1267 _ => {
1268 quote! { serde_json::Value }
1270 }
1271 };
1272
1273 let override_key = format!("{schema_name}.{field_name}");
1275 let is_nullable_override = self
1276 .config
1277 .nullable_field_overrides
1278 .get(&override_key)
1279 .copied()
1280 .unwrap_or(false);
1281
1282 if is_required && !prop.nullable && !is_nullable_override {
1283 if prop.default.is_some() && self.type_lacks_default(&prop.schema_type, analysis) {
1286 quote! { Option<#base_type> }
1287 } else {
1288 base_type
1289 }
1290 } else {
1291 quote! { Option<#base_type> }
1292 }
1293 }
1294
1295 fn generate_serde_field_attrs(
1296 &self,
1297 field_name: &str,
1298 prop: &crate::analysis::PropertyInfo,
1299 is_required: bool,
1300 analysis: &crate::analysis::SchemaAnalysis,
1301 ) -> TokenStream {
1302 let mut attrs = Vec::new();
1303
1304 let rust_field_name = self.to_rust_field_name(field_name);
1307 let comparison_name = rust_field_name
1308 .strip_prefix("r#")
1309 .unwrap_or(&rust_field_name);
1310 if comparison_name != field_name {
1311 attrs.push(quote! { rename = #field_name });
1312 }
1313
1314 if !is_required || prop.nullable {
1316 attrs.push(quote! { skip_serializing_if = "Option::is_none" });
1317 }
1318
1319 if prop.default.is_some()
1323 && (is_required && !prop.nullable)
1324 && !self.type_lacks_default(&prop.schema_type, analysis)
1325 {
1326 attrs.push(quote! { default });
1327 }
1328
1329 if attrs.is_empty() {
1330 TokenStream::new()
1331 } else {
1332 quote! { #[serde(#(#attrs),*)] }
1333 }
1334 }
1335
1336 fn type_lacks_default(
1340 &self,
1341 schema_type: &crate::analysis::SchemaType,
1342 analysis: &crate::analysis::SchemaAnalysis,
1343 ) -> bool {
1344 use crate::analysis::SchemaType;
1345 match schema_type {
1346 SchemaType::DiscriminatedUnion { .. } | SchemaType::Union { .. } => true,
1347 SchemaType::Reference { target } => {
1348 if let Some(schema) = analysis.schemas.get(target) {
1349 self.type_lacks_default(&schema.schema_type, analysis)
1350 } else {
1351 false
1352 }
1353 }
1354 _ => false,
1355 }
1356 }
1357
1358 fn generate_specta_field_attrs(&self, field_name: &str) -> TokenStream {
1359 if !self.config.enable_specta {
1360 return TokenStream::new();
1361 }
1362
1363 let camel_case_name = self.to_camel_case(field_name);
1365
1366 if camel_case_name != field_name {
1368 quote! { #[cfg_attr(feature = "specta", specta(rename = #camel_case_name))] }
1369 } else {
1370 TokenStream::new()
1371 }
1372 }
1373
1374 fn to_rust_enum_variant(&self, s: &str) -> String {
1375 let mut result = String::new();
1377 let mut next_upper = true;
1378 let mut prev_was_upper = false;
1379
1380 for (i, c) in s.chars().enumerate() {
1381 match c {
1382 'a'..='z' => {
1383 if next_upper {
1384 result.push(c.to_ascii_uppercase());
1385 next_upper = false;
1386 } else {
1387 result.push(c);
1388 }
1389 prev_was_upper = false;
1390 }
1391 'A'..='Z' => {
1392 if next_upper || (!prev_was_upper && i > 0) {
1393 result.push(c);
1395 next_upper = false;
1396 } else {
1397 result.push(c.to_ascii_lowercase());
1399 }
1400 prev_was_upper = true;
1401 }
1402 '0'..='9' => {
1403 result.push(c);
1404 next_upper = false;
1405 prev_was_upper = false;
1406 }
1407 '.' | '-' | '_' | ' ' | '@' | '#' | '$' | '/' | '\\' => {
1408 next_upper = true;
1410 prev_was_upper = false;
1411 }
1412 _ => {
1413 next_upper = true;
1415 prev_was_upper = false;
1416 }
1417 }
1418 }
1419
1420 if result.is_empty() {
1422 result = "Value".to_string();
1423 }
1424
1425 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1427 result = format!("Variant{result}");
1428 }
1429
1430 match result.as_str() {
1432 "Null" => "NullValue".to_string(),
1433 "True" => "TrueValue".to_string(),
1434 "False" => "FalseValue".to_string(),
1435 "Type" => "Type_".to_string(),
1436 "Match" => "Match_".to_string(),
1437 "Fn" => "Fn_".to_string(),
1438 "Impl" => "Impl_".to_string(),
1439 "Trait" => "Trait_".to_string(),
1440 "Struct" => "Struct_".to_string(),
1441 "Enum" => "Enum_".to_string(),
1442 "Mod" => "Mod_".to_string(),
1443 "Use" => "Use_".to_string(),
1444 "Pub" => "Pub_".to_string(),
1445 "Const" => "Const_".to_string(),
1446 "Static" => "Static_".to_string(),
1447 "Let" => "Let_".to_string(),
1448 "Mut" => "Mut_".to_string(),
1449 "Ref" => "Ref_".to_string(),
1450 "Move" => "Move_".to_string(),
1451 "Return" => "Return_".to_string(),
1452 "If" => "If_".to_string(),
1453 "Else" => "Else_".to_string(),
1454 "While" => "While_".to_string(),
1455 "For" => "For_".to_string(),
1456 "Loop" => "Loop_".to_string(),
1457 "Break" => "Break_".to_string(),
1458 "Continue" => "Continue_".to_string(),
1459 "Self" => "Self_".to_string(),
1460 "Super" => "Super_".to_string(),
1461 "Crate" => "Crate_".to_string(),
1462 "Async" => "Async_".to_string(),
1463 "Await" => "Await_".to_string(),
1464 _ => result,
1465 }
1466 }
1467
1468 #[allow(dead_code)]
1469 fn to_rust_identifier(&self, s: &str) -> String {
1470 let mut result = s
1472 .chars()
1473 .map(|c| match c {
1474 'a'..='z' | 'A'..='Z' | '0'..='9' => c,
1475 '.' | '-' | '_' | ' ' | '@' | '#' | '$' | '/' | '\\' => '_',
1476 _ => '_',
1477 })
1478 .collect::<String>();
1479
1480 result = result.trim_matches('_').to_string();
1482
1483 if result.is_empty() {
1485 result = "value".to_string();
1486 }
1487
1488 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1490 result = format!("variant_{result}");
1491 }
1492
1493 match result.as_str() {
1495 "null" => "null_value".to_string(),
1496 "true" => "true_value".to_string(),
1497 "false" => "false_value".to_string(),
1498 "type" => "type_".to_string(),
1499 "match" => "match_".to_string(),
1500 "fn" => "fn_".to_string(),
1501 "impl" => "impl_".to_string(),
1502 "trait" => "trait_".to_string(),
1503 "struct" => "struct_".to_string(),
1504 "enum" => "enum_".to_string(),
1505 "mod" => "mod_".to_string(),
1506 "use" => "use_".to_string(),
1507 "pub" => "pub_".to_string(),
1508 "const" => "const_".to_string(),
1509 "static" => "static_".to_string(),
1510 "let" => "let_".to_string(),
1511 "mut" => "mut_".to_string(),
1512 "ref" => "ref_".to_string(),
1513 "move" => "move_".to_string(),
1514 "return" => "return_".to_string(),
1515 "if" => "if_".to_string(),
1516 "else" => "else_".to_string(),
1517 "while" => "while_".to_string(),
1518 "for" => "for_".to_string(),
1519 "loop" => "loop_".to_string(),
1520 "break" => "break_".to_string(),
1521 "continue" => "continue_".to_string(),
1522 "self" => "self_".to_string(),
1523 "super" => "super_".to_string(),
1524 "crate" => "crate_".to_string(),
1525 "async" => "async_".to_string(),
1526 "await" => "await_".to_string(),
1527 "override" => "override_".to_string(),
1529 "box" => "box_".to_string(),
1530 "dyn" => "dyn_".to_string(),
1531 "where" => "where_".to_string(),
1532 "in" => "in_".to_string(),
1533 "abstract" => "abstract_".to_string(),
1535 "become" => "become_".to_string(),
1536 "do" => "do_".to_string(),
1537 "final" => "final_".to_string(),
1538 "macro" => "macro_".to_string(),
1539 "priv" => "priv_".to_string(),
1540 "try" => "try_".to_string(),
1541 "typeof" => "typeof_".to_string(),
1542 "unsized" => "unsized_".to_string(),
1543 "virtual" => "virtual_".to_string(),
1544 "yield" => "yield_".to_string(),
1545 _ => result,
1546 }
1547 }
1548
1549 fn sanitize_doc_comment(&self, desc: &str) -> String {
1550 let mut result = desc.to_string();
1552
1553 if result.contains('\n')
1561 && (result.contains('{')
1562 || result.contains("```")
1563 || result.contains("Human:")
1564 || result.contains("Assistant:")
1565 || result
1566 .lines()
1567 .any(|line| line.trim().starts_with('"') && line.trim().ends_with('"')))
1568 {
1569 if result.contains("```") {
1571 result = result.replace("```", "```ignore");
1572 } else {
1573 if result.lines().any(|line| {
1575 let trimmed = line.trim();
1576 trimmed.starts_with('"') && trimmed.ends_with('"') && trimmed.len() > 2
1577 }) {
1578 result = format!("```ignore\n{result}\n```");
1579 }
1580 }
1581 }
1582
1583 result
1584 }
1585
1586 pub(crate) fn to_rust_type_name(&self, s: &str) -> String {
1587 let mut result = String::new();
1589 let mut next_upper = true;
1590 let mut prev_was_lower = false;
1591
1592 for c in s.chars() {
1593 match c {
1594 'a'..='z' => {
1595 if next_upper {
1596 result.push(c.to_ascii_uppercase());
1597 next_upper = false;
1598 } else {
1599 result.push(c);
1600 }
1601 prev_was_lower = true;
1602 }
1603 'A'..='Z' => {
1604 result.push(c);
1605 next_upper = false;
1606 prev_was_lower = false;
1607 }
1608 '0'..='9' => {
1609 if prev_was_lower && !result.chars().last().unwrap_or(' ').is_ascii_digit() {
1612 }
1614 result.push(c);
1615 next_upper = false;
1616 prev_was_lower = false;
1617 }
1618 '_' | '-' | '.' | ' ' => {
1619 next_upper = true;
1621 prev_was_lower = false;
1622 }
1623 _ => {
1624 next_upper = true;
1626 prev_was_lower = false;
1627 }
1628 }
1629 }
1630
1631 if result.is_empty() {
1633 result = "Type".to_string();
1634 }
1635
1636 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1638 result = format!("Type{result}");
1639 }
1640
1641 result
1642 }
1643
1644 fn to_rust_field_name(&self, s: &str) -> String {
1645 let mut result = String::new();
1647 let mut prev_was_upper = false;
1648 let mut prev_was_underscore = false;
1649
1650 for (i, c) in s.chars().enumerate() {
1651 match c {
1652 'A'..='Z' => {
1653 if i > 0 && !prev_was_upper && !prev_was_underscore {
1655 result.push('_');
1656 }
1657 result.push(c.to_ascii_lowercase());
1658 prev_was_upper = true;
1659 prev_was_underscore = false;
1660 }
1661 'a'..='z' | '0'..='9' => {
1662 result.push(c);
1663 prev_was_upper = false;
1664 prev_was_underscore = false;
1665 }
1666 '-' | '.' | '_' | '@' | '#' | '$' | ' ' => {
1667 if !prev_was_underscore && !result.is_empty() {
1668 result.push('_');
1669 prev_was_underscore = true;
1670 }
1671 prev_was_upper = false;
1672 }
1673 _ => {
1674 if !prev_was_underscore && !result.is_empty() {
1676 result.push('_');
1677 }
1678 prev_was_upper = false;
1679 prev_was_underscore = true;
1680 }
1681 }
1682 }
1683
1684 let mut result = result.trim_matches('_').to_string();
1686 if result.is_empty() {
1687 return "field".to_string();
1688 }
1689
1690 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1692 result = format!("field_{result}");
1693 }
1694
1695 if Self::is_rust_keyword(&result) {
1697 format!("r#{result}")
1698 } else {
1699 result
1700 }
1701 }
1702
1703 pub fn is_rust_keyword(s: &str) -> bool {
1705 matches!(
1706 s,
1707 "type"
1708 | "match"
1709 | "fn"
1710 | "struct"
1711 | "enum"
1712 | "impl"
1713 | "trait"
1714 | "mod"
1715 | "use"
1716 | "pub"
1717 | "const"
1718 | "static"
1719 | "let"
1720 | "mut"
1721 | "ref"
1722 | "move"
1723 | "return"
1724 | "if"
1725 | "else"
1726 | "while"
1727 | "for"
1728 | "loop"
1729 | "break"
1730 | "continue"
1731 | "self"
1732 | "super"
1733 | "crate"
1734 | "async"
1735 | "await"
1736 | "override"
1737 | "box"
1738 | "dyn"
1739 | "where"
1740 | "in"
1741 | "abstract"
1742 | "become"
1743 | "do"
1744 | "final"
1745 | "macro"
1746 | "priv"
1747 | "try"
1748 | "typeof"
1749 | "unsized"
1750 | "virtual"
1751 | "yield"
1752 )
1753 }
1754
1755 pub fn to_field_ident(name: &str) -> proc_macro2::Ident {
1757 if let Some(raw) = name.strip_prefix("r#") {
1758 proc_macro2::Ident::new_raw(raw, proc_macro2::Span::call_site())
1759 } else {
1760 proc_macro2::Ident::new(name, proc_macro2::Span::call_site())
1761 }
1762 }
1763
1764 fn to_camel_case(&self, s: &str) -> String {
1765 let mut result = String::new();
1767 let mut capitalize_next = false;
1768
1769 for (i, c) in s.chars().enumerate() {
1770 match c {
1771 '_' | '-' | '.' | ' ' => {
1772 capitalize_next = true;
1774 }
1775 'A'..='Z' => {
1776 if i == 0 {
1777 result.push(c.to_ascii_lowercase());
1779 } else if capitalize_next {
1780 result.push(c);
1781 capitalize_next = false;
1782 } else {
1783 result.push(c.to_ascii_lowercase());
1784 }
1785 }
1786 'a'..='z' | '0'..='9' => {
1787 if capitalize_next {
1788 result.push(c.to_ascii_uppercase());
1789 capitalize_next = false;
1790 } else {
1791 result.push(c);
1792 }
1793 }
1794 _ => {
1795 capitalize_next = true;
1797 }
1798 }
1799 }
1800
1801 if result.is_empty() {
1802 return "field".to_string();
1803 }
1804
1805 result
1806 }
1807
1808 fn generate_composition_struct(
1809 &self,
1810 schema: &crate::analysis::AnalyzedSchema,
1811 schemas: &[crate::analysis::SchemaRef],
1812 ) -> Result<TokenStream> {
1813 let struct_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
1814
1815 let fields = schemas.iter().enumerate().map(|(i, schema_ref)| {
1821 let field_name = format_ident!("part_{}", i);
1822 let field_type = format_ident!("{}", self.to_rust_type_name(&schema_ref.target));
1823
1824 quote! {
1825 #[serde(flatten)]
1826 pub #field_name: #field_type,
1827 }
1828 });
1829
1830 let doc_comment = if let Some(desc) = &schema.description {
1831 quote! { #[doc = #desc] }
1832 } else {
1833 TokenStream::new()
1834 };
1835
1836 let derives = if self.config.enable_specta {
1838 quote! {
1839 #[derive(Debug, Clone, Deserialize, Serialize)]
1840 #[cfg_attr(feature = "specta", derive(specta::Type))]
1841 }
1842 } else {
1843 quote! {
1844 #[derive(Debug, Clone, Deserialize, Serialize)]
1845 }
1846 };
1847
1848 Ok(quote! {
1849 #doc_comment
1850 #derives
1851 pub struct #struct_name {
1852 #(#fields)*
1853 }
1854 })
1855 }
1856
1857 #[allow(dead_code)]
1858 fn find_missing_types(&self, analysis: &SchemaAnalysis) -> std::collections::HashSet<String> {
1859 let mut missing = std::collections::HashSet::new();
1860 let defined_types: std::collections::HashSet<String> =
1861 analysis.schemas.keys().cloned().collect();
1862
1863 for schema in analysis.schemas.values() {
1865 match &schema.schema_type {
1866 crate::analysis::SchemaType::Union { variants } => {
1867 for variant in variants {
1868 if !defined_types.contains(&variant.target) {
1869 missing.insert(variant.target.clone());
1870 }
1871 }
1872 }
1873 crate::analysis::SchemaType::DiscriminatedUnion { variants, .. } => {
1874 for variant in variants {
1875 if !defined_types.contains(&variant.type_name) {
1876 missing.insert(variant.type_name.clone());
1877 }
1878 }
1879 }
1880 crate::analysis::SchemaType::Object { properties, .. } => {
1881 let mut sorted_props: Vec<_> = properties.iter().collect();
1883 sorted_props.sort_by_key(|(name, _)| name.as_str());
1884 for (_, prop) in sorted_props {
1885 if let crate::analysis::SchemaType::Reference { target } = &prop.schema_type
1886 {
1887 if !defined_types.contains(target) {
1888 missing.insert(target.clone());
1889 }
1890 }
1891 }
1892 }
1893 crate::analysis::SchemaType::Reference { target }
1894 if !defined_types.contains(target) =>
1895 {
1896 missing.insert(target.clone());
1897 }
1898 _ => {}
1899 }
1900 }
1901
1902 missing
1903 }
1904
1905 #[allow(clippy::only_used_in_recursion)]
1906 fn generate_array_item_type(
1907 &self,
1908 item_type: &crate::analysis::SchemaType,
1909 analysis: &crate::analysis::SchemaAnalysis,
1910 ) -> TokenStream {
1911 use crate::analysis::SchemaType;
1912
1913 match item_type {
1914 SchemaType::Primitive { rust_type } => {
1915 if rust_type.contains("::") {
1917 let parts: Vec<&str> = rust_type.split("::").collect();
1918 if parts.len() == 2 {
1919 let module = format_ident!("{}", parts[0]);
1920 let type_name = format_ident!("{}", parts[1]);
1921 quote! { #module::#type_name }
1922 } else {
1923 let path_parts: Vec<_> =
1925 parts.iter().map(|p| format_ident!("{}", p)).collect();
1926 quote! { #(#path_parts)::* }
1927 }
1928 } else {
1929 let type_ident = format_ident!("{}", rust_type);
1930 quote! { #type_ident }
1931 }
1932 }
1933 SchemaType::Reference { target } => {
1934 let target_type = format_ident!("{}", self.to_rust_type_name(target));
1935 if analysis.dependencies.recursive_schemas.contains(target) {
1937 quote! { Box<#target_type> }
1938 } else {
1939 quote! { #target_type }
1940 }
1941 }
1942 SchemaType::Array { item_type } => {
1943 let inner_type = self.generate_array_item_type(item_type, analysis);
1945 quote! { Vec<#inner_type> }
1946 }
1947 _ => {
1948 quote! { serde_json::Value }
1950 }
1951 }
1952 }
1953
1954 fn type_name_to_variant_name(&self, type_name: &str) -> String {
1956 match type_name {
1958 "bool" => return "Boolean".to_string(),
1959 "i8" | "i16" | "i32" | "i64" | "i128" => return "Integer".to_string(),
1960 "u8" | "u16" | "u32" | "u64" | "u128" => return "UnsignedInteger".to_string(),
1961 "f32" | "f64" => return "Number".to_string(),
1962 "String" => return "String".to_string(),
1963 _ => {}
1964 }
1965
1966 if type_name.starts_with("Vec<") && type_name.ends_with(">") {
1968 let inner = &type_name[4..type_name.len() - 1];
1969 if inner.starts_with("Vec<") && inner.ends_with(">") {
1971 let inner_inner = &inner[4..inner.len() - 1];
1972 return format!("{}ArrayArray", self.type_name_to_variant_name(inner_inner));
1973 }
1974 return format!("{}Array", self.type_name_to_variant_name(inner));
1975 }
1976
1977 let clean_name = type_name
1983 .trim_end_matches("Type")
1984 .trim_end_matches("Schema")
1985 .trim_end_matches("Item");
1986
1987 self.to_rust_type_name(clean_name)
1989 }
1990
1991 fn ensure_unique_variant_name_generator(
1993 &self,
1994 base_name: String,
1995 used_names: &mut std::collections::HashSet<String>,
1996 fallback_index: usize,
1997 ) -> String {
1998 if used_names.insert(base_name.clone()) {
1999 return base_name;
2000 }
2001
2002 for i in 2..100 {
2004 let numbered_name = format!("{base_name}{i}");
2005 if used_names.insert(numbered_name.clone()) {
2006 return numbered_name;
2007 }
2008 }
2009
2010 let fallback = format!("Variant{fallback_index}");
2012 used_names.insert(fallback.clone());
2013 fallback
2014 }
2015
2016 fn find_request_type_for_operation(
2018 &self,
2019 operation_id: &str,
2020 analysis: &SchemaAnalysis,
2021 ) -> Option<String> {
2022 analysis.operations.get(operation_id).and_then(|op| {
2024 op.request_body
2025 .as_ref()
2026 .and_then(|rb| rb.schema_name().map(|s| s.to_string()))
2027 })
2028 }
2029
2030 fn resolve_streaming_event_type(
2032 &self,
2033 endpoint: &crate::streaming::StreamingEndpoint,
2034 analysis: &SchemaAnalysis,
2035 ) -> Result<String> {
2036 match &endpoint.event_flow {
2037 crate::streaming::EventFlow::Simple => {
2038 if analysis.schemas.contains_key(&endpoint.event_union_type) {
2041 Ok(endpoint.event_union_type.to_string())
2042 } else {
2043 Err(crate::error::GeneratorError::ValidationError(format!(
2044 "Streaming response type '{}' not found in schema for simple streaming endpoint '{}'",
2045 endpoint.event_union_type, endpoint.operation_id
2046 )))
2047 }
2048 }
2049 crate::streaming::EventFlow::StartDeltaStop { .. } => {
2050 if analysis.schemas.contains_key(&endpoint.event_union_type) {
2053 Ok(endpoint.event_union_type.to_string())
2054 } else {
2055 Err(crate::error::GeneratorError::ValidationError(format!(
2056 "Event union type '{}' not found in schema for complex streaming endpoint '{}'",
2057 endpoint.event_union_type, endpoint.operation_id
2058 )))
2059 }
2060 }
2061 }
2062 }
2063
2064 fn generate_streaming_error_types(&self) -> Result<TokenStream> {
2066 Ok(quote! {
2067 #[derive(Debug, thiserror::Error)]
2069 pub enum StreamingError {
2070 #[error("Connection error: {0}")]
2071 Connection(String),
2072 #[error("HTTP error: {status}")]
2073 Http { status: u16 },
2074 #[error("SSE parsing error: {0}")]
2075 Parsing(String),
2076 #[error("Authentication error: {0}")]
2077 Authentication(String),
2078 #[error("Rate limit error: {0}")]
2079 RateLimit(String),
2080 #[error("API error: {0}")]
2081 Api(String),
2082 #[error("Timeout error: {0}")]
2083 Timeout(String),
2084 #[error("JSON serialization/deserialization error: {0}")]
2085 Json(#[from] serde_json::Error),
2086 #[error("Request error: {0}")]
2087 Request(reqwest::Error),
2088 }
2089
2090 impl From<reqwest::header::InvalidHeaderValue> for StreamingError {
2091 fn from(err: reqwest::header::InvalidHeaderValue) -> Self {
2092 StreamingError::Api(format!("Invalid header value: {}", err))
2093 }
2094 }
2095
2096 impl From<reqwest::Error> for StreamingError {
2097 fn from(err: reqwest::Error) -> Self {
2098 if err.is_timeout() {
2099 StreamingError::Timeout(err.to_string())
2100 } else if err.is_status() {
2101 if let Some(status) = err.status() {
2102 StreamingError::Http { status: status.as_u16() }
2103 } else {
2104 StreamingError::Connection(err.to_string())
2105 }
2106 } else {
2107 StreamingError::Request(err)
2108 }
2109 }
2110 }
2111 })
2112 }
2113
2114 fn generate_endpoint_trait(
2116 &self,
2117 endpoint: &crate::streaming::StreamingEndpoint,
2118 analysis: &SchemaAnalysis,
2119 ) -> Result<TokenStream> {
2120 use crate::streaming::HttpMethod;
2121
2122 let trait_name = format_ident!(
2123 "{}StreamingClient",
2124 self.to_rust_type_name(&endpoint.operation_id)
2125 );
2126 let method_name =
2127 format_ident!("stream_{}", self.to_rust_field_name(&endpoint.operation_id));
2128 let event_type =
2129 format_ident!("{}", self.resolve_streaming_event_type(endpoint, analysis)?);
2130
2131 let method_signature = match endpoint.http_method {
2133 HttpMethod::Get => {
2134 let mut param_defs = Vec::new();
2136 for qp in &endpoint.query_parameters {
2137 let param_name = format_ident!("{}", self.to_rust_field_name(&qp.name));
2138 if qp.required {
2139 param_defs.push(quote! { #param_name: &str });
2140 } else {
2141 param_defs.push(quote! { #param_name: Option<&str> });
2142 }
2143 }
2144 quote! {
2145 async fn #method_name(
2146 &self,
2147 #(#param_defs),*
2148 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error>;
2149 }
2150 }
2151 HttpMethod::Post => {
2152 let request_type = self
2154 .find_request_type_for_operation(&endpoint.operation_id, analysis)
2155 .unwrap_or_else(|| "serde_json::Value".to_string());
2156 let request_type_ident = if request_type.contains("::") {
2157 let parts: Vec<&str> = request_type.split("::").collect();
2158 let path_parts: Vec<_> = parts.iter().map(|p| format_ident!("{}", p)).collect();
2159 quote! { #(#path_parts)::* }
2160 } else {
2161 let ident = format_ident!("{}", request_type);
2162 quote! { #ident }
2163 };
2164 quote! {
2165 async fn #method_name(
2166 &self,
2167 request: #request_type_ident,
2168 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error>;
2169 }
2170 }
2171 };
2172
2173 Ok(quote! {
2174 #[async_trait]
2176 pub trait #trait_name {
2177 type Error: std::error::Error + Send + Sync + 'static;
2178
2179 #method_signature
2181 }
2182 })
2183 }
2184
2185 fn generate_streaming_client_impl(
2187 &self,
2188 streaming_config: &crate::streaming::StreamingConfig,
2189 analysis: &SchemaAnalysis,
2190 ) -> Result<TokenStream> {
2191 let client_name = format_ident!(
2192 "{}Client",
2193 self.to_rust_type_name(&streaming_config.client_module_name)
2194 );
2195
2196 let mut struct_fields = vec![
2199 quote! { base_url: String },
2200 quote! { api_key: Option<String> },
2201 quote! { http_client: reqwest::Client },
2202 quote! { custom_headers: std::collections::BTreeMap<String, String> },
2203 ];
2204
2205 let has_optional_headers = !streaming_config
2206 .endpoints
2207 .iter()
2208 .all(|e| e.optional_headers.is_empty());
2209
2210 if has_optional_headers {
2211 struct_fields
2212 .push(quote! { optional_headers: std::collections::BTreeMap<String, String> });
2213 }
2214
2215 let default_base_url = if let Some(ref streaming_config) = self.config.streaming_config {
2218 streaming_config
2219 .endpoints
2220 .first()
2221 .and_then(|e| e.base_url.as_deref())
2222 .unwrap_or("https://api.example.com")
2223 } else {
2224 "https://api.example.com"
2225 };
2226
2227 let constructor_fields = if has_optional_headers {
2229 quote! {
2230 base_url: #default_base_url.to_string(),
2231 api_key: None,
2232 http_client: reqwest::Client::new(),
2233 custom_headers: std::collections::BTreeMap::new(),
2234 optional_headers: std::collections::BTreeMap::new(),
2235 }
2236 } else {
2237 quote! {
2238 base_url: #default_base_url.to_string(),
2239 api_key: None,
2240 http_client: reqwest::Client::new(),
2241 custom_headers: std::collections::BTreeMap::new(),
2242 }
2243 };
2244
2245 let optional_headers_method = if has_optional_headers {
2247 quote! {
2248 pub fn set_optional_headers(&mut self, headers: std::collections::BTreeMap<String, String>) {
2250 self.optional_headers = headers;
2251 }
2252 }
2253 } else {
2254 TokenStream::new()
2255 };
2256
2257 let constructor = quote! {
2258 impl #client_name {
2259 pub fn new() -> Self {
2261 Self {
2262 #constructor_fields
2263 }
2264 }
2265
2266 pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
2268 self.base_url = base_url.into();
2269 self
2270 }
2271
2272 pub fn with_api_key(mut self, api_key: impl Into<String>) -> Self {
2274 self.api_key = Some(api_key.into());
2275 self
2276 }
2277
2278 pub fn with_header(
2280 mut self,
2281 name: impl Into<String>,
2282 value: impl Into<String>,
2283 ) -> Self {
2284 self.custom_headers.insert(name.into(), value.into());
2285 self
2286 }
2287
2288 pub fn with_http_client(mut self, client: reqwest::Client) -> Self {
2290 self.http_client = client;
2291 self
2292 }
2293
2294 #optional_headers_method
2295 }
2296 };
2297
2298 let mut trait_impls = Vec::new();
2300 for endpoint in &streaming_config.endpoints {
2301 let trait_impl = self.generate_endpoint_trait_impl(endpoint, &client_name, analysis)?;
2302 trait_impls.push(trait_impl);
2303 }
2304
2305 let default_impl = quote! {
2307 impl Default for #client_name {
2308 fn default() -> Self {
2309 Self::new()
2310 }
2311 }
2312 };
2313
2314 Ok(quote! {
2315 #[derive(Debug, Clone)]
2317 pub struct #client_name {
2318 #(#struct_fields,)*
2319 }
2320
2321 #constructor
2322
2323 #default_impl
2324
2325 #(#trait_impls)*
2326 })
2327 }
2328
2329 fn generate_endpoint_trait_impl(
2331 &self,
2332 endpoint: &crate::streaming::StreamingEndpoint,
2333 client_name: &proc_macro2::Ident,
2334 analysis: &SchemaAnalysis,
2335 ) -> Result<TokenStream> {
2336 use crate::streaming::HttpMethod;
2337
2338 let trait_name = format_ident!(
2339 "{}StreamingClient",
2340 self.to_rust_type_name(&endpoint.operation_id)
2341 );
2342 let method_name =
2343 format_ident!("stream_{}", self.to_rust_field_name(&endpoint.operation_id));
2344 let event_type =
2345 format_ident!("{}", self.resolve_streaming_event_type(endpoint, analysis)?);
2346
2347 let mut header_setup = Vec::new();
2349 for (name, value) in &endpoint.required_headers {
2350 header_setup.push(quote! {
2351 headers.insert(#name, HeaderValue::from_static(#value));
2352 });
2353 }
2354
2355 if let Some(auth_header) = &endpoint.auth_header {
2358 match auth_header {
2359 crate::streaming::AuthHeader::Bearer(header_name) => {
2360 header_setup.push(quote! {
2361 if let Some(ref api_key) = self.api_key {
2362 headers.insert(#header_name, HeaderValue::from_str(&format!("Bearer {}", api_key))?);
2363 }
2364 });
2365 }
2366 crate::streaming::AuthHeader::ApiKey(header_name) => {
2367 header_setup.push(quote! {
2368 if let Some(ref api_key) = self.api_key {
2369 headers.insert(#header_name, HeaderValue::from_str(api_key)?);
2370 }
2371 });
2372 }
2373 }
2374 } else {
2375 header_setup.push(quote! {
2377 if let Some(ref api_key) = self.api_key {
2378 headers.insert("Authorization", HeaderValue::from_str(&format!("Bearer {}", api_key))?);
2379 }
2380 });
2381 }
2382
2383 header_setup.push(quote! {
2385 for (name, value) in &self.custom_headers {
2386 if let (Ok(header_name), Ok(header_value)) = (reqwest::header::HeaderName::from_bytes(name.as_bytes()), HeaderValue::from_str(value)) {
2387 headers.insert(header_name, header_value);
2388 }
2389 }
2390 });
2391
2392 if !endpoint.optional_headers.is_empty() {
2394 header_setup.push(quote! {
2395 for (key, value) in &self.optional_headers {
2396 if let (Ok(header_name), Ok(header_value)) = (reqwest::header::HeaderName::from_bytes(key.as_bytes()), HeaderValue::from_str(value)) {
2397 headers.insert(header_name, header_value);
2398 }
2399 }
2400 });
2401 }
2402
2403 match endpoint.http_method {
2405 HttpMethod::Get => self.generate_get_streaming_impl(
2406 endpoint,
2407 client_name,
2408 &trait_name,
2409 &method_name,
2410 &event_type,
2411 &header_setup,
2412 ),
2413 HttpMethod::Post => self.generate_post_streaming_impl(
2414 endpoint,
2415 client_name,
2416 &trait_name,
2417 &method_name,
2418 &event_type,
2419 &header_setup,
2420 analysis,
2421 ),
2422 }
2423 }
2424
2425 fn generate_get_streaming_impl(
2427 &self,
2428 endpoint: &crate::streaming::StreamingEndpoint,
2429 client_name: &proc_macro2::Ident,
2430 trait_name: &proc_macro2::Ident,
2431 method_name: &proc_macro2::Ident,
2432 event_type: &proc_macro2::Ident,
2433 header_setup: &[TokenStream],
2434 ) -> Result<TokenStream> {
2435 let path = &endpoint.path;
2436
2437 let mut param_defs = Vec::new();
2439 let mut query_params = Vec::new();
2440
2441 for qp in &endpoint.query_parameters {
2442 let param_name = format_ident!("{}", self.to_rust_field_name(&qp.name));
2443 let param_name_str = &qp.name;
2444
2445 if qp.required {
2446 param_defs.push(quote! { #param_name: &str });
2447 query_params.push(quote! {
2448 url.query_pairs_mut().append_pair(#param_name_str, #param_name);
2449 });
2450 } else {
2451 param_defs.push(quote! { #param_name: Option<&str> });
2452 query_params.push(quote! {
2453 if let Some(v) = #param_name {
2454 url.query_pairs_mut().append_pair(#param_name_str, v);
2455 }
2456 });
2457 }
2458 }
2459
2460 let url_construction = quote! {
2462 let base_url = url::Url::parse(&self.base_url)
2463 .map_err(|e| StreamingError::Connection(format!("Invalid base URL: {}", e)))?;
2464 let path_to_join = #path.trim_start_matches('/');
2465 let mut url = base_url.join(path_to_join)
2466 .map_err(|e| StreamingError::Connection(format!("URL join error: {}", e)))?;
2467 #(#query_params)*
2468 };
2469
2470 let instrument_skip = quote! { #[instrument(skip(self), name = "streaming_get_request")] };
2471
2472 Ok(quote! {
2473 #[async_trait]
2474 impl #trait_name for #client_name {
2475 type Error = StreamingError;
2476
2477 #instrument_skip
2478 async fn #method_name(
2479 &self,
2480 #(#param_defs),*
2481 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error> {
2482 debug!("Starting streaming GET request");
2483
2484 let mut headers = HeaderMap::new();
2485 #(#header_setup)*
2486
2487 #url_construction
2488 let url_str = url.to_string();
2489 debug!("Making streaming GET request to: {}", url_str);
2490
2491 let request_builder = self.http_client
2492 .get(url_str)
2493 .headers(headers);
2494
2495 debug!("Creating SSE stream from request");
2496 let stream = parse_sse_stream::<#event_type>(request_builder).await?;
2497 info!("SSE stream created successfully");
2498 Ok(Box::pin(stream))
2499 }
2500 }
2501 })
2502 }
2503
2504 #[allow(clippy::too_many_arguments)]
2506 fn generate_post_streaming_impl(
2507 &self,
2508 endpoint: &crate::streaming::StreamingEndpoint,
2509 client_name: &proc_macro2::Ident,
2510 trait_name: &proc_macro2::Ident,
2511 method_name: &proc_macro2::Ident,
2512 event_type: &proc_macro2::Ident,
2513 header_setup: &[TokenStream],
2514 analysis: &SchemaAnalysis,
2515 ) -> Result<TokenStream> {
2516 let path = &endpoint.path;
2517
2518 let request_type = self
2520 .find_request_type_for_operation(&endpoint.operation_id, analysis)
2521 .unwrap_or_else(|| "serde_json::Value".to_string());
2522 let request_type_ident = if request_type.contains("::") {
2523 let parts: Vec<&str> = request_type.split("::").collect();
2524 let path_parts: Vec<_> = parts.iter().map(|p| format_ident!("{}", p)).collect();
2525 quote! { #(#path_parts)::* }
2526 } else {
2527 let ident = format_ident!("{}", request_type);
2528 quote! { #ident }
2529 };
2530
2531 let url_construction = quote! {
2533 let base_url = url::Url::parse(&self.base_url)
2534 .map_err(|e| StreamingError::Connection(format!("Invalid base URL: {}", e)))?;
2535 let path_to_join = #path.trim_start_matches('/');
2536 let url = base_url.join(path_to_join)
2537 .map_err(|e| StreamingError::Connection(format!("URL join error: {}", e)))?
2538 .to_string();
2539 };
2540
2541 let stream_param = &endpoint.stream_parameter;
2543 let stream_setup = if stream_param.is_empty() {
2544 quote! {
2545 let streaming_request = request;
2546 }
2547 } else {
2548 quote! {
2549 let mut streaming_request = request;
2551 if let Ok(mut request_value) = serde_json::to_value(&streaming_request) {
2552 if let Some(obj) = request_value.as_object_mut() {
2553 obj.insert(#stream_param.to_string(), serde_json::Value::Bool(true));
2554 }
2555 streaming_request = serde_json::from_value(request_value)?;
2556 }
2557 }
2558 };
2559
2560 Ok(quote! {
2561 #[async_trait]
2562 impl #trait_name for #client_name {
2563 type Error = StreamingError;
2564
2565 #[instrument(skip(self, request), name = "streaming_post_request")]
2566 async fn #method_name(
2567 &self,
2568 request: #request_type_ident,
2569 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error> {
2570 debug!("Starting streaming POST request");
2571
2572 #stream_setup
2573
2574 let mut headers = HeaderMap::new();
2575 #(#header_setup)*
2576
2577 #url_construction
2578 debug!("Making streaming POST request to: {}", url);
2579
2580 let request_builder = self.http_client
2581 .post(&url)
2582 .headers(headers)
2583 .json(&streaming_request);
2584
2585 debug!("Creating SSE stream from request");
2586 let stream = parse_sse_stream::<#event_type>(request_builder).await?;
2587 info!("SSE stream created successfully");
2588 Ok(Box::pin(stream))
2589 }
2590 }
2591 })
2592 }
2593
2594 fn generate_sse_parser_utilities(
2596 &self,
2597 _streaming_config: &crate::streaming::StreamingConfig,
2598 ) -> Result<TokenStream> {
2599 Ok(quote! {
2600 pub async fn parse_sse_stream<T>(
2602 request_builder: reqwest::RequestBuilder
2603 ) -> Result<impl Stream<Item = Result<T, StreamingError>>, StreamingError>
2604 where
2605 T: serde::de::DeserializeOwned + Send + 'static,
2606 {
2607 let mut event_source = reqwest_eventsource::EventSource::new(request_builder).map_err(|e| {
2608 StreamingError::Connection(format!("Failed to create event source: {}", e))
2609 })?;
2610
2611 let stream = event_source.filter_map(|event_result| async move {
2612 match event_result {
2613 Ok(reqwest_eventsource::Event::Open) => {
2614 debug!("SSE connection opened");
2615 None
2616 }
2617 Ok(reqwest_eventsource::Event::Message(message)) => {
2618 if message.event == "ping" {
2620 debug!("Received SSE ping event, skipping");
2621 return None;
2622 }
2623
2624 if message.data.trim().is_empty() {
2626 debug!("Empty SSE data, skipping");
2627 return None;
2628 }
2629
2630 if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(&message.data) {
2632 if let Some(event_type) = json_value.get("event").and_then(|v| v.as_str()) {
2633 if event_type == "ping" {
2634 debug!("Received ping event in JSON data, skipping");
2635 return None;
2636 }
2637 }
2638
2639 match serde_json::from_value::<T>(json_value) {
2641 Ok(parsed_event) => {
2642 Some(Ok(parsed_event))
2643 }
2644 Err(e) => {
2645 if message.data.contains("ping") || message.event.contains("ping") {
2646 debug!("Ignoring ping-related event: {}", message.data);
2647 None
2648 } else {
2649 Some(Err(StreamingError::Parsing(
2650 format!("Failed to parse SSE event: {} (raw: {})", e, message.data)
2651 )))
2652 }
2653 }
2654 }
2655 } else {
2656 Some(Err(StreamingError::Parsing(
2658 format!("SSE event is not valid JSON: {}", message.data)
2659 )))
2660 }
2661 }
2662 Err(e) => {
2663 match e {
2665 reqwest_eventsource::Error::StreamEnded => {
2666 debug!("SSE stream completed normally");
2667 None }
2669 reqwest_eventsource::Error::InvalidStatusCode(status, response) => {
2670 let status_code = status.as_u16();
2672
2673 let error_body = match response.text().await {
2675 Ok(body) => body,
2676 Err(_) => "Failed to read error response body".to_string()
2677 };
2678
2679 error!("SSE connection error - HTTP {}: {}", status_code, error_body);
2680
2681 let detailed_error = format!(
2682 "HTTP {} error: {}",
2683 status_code,
2684 error_body
2685 );
2686
2687 Some(Err(StreamingError::Connection(detailed_error)))
2688 }
2689 _ => {
2690 let error_str = e.to_string();
2691 if error_str.contains("stream closed") {
2692 debug!("SSE stream closed");
2693 None
2694 } else {
2695 error!("SSE connection error: {}", e);
2696 Some(Err(StreamingError::Connection(error_str)))
2697 }
2698 }
2699 }
2700 }
2701 }
2702 });
2703
2704 Ok(stream)
2705 }
2706 })
2707 }
2708
2709 fn generate_reconnection_utilities(
2711 &self,
2712 reconnect_config: &crate::streaming::ReconnectionConfig,
2713 ) -> Result<TokenStream> {
2714 let max_retries = reconnect_config.max_retries;
2715 let initial_delay = reconnect_config.initial_delay_ms;
2716 let max_delay = reconnect_config.max_delay_ms;
2717 let backoff_multiplier = reconnect_config.backoff_multiplier;
2718
2719 Ok(quote! {
2720 #[derive(Debug, Clone)]
2722 pub struct ReconnectionManager {
2723 max_retries: u32,
2724 initial_delay_ms: u64,
2725 max_delay_ms: u64,
2726 backoff_multiplier: f64,
2727 current_attempt: u32,
2728 }
2729
2730 impl ReconnectionManager {
2731 pub fn new() -> Self {
2733 Self {
2734 max_retries: #max_retries,
2735 initial_delay_ms: #initial_delay,
2736 max_delay_ms: #max_delay,
2737 backoff_multiplier: #backoff_multiplier,
2738 current_attempt: 0,
2739 }
2740 }
2741
2742 pub fn should_retry(&self) -> bool {
2744 self.current_attempt < self.max_retries
2745 }
2746
2747 pub fn next_retry_delay(&mut self) -> Duration {
2749 if !self.should_retry() {
2750 return Duration::from_secs(0);
2751 }
2752
2753 let delay_ms = (self.initial_delay_ms as f64
2754 * self.backoff_multiplier.powi(self.current_attempt as i32)) as u64;
2755 let delay_ms = delay_ms.min(self.max_delay_ms);
2756
2757 self.current_attempt += 1;
2758 Duration::from_millis(delay_ms)
2759 }
2760
2761 pub fn reset(&mut self) {
2763 self.current_attempt = 0;
2764 }
2765
2766 pub fn current_attempt(&self) -> u32 {
2768 self.current_attempt
2769 }
2770 }
2771
2772 impl Default for ReconnectionManager {
2773 fn default() -> Self {
2774 Self::new()
2775 }
2776 }
2777 })
2778 }
2779}