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 == "serde_json::Value" {
1133 quote! { serde_json::Value }
1136 } else if variant.target.starts_with("Vec<") && variant.target.ends_with(">") {
1137 let inner = &variant.target[4..variant.target.len() - 1];
1139
1140 if inner.starts_with("Vec<") && inner.ends_with(">") {
1142 let inner_inner = &inner[4..inner.len() - 1];
1143 let inner_inner_type = if matches!(
1144 inner_inner,
1145 "bool"
1146 | "i8"
1147 | "i16"
1148 | "i32"
1149 | "i64"
1150 | "i128"
1151 | "u8"
1152 | "u16"
1153 | "u32"
1154 | "u64"
1155 | "u128"
1156 | "f32"
1157 | "f64"
1158 | "String"
1159 ) {
1160 format_ident!("{}", inner_inner)
1161 } else {
1162 format_ident!("{}", self.to_rust_type_name(inner_inner))
1163 };
1164 quote! { Vec<Vec<#inner_inner_type>> }
1165 } else {
1166 let inner_type = if matches!(
1167 inner,
1168 "bool"
1169 | "i8"
1170 | "i16"
1171 | "i32"
1172 | "i64"
1173 | "i128"
1174 | "u8"
1175 | "u16"
1176 | "u32"
1177 | "u64"
1178 | "u128"
1179 | "f32"
1180 | "f64"
1181 | "String"
1182 ) {
1183 format_ident!("{}", inner)
1184 } else {
1185 format_ident!("{}", self.to_rust_type_name(inner))
1186 };
1187 quote! { Vec<#inner_type> }
1188 }
1189 } else {
1190 let type_ident = format_ident!("{}", self.to_rust_type_name(&variant.target));
1191 quote! { #type_ident }
1192 };
1193
1194 quote! {
1195 #variant_name_ident(#variant_type_tokens),
1196 }
1197 });
1198
1199 let doc_comment = if let Some(desc) = &schema.description {
1200 quote! { #[doc = #desc] }
1201 } else {
1202 TokenStream::new()
1203 };
1204
1205 let derives = if self.config.enable_specta {
1207 quote! {
1208 #[derive(Debug, Clone, Deserialize, Serialize)]
1209 #[cfg_attr(feature = "specta", derive(specta::Type))]
1210 #[serde(untagged)]
1211 }
1212 } else {
1213 quote! {
1214 #[derive(Debug, Clone, Deserialize, Serialize)]
1215 #[serde(untagged)]
1216 }
1217 };
1218
1219 Ok(quote! {
1220 #doc_comment
1221 #derives
1222 pub enum #enum_name {
1223 #(#enum_variants)*
1224 }
1225 })
1226 }
1227
1228 fn generate_field_type(
1229 &self,
1230 schema_name: &str,
1231 field_name: &str,
1232 prop: &crate::analysis::PropertyInfo,
1233 is_required: bool,
1234 analysis: &crate::analysis::SchemaAnalysis,
1235 ) -> TokenStream {
1236 use crate::analysis::SchemaType;
1237
1238 let base_type = match &prop.schema_type {
1239 SchemaType::Primitive { rust_type } => {
1240 if rust_type.contains("::") {
1242 let parts: Vec<&str> = rust_type.split("::").collect();
1243 if parts.len() == 2 {
1244 let module = format_ident!("{}", parts[0]);
1245 let type_name = format_ident!("{}", parts[1]);
1246 quote! { #module::#type_name }
1247 } else {
1248 let path_parts: Vec<_> =
1250 parts.iter().map(|p| format_ident!("{}", p)).collect();
1251 quote! { #(#path_parts)::* }
1252 }
1253 } else {
1254 let type_ident = format_ident!("{}", rust_type);
1255 quote! { #type_ident }
1256 }
1257 }
1258 SchemaType::Reference { target } => {
1259 let target_type = format_ident!("{}", self.to_rust_type_name(target));
1260 if analysis.dependencies.recursive_schemas.contains(target) {
1262 quote! { Box<#target_type> }
1263 } else {
1264 quote! { #target_type }
1265 }
1266 }
1267 SchemaType::Array { item_type } => {
1268 let inner_type = self.generate_array_item_type(item_type, analysis);
1269 quote! { Vec<#inner_type> }
1270 }
1271 _ => {
1272 quote! { serde_json::Value }
1274 }
1275 };
1276
1277 let override_key = format!("{schema_name}.{field_name}");
1279 let is_nullable_override = self
1280 .config
1281 .nullable_field_overrides
1282 .get(&override_key)
1283 .copied()
1284 .unwrap_or(false);
1285
1286 if is_required && !prop.nullable && !is_nullable_override {
1287 if prop.default.is_some() && self.type_lacks_default(&prop.schema_type, analysis) {
1290 quote! { Option<#base_type> }
1291 } else {
1292 base_type
1293 }
1294 } else {
1295 quote! { Option<#base_type> }
1296 }
1297 }
1298
1299 fn generate_serde_field_attrs(
1300 &self,
1301 field_name: &str,
1302 prop: &crate::analysis::PropertyInfo,
1303 is_required: bool,
1304 analysis: &crate::analysis::SchemaAnalysis,
1305 ) -> TokenStream {
1306 let mut attrs = Vec::new();
1307
1308 let rust_field_name = self.to_rust_field_name(field_name);
1311 let comparison_name = rust_field_name
1312 .strip_prefix("r#")
1313 .unwrap_or(&rust_field_name);
1314 if comparison_name != field_name {
1315 attrs.push(quote! { rename = #field_name });
1316 }
1317
1318 if !is_required || prop.nullable {
1320 attrs.push(quote! { skip_serializing_if = "Option::is_none" });
1321 }
1322
1323 if prop.default.is_some()
1327 && (is_required && !prop.nullable)
1328 && !self.type_lacks_default(&prop.schema_type, analysis)
1329 {
1330 attrs.push(quote! { default });
1331 }
1332
1333 if attrs.is_empty() {
1334 TokenStream::new()
1335 } else {
1336 quote! { #[serde(#(#attrs),*)] }
1337 }
1338 }
1339
1340 fn type_lacks_default(
1344 &self,
1345 schema_type: &crate::analysis::SchemaType,
1346 analysis: &crate::analysis::SchemaAnalysis,
1347 ) -> bool {
1348 use crate::analysis::SchemaType;
1349 match schema_type {
1350 SchemaType::DiscriminatedUnion { .. } | SchemaType::Union { .. } => true,
1351 SchemaType::Reference { target } => {
1352 if let Some(schema) = analysis.schemas.get(target) {
1353 self.type_lacks_default(&schema.schema_type, analysis)
1354 } else {
1355 false
1356 }
1357 }
1358 _ => false,
1359 }
1360 }
1361
1362 fn generate_specta_field_attrs(&self, field_name: &str) -> TokenStream {
1363 if !self.config.enable_specta {
1364 return TokenStream::new();
1365 }
1366
1367 let camel_case_name = self.to_camel_case(field_name);
1369
1370 if camel_case_name != field_name {
1372 quote! { #[cfg_attr(feature = "specta", specta(rename = #camel_case_name))] }
1373 } else {
1374 TokenStream::new()
1375 }
1376 }
1377
1378 fn to_rust_enum_variant(&self, s: &str) -> String {
1379 let mut result = String::new();
1381 let mut next_upper = true;
1382 let mut prev_was_upper = false;
1383
1384 for (i, c) in s.chars().enumerate() {
1385 match c {
1386 'a'..='z' => {
1387 if next_upper {
1388 result.push(c.to_ascii_uppercase());
1389 next_upper = false;
1390 } else {
1391 result.push(c);
1392 }
1393 prev_was_upper = false;
1394 }
1395 'A'..='Z' => {
1396 if next_upper || (!prev_was_upper && i > 0) {
1397 result.push(c);
1399 next_upper = false;
1400 } else {
1401 result.push(c.to_ascii_lowercase());
1403 }
1404 prev_was_upper = true;
1405 }
1406 '0'..='9' => {
1407 result.push(c);
1408 next_upper = false;
1409 prev_was_upper = false;
1410 }
1411 '.' | '-' | '_' | ' ' | '@' | '#' | '$' | '/' | '\\' => {
1412 next_upper = true;
1414 prev_was_upper = false;
1415 }
1416 _ => {
1417 next_upper = true;
1419 prev_was_upper = false;
1420 }
1421 }
1422 }
1423
1424 if result.is_empty() {
1426 result = "Value".to_string();
1427 }
1428
1429 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1431 result = format!("Variant{result}");
1432 }
1433
1434 match result.as_str() {
1436 "Null" => "NullValue".to_string(),
1437 "True" => "TrueValue".to_string(),
1438 "False" => "FalseValue".to_string(),
1439 "Type" => "Type_".to_string(),
1440 "Match" => "Match_".to_string(),
1441 "Fn" => "Fn_".to_string(),
1442 "Impl" => "Impl_".to_string(),
1443 "Trait" => "Trait_".to_string(),
1444 "Struct" => "Struct_".to_string(),
1445 "Enum" => "Enum_".to_string(),
1446 "Mod" => "Mod_".to_string(),
1447 "Use" => "Use_".to_string(),
1448 "Pub" => "Pub_".to_string(),
1449 "Const" => "Const_".to_string(),
1450 "Static" => "Static_".to_string(),
1451 "Let" => "Let_".to_string(),
1452 "Mut" => "Mut_".to_string(),
1453 "Ref" => "Ref_".to_string(),
1454 "Move" => "Move_".to_string(),
1455 "Return" => "Return_".to_string(),
1456 "If" => "If_".to_string(),
1457 "Else" => "Else_".to_string(),
1458 "While" => "While_".to_string(),
1459 "For" => "For_".to_string(),
1460 "Loop" => "Loop_".to_string(),
1461 "Break" => "Break_".to_string(),
1462 "Continue" => "Continue_".to_string(),
1463 "Self" => "Self_".to_string(),
1464 "Super" => "Super_".to_string(),
1465 "Crate" => "Crate_".to_string(),
1466 "Async" => "Async_".to_string(),
1467 "Await" => "Await_".to_string(),
1468 _ => result,
1469 }
1470 }
1471
1472 #[allow(dead_code)]
1473 fn to_rust_identifier(&self, s: &str) -> String {
1474 let mut result = s
1476 .chars()
1477 .map(|c| match c {
1478 'a'..='z' | 'A'..='Z' | '0'..='9' => c,
1479 '.' | '-' | '_' | ' ' | '@' | '#' | '$' | '/' | '\\' => '_',
1480 _ => '_',
1481 })
1482 .collect::<String>();
1483
1484 result = result.trim_matches('_').to_string();
1486
1487 if result.is_empty() {
1489 result = "value".to_string();
1490 }
1491
1492 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1494 result = format!("variant_{result}");
1495 }
1496
1497 match result.as_str() {
1499 "null" => "null_value".to_string(),
1500 "true" => "true_value".to_string(),
1501 "false" => "false_value".to_string(),
1502 "type" => "type_".to_string(),
1503 "match" => "match_".to_string(),
1504 "fn" => "fn_".to_string(),
1505 "impl" => "impl_".to_string(),
1506 "trait" => "trait_".to_string(),
1507 "struct" => "struct_".to_string(),
1508 "enum" => "enum_".to_string(),
1509 "mod" => "mod_".to_string(),
1510 "use" => "use_".to_string(),
1511 "pub" => "pub_".to_string(),
1512 "const" => "const_".to_string(),
1513 "static" => "static_".to_string(),
1514 "let" => "let_".to_string(),
1515 "mut" => "mut_".to_string(),
1516 "ref" => "ref_".to_string(),
1517 "move" => "move_".to_string(),
1518 "return" => "return_".to_string(),
1519 "if" => "if_".to_string(),
1520 "else" => "else_".to_string(),
1521 "while" => "while_".to_string(),
1522 "for" => "for_".to_string(),
1523 "loop" => "loop_".to_string(),
1524 "break" => "break_".to_string(),
1525 "continue" => "continue_".to_string(),
1526 "self" => "self_".to_string(),
1527 "super" => "super_".to_string(),
1528 "crate" => "crate_".to_string(),
1529 "async" => "async_".to_string(),
1530 "await" => "await_".to_string(),
1531 "override" => "override_".to_string(),
1533 "box" => "box_".to_string(),
1534 "dyn" => "dyn_".to_string(),
1535 "where" => "where_".to_string(),
1536 "in" => "in_".to_string(),
1537 "abstract" => "abstract_".to_string(),
1539 "become" => "become_".to_string(),
1540 "do" => "do_".to_string(),
1541 "final" => "final_".to_string(),
1542 "macro" => "macro_".to_string(),
1543 "priv" => "priv_".to_string(),
1544 "try" => "try_".to_string(),
1545 "typeof" => "typeof_".to_string(),
1546 "unsized" => "unsized_".to_string(),
1547 "virtual" => "virtual_".to_string(),
1548 "yield" => "yield_".to_string(),
1549 _ => result,
1550 }
1551 }
1552
1553 fn sanitize_doc_comment(&self, desc: &str) -> String {
1554 let mut result = desc.to_string();
1556
1557 if result.contains('\n')
1565 && (result.contains('{')
1566 || result.contains("```")
1567 || result.contains("Human:")
1568 || result.contains("Assistant:")
1569 || result
1570 .lines()
1571 .any(|line| line.trim().starts_with('"') && line.trim().ends_with('"')))
1572 {
1573 if result.contains("```") {
1575 result = result.replace("```", "```ignore");
1576 } else {
1577 if result.lines().any(|line| {
1579 let trimmed = line.trim();
1580 trimmed.starts_with('"') && trimmed.ends_with('"') && trimmed.len() > 2
1581 }) {
1582 result = format!("```ignore\n{result}\n```");
1583 }
1584 }
1585 }
1586
1587 result
1588 }
1589
1590 pub(crate) fn to_rust_type_name(&self, s: &str) -> String {
1591 let mut result = String::new();
1593 let mut next_upper = true;
1594 let mut prev_was_lower = false;
1595
1596 for c in s.chars() {
1597 match c {
1598 'a'..='z' => {
1599 if next_upper {
1600 result.push(c.to_ascii_uppercase());
1601 next_upper = false;
1602 } else {
1603 result.push(c);
1604 }
1605 prev_was_lower = true;
1606 }
1607 'A'..='Z' => {
1608 result.push(c);
1609 next_upper = false;
1610 prev_was_lower = false;
1611 }
1612 '0'..='9' => {
1613 if prev_was_lower && !result.chars().last().unwrap_or(' ').is_ascii_digit() {
1616 }
1618 result.push(c);
1619 next_upper = false;
1620 prev_was_lower = false;
1621 }
1622 '_' | '-' | '.' | ' ' => {
1623 next_upper = true;
1625 prev_was_lower = false;
1626 }
1627 _ => {
1628 next_upper = true;
1630 prev_was_lower = false;
1631 }
1632 }
1633 }
1634
1635 if result.is_empty() {
1637 result = "Type".to_string();
1638 }
1639
1640 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1642 result = format!("Type{result}");
1643 }
1644
1645 result
1646 }
1647
1648 fn to_rust_field_name(&self, s: &str) -> String {
1649 let mut result = String::new();
1651 let mut prev_was_upper = false;
1652 let mut prev_was_underscore = false;
1653
1654 for (i, c) in s.chars().enumerate() {
1655 match c {
1656 'A'..='Z' => {
1657 if i > 0 && !prev_was_upper && !prev_was_underscore {
1659 result.push('_');
1660 }
1661 result.push(c.to_ascii_lowercase());
1662 prev_was_upper = true;
1663 prev_was_underscore = false;
1664 }
1665 'a'..='z' | '0'..='9' => {
1666 result.push(c);
1667 prev_was_upper = false;
1668 prev_was_underscore = false;
1669 }
1670 '-' | '.' | '_' | '@' | '#' | '$' | ' ' => {
1671 if !prev_was_underscore && !result.is_empty() {
1672 result.push('_');
1673 prev_was_underscore = true;
1674 }
1675 prev_was_upper = false;
1676 }
1677 _ => {
1678 if !prev_was_underscore && !result.is_empty() {
1680 result.push('_');
1681 }
1682 prev_was_upper = false;
1683 prev_was_underscore = true;
1684 }
1685 }
1686 }
1687
1688 let mut result = result.trim_matches('_').to_string();
1690 if result.is_empty() {
1691 return "field".to_string();
1692 }
1693
1694 if result.chars().next().is_some_and(|c| c.is_ascii_digit()) {
1696 result = format!("field_{result}");
1697 }
1698
1699 if Self::is_rust_keyword(&result) {
1701 format!("r#{result}")
1702 } else {
1703 result
1704 }
1705 }
1706
1707 pub fn is_rust_keyword(s: &str) -> bool {
1709 matches!(
1710 s,
1711 "type"
1712 | "match"
1713 | "fn"
1714 | "struct"
1715 | "enum"
1716 | "impl"
1717 | "trait"
1718 | "mod"
1719 | "use"
1720 | "pub"
1721 | "const"
1722 | "static"
1723 | "let"
1724 | "mut"
1725 | "ref"
1726 | "move"
1727 | "return"
1728 | "if"
1729 | "else"
1730 | "while"
1731 | "for"
1732 | "loop"
1733 | "break"
1734 | "continue"
1735 | "self"
1736 | "super"
1737 | "crate"
1738 | "async"
1739 | "await"
1740 | "override"
1741 | "box"
1742 | "dyn"
1743 | "where"
1744 | "in"
1745 | "abstract"
1746 | "become"
1747 | "do"
1748 | "final"
1749 | "macro"
1750 | "priv"
1751 | "try"
1752 | "typeof"
1753 | "unsized"
1754 | "virtual"
1755 | "yield"
1756 )
1757 }
1758
1759 pub fn to_field_ident(name: &str) -> proc_macro2::Ident {
1761 if let Some(raw) = name.strip_prefix("r#") {
1762 proc_macro2::Ident::new_raw(raw, proc_macro2::Span::call_site())
1763 } else {
1764 proc_macro2::Ident::new(name, proc_macro2::Span::call_site())
1765 }
1766 }
1767
1768 fn to_camel_case(&self, s: &str) -> String {
1769 let mut result = String::new();
1771 let mut capitalize_next = false;
1772
1773 for (i, c) in s.chars().enumerate() {
1774 match c {
1775 '_' | '-' | '.' | ' ' => {
1776 capitalize_next = true;
1778 }
1779 'A'..='Z' => {
1780 if i == 0 {
1781 result.push(c.to_ascii_lowercase());
1783 } else if capitalize_next {
1784 result.push(c);
1785 capitalize_next = false;
1786 } else {
1787 result.push(c.to_ascii_lowercase());
1788 }
1789 }
1790 'a'..='z' | '0'..='9' => {
1791 if capitalize_next {
1792 result.push(c.to_ascii_uppercase());
1793 capitalize_next = false;
1794 } else {
1795 result.push(c);
1796 }
1797 }
1798 _ => {
1799 capitalize_next = true;
1801 }
1802 }
1803 }
1804
1805 if result.is_empty() {
1806 return "field".to_string();
1807 }
1808
1809 result
1810 }
1811
1812 fn generate_composition_struct(
1813 &self,
1814 schema: &crate::analysis::AnalyzedSchema,
1815 schemas: &[crate::analysis::SchemaRef],
1816 ) -> Result<TokenStream> {
1817 let struct_name = format_ident!("{}", self.to_rust_type_name(&schema.name));
1818
1819 let fields = schemas.iter().enumerate().map(|(i, schema_ref)| {
1825 let field_name = format_ident!("part_{}", i);
1826 let field_type = format_ident!("{}", self.to_rust_type_name(&schema_ref.target));
1827
1828 quote! {
1829 #[serde(flatten)]
1830 pub #field_name: #field_type,
1831 }
1832 });
1833
1834 let doc_comment = if let Some(desc) = &schema.description {
1835 quote! { #[doc = #desc] }
1836 } else {
1837 TokenStream::new()
1838 };
1839
1840 let derives = if self.config.enable_specta {
1842 quote! {
1843 #[derive(Debug, Clone, Deserialize, Serialize)]
1844 #[cfg_attr(feature = "specta", derive(specta::Type))]
1845 }
1846 } else {
1847 quote! {
1848 #[derive(Debug, Clone, Deserialize, Serialize)]
1849 }
1850 };
1851
1852 Ok(quote! {
1853 #doc_comment
1854 #derives
1855 pub struct #struct_name {
1856 #(#fields)*
1857 }
1858 })
1859 }
1860
1861 #[allow(dead_code)]
1862 fn find_missing_types(&self, analysis: &SchemaAnalysis) -> std::collections::HashSet<String> {
1863 let mut missing = std::collections::HashSet::new();
1864 let defined_types: std::collections::HashSet<String> =
1865 analysis.schemas.keys().cloned().collect();
1866
1867 for schema in analysis.schemas.values() {
1869 match &schema.schema_type {
1870 crate::analysis::SchemaType::Union { variants } => {
1871 for variant in variants {
1872 if !defined_types.contains(&variant.target) {
1873 missing.insert(variant.target.clone());
1874 }
1875 }
1876 }
1877 crate::analysis::SchemaType::DiscriminatedUnion { variants, .. } => {
1878 for variant in variants {
1879 if !defined_types.contains(&variant.type_name) {
1880 missing.insert(variant.type_name.clone());
1881 }
1882 }
1883 }
1884 crate::analysis::SchemaType::Object { properties, .. } => {
1885 let mut sorted_props: Vec<_> = properties.iter().collect();
1887 sorted_props.sort_by_key(|(name, _)| name.as_str());
1888 for (_, prop) in sorted_props {
1889 if let crate::analysis::SchemaType::Reference { target } = &prop.schema_type
1890 {
1891 if !defined_types.contains(target) {
1892 missing.insert(target.clone());
1893 }
1894 }
1895 }
1896 }
1897 crate::analysis::SchemaType::Reference { target }
1898 if !defined_types.contains(target) =>
1899 {
1900 missing.insert(target.clone());
1901 }
1902 _ => {}
1903 }
1904 }
1905
1906 missing
1907 }
1908
1909 #[allow(clippy::only_used_in_recursion)]
1910 fn generate_array_item_type(
1911 &self,
1912 item_type: &crate::analysis::SchemaType,
1913 analysis: &crate::analysis::SchemaAnalysis,
1914 ) -> TokenStream {
1915 use crate::analysis::SchemaType;
1916
1917 match item_type {
1918 SchemaType::Primitive { rust_type } => {
1919 if rust_type.contains("::") {
1921 let parts: Vec<&str> = rust_type.split("::").collect();
1922 if parts.len() == 2 {
1923 let module = format_ident!("{}", parts[0]);
1924 let type_name = format_ident!("{}", parts[1]);
1925 quote! { #module::#type_name }
1926 } else {
1927 let path_parts: Vec<_> =
1929 parts.iter().map(|p| format_ident!("{}", p)).collect();
1930 quote! { #(#path_parts)::* }
1931 }
1932 } else {
1933 let type_ident = format_ident!("{}", rust_type);
1934 quote! { #type_ident }
1935 }
1936 }
1937 SchemaType::Reference { target } => {
1938 let target_type = format_ident!("{}", self.to_rust_type_name(target));
1939 if analysis.dependencies.recursive_schemas.contains(target) {
1941 quote! { Box<#target_type> }
1942 } else {
1943 quote! { #target_type }
1944 }
1945 }
1946 SchemaType::Array { item_type } => {
1947 let inner_type = self.generate_array_item_type(item_type, analysis);
1949 quote! { Vec<#inner_type> }
1950 }
1951 _ => {
1952 quote! { serde_json::Value }
1954 }
1955 }
1956 }
1957
1958 fn type_name_to_variant_name(&self, type_name: &str) -> String {
1960 match type_name {
1962 "bool" => return "Boolean".to_string(),
1963 "i8" | "i16" | "i32" | "i64" | "i128" => return "Integer".to_string(),
1964 "u8" | "u16" | "u32" | "u64" | "u128" => return "UnsignedInteger".to_string(),
1965 "f32" | "f64" => return "Number".to_string(),
1966 "String" => return "String".to_string(),
1967 "serde_json::Value" => return "Value".to_string(),
1968 _ => {}
1969 }
1970
1971 if type_name.starts_with("Vec<") && type_name.ends_with(">") {
1973 let inner = &type_name[4..type_name.len() - 1];
1974 if inner.starts_with("Vec<") && inner.ends_with(">") {
1976 let inner_inner = &inner[4..inner.len() - 1];
1977 return format!("{}ArrayArray", self.type_name_to_variant_name(inner_inner));
1978 }
1979 return format!("{}Array", self.type_name_to_variant_name(inner));
1980 }
1981
1982 let clean_name = type_name
1988 .trim_end_matches("Type")
1989 .trim_end_matches("Schema")
1990 .trim_end_matches("Item");
1991
1992 self.to_rust_type_name(clean_name)
1994 }
1995
1996 fn ensure_unique_variant_name_generator(
1998 &self,
1999 base_name: String,
2000 used_names: &mut std::collections::HashSet<String>,
2001 fallback_index: usize,
2002 ) -> String {
2003 if used_names.insert(base_name.clone()) {
2004 return base_name;
2005 }
2006
2007 for i in 2..100 {
2009 let numbered_name = format!("{base_name}{i}");
2010 if used_names.insert(numbered_name.clone()) {
2011 return numbered_name;
2012 }
2013 }
2014
2015 let fallback = format!("Variant{fallback_index}");
2017 used_names.insert(fallback.clone());
2018 fallback
2019 }
2020
2021 fn find_request_type_for_operation(
2023 &self,
2024 operation_id: &str,
2025 analysis: &SchemaAnalysis,
2026 ) -> Option<String> {
2027 analysis.operations.get(operation_id).and_then(|op| {
2029 op.request_body
2030 .as_ref()
2031 .and_then(|rb| rb.schema_name().map(|s| s.to_string()))
2032 })
2033 }
2034
2035 fn resolve_streaming_event_type(
2037 &self,
2038 endpoint: &crate::streaming::StreamingEndpoint,
2039 analysis: &SchemaAnalysis,
2040 ) -> Result<String> {
2041 match &endpoint.event_flow {
2042 crate::streaming::EventFlow::Simple => {
2043 if analysis.schemas.contains_key(&endpoint.event_union_type) {
2046 Ok(endpoint.event_union_type.to_string())
2047 } else {
2048 Err(crate::error::GeneratorError::ValidationError(format!(
2049 "Streaming response type '{}' not found in schema for simple streaming endpoint '{}'",
2050 endpoint.event_union_type, endpoint.operation_id
2051 )))
2052 }
2053 }
2054 crate::streaming::EventFlow::StartDeltaStop { .. } => {
2055 if analysis.schemas.contains_key(&endpoint.event_union_type) {
2058 Ok(endpoint.event_union_type.to_string())
2059 } else {
2060 Err(crate::error::GeneratorError::ValidationError(format!(
2061 "Event union type '{}' not found in schema for complex streaming endpoint '{}'",
2062 endpoint.event_union_type, endpoint.operation_id
2063 )))
2064 }
2065 }
2066 }
2067 }
2068
2069 fn generate_streaming_error_types(&self) -> Result<TokenStream> {
2071 Ok(quote! {
2072 #[derive(Debug, thiserror::Error)]
2074 pub enum StreamingError {
2075 #[error("Connection error: {0}")]
2076 Connection(String),
2077 #[error("HTTP error: {status}")]
2078 Http { status: u16 },
2079 #[error("SSE parsing error: {0}")]
2080 Parsing(String),
2081 #[error("Authentication error: {0}")]
2082 Authentication(String),
2083 #[error("Rate limit error: {0}")]
2084 RateLimit(String),
2085 #[error("API error: {0}")]
2086 Api(String),
2087 #[error("Timeout error: {0}")]
2088 Timeout(String),
2089 #[error("JSON serialization/deserialization error: {0}")]
2090 Json(#[from] serde_json::Error),
2091 #[error("Request error: {0}")]
2092 Request(reqwest::Error),
2093 }
2094
2095 impl From<reqwest::header::InvalidHeaderValue> for StreamingError {
2096 fn from(err: reqwest::header::InvalidHeaderValue) -> Self {
2097 StreamingError::Api(format!("Invalid header value: {}", err))
2098 }
2099 }
2100
2101 impl From<reqwest::Error> for StreamingError {
2102 fn from(err: reqwest::Error) -> Self {
2103 if err.is_timeout() {
2104 StreamingError::Timeout(err.to_string())
2105 } else if err.is_status() {
2106 if let Some(status) = err.status() {
2107 StreamingError::Http { status: status.as_u16() }
2108 } else {
2109 StreamingError::Connection(err.to_string())
2110 }
2111 } else {
2112 StreamingError::Request(err)
2113 }
2114 }
2115 }
2116 })
2117 }
2118
2119 fn generate_endpoint_trait(
2121 &self,
2122 endpoint: &crate::streaming::StreamingEndpoint,
2123 analysis: &SchemaAnalysis,
2124 ) -> Result<TokenStream> {
2125 use crate::streaming::HttpMethod;
2126
2127 let trait_name = format_ident!(
2128 "{}StreamingClient",
2129 self.to_rust_type_name(&endpoint.operation_id)
2130 );
2131 let method_name =
2132 format_ident!("stream_{}", self.to_rust_field_name(&endpoint.operation_id));
2133 let event_type =
2134 format_ident!("{}", self.resolve_streaming_event_type(endpoint, analysis)?);
2135
2136 let method_signature = match endpoint.http_method {
2138 HttpMethod::Get => {
2139 let mut param_defs = Vec::new();
2141 for qp in &endpoint.query_parameters {
2142 let param_name = format_ident!("{}", self.to_rust_field_name(&qp.name));
2143 if qp.required {
2144 param_defs.push(quote! { #param_name: &str });
2145 } else {
2146 param_defs.push(quote! { #param_name: Option<&str> });
2147 }
2148 }
2149 quote! {
2150 async fn #method_name(
2151 &self,
2152 #(#param_defs),*
2153 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error>;
2154 }
2155 }
2156 HttpMethod::Post => {
2157 let request_type = self
2159 .find_request_type_for_operation(&endpoint.operation_id, analysis)
2160 .unwrap_or_else(|| "serde_json::Value".to_string());
2161 let request_type_ident = if request_type.contains("::") {
2162 let parts: Vec<&str> = request_type.split("::").collect();
2163 let path_parts: Vec<_> = parts.iter().map(|p| format_ident!("{}", p)).collect();
2164 quote! { #(#path_parts)::* }
2165 } else {
2166 let ident = format_ident!("{}", request_type);
2167 quote! { #ident }
2168 };
2169 quote! {
2170 async fn #method_name(
2171 &self,
2172 request: #request_type_ident,
2173 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error>;
2174 }
2175 }
2176 };
2177
2178 Ok(quote! {
2179 #[async_trait]
2181 pub trait #trait_name {
2182 type Error: std::error::Error + Send + Sync + 'static;
2183
2184 #method_signature
2186 }
2187 })
2188 }
2189
2190 fn generate_streaming_client_impl(
2192 &self,
2193 streaming_config: &crate::streaming::StreamingConfig,
2194 analysis: &SchemaAnalysis,
2195 ) -> Result<TokenStream> {
2196 let client_name = format_ident!(
2197 "{}Client",
2198 self.to_rust_type_name(&streaming_config.client_module_name)
2199 );
2200
2201 let mut struct_fields = vec![
2204 quote! { base_url: String },
2205 quote! { api_key: Option<String> },
2206 quote! { http_client: reqwest::Client },
2207 quote! { custom_headers: std::collections::BTreeMap<String, String> },
2208 ];
2209
2210 let has_optional_headers = !streaming_config
2211 .endpoints
2212 .iter()
2213 .all(|e| e.optional_headers.is_empty());
2214
2215 if has_optional_headers {
2216 struct_fields
2217 .push(quote! { optional_headers: std::collections::BTreeMap<String, String> });
2218 }
2219
2220 let default_base_url = if let Some(ref streaming_config) = self.config.streaming_config {
2223 streaming_config
2224 .endpoints
2225 .first()
2226 .and_then(|e| e.base_url.as_deref())
2227 .unwrap_or("https://api.example.com")
2228 } else {
2229 "https://api.example.com"
2230 };
2231
2232 let constructor_fields = if has_optional_headers {
2234 quote! {
2235 base_url: #default_base_url.to_string(),
2236 api_key: None,
2237 http_client: reqwest::Client::new(),
2238 custom_headers: std::collections::BTreeMap::new(),
2239 optional_headers: std::collections::BTreeMap::new(),
2240 }
2241 } else {
2242 quote! {
2243 base_url: #default_base_url.to_string(),
2244 api_key: None,
2245 http_client: reqwest::Client::new(),
2246 custom_headers: std::collections::BTreeMap::new(),
2247 }
2248 };
2249
2250 let optional_headers_method = if has_optional_headers {
2252 quote! {
2253 pub fn set_optional_headers(&mut self, headers: std::collections::BTreeMap<String, String>) {
2255 self.optional_headers = headers;
2256 }
2257 }
2258 } else {
2259 TokenStream::new()
2260 };
2261
2262 let constructor = quote! {
2263 impl #client_name {
2264 pub fn new() -> Self {
2266 Self {
2267 #constructor_fields
2268 }
2269 }
2270
2271 pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
2273 self.base_url = base_url.into();
2274 self
2275 }
2276
2277 pub fn with_api_key(mut self, api_key: impl Into<String>) -> Self {
2279 self.api_key = Some(api_key.into());
2280 self
2281 }
2282
2283 pub fn with_header(
2285 mut self,
2286 name: impl Into<String>,
2287 value: impl Into<String>,
2288 ) -> Self {
2289 self.custom_headers.insert(name.into(), value.into());
2290 self
2291 }
2292
2293 pub fn with_http_client(mut self, client: reqwest::Client) -> Self {
2295 self.http_client = client;
2296 self
2297 }
2298
2299 #optional_headers_method
2300 }
2301 };
2302
2303 let mut trait_impls = Vec::new();
2305 for endpoint in &streaming_config.endpoints {
2306 let trait_impl = self.generate_endpoint_trait_impl(endpoint, &client_name, analysis)?;
2307 trait_impls.push(trait_impl);
2308 }
2309
2310 let default_impl = quote! {
2312 impl Default for #client_name {
2313 fn default() -> Self {
2314 Self::new()
2315 }
2316 }
2317 };
2318
2319 Ok(quote! {
2320 #[derive(Debug, Clone)]
2322 pub struct #client_name {
2323 #(#struct_fields,)*
2324 }
2325
2326 #constructor
2327
2328 #default_impl
2329
2330 #(#trait_impls)*
2331 })
2332 }
2333
2334 fn generate_endpoint_trait_impl(
2336 &self,
2337 endpoint: &crate::streaming::StreamingEndpoint,
2338 client_name: &proc_macro2::Ident,
2339 analysis: &SchemaAnalysis,
2340 ) -> Result<TokenStream> {
2341 use crate::streaming::HttpMethod;
2342
2343 let trait_name = format_ident!(
2344 "{}StreamingClient",
2345 self.to_rust_type_name(&endpoint.operation_id)
2346 );
2347 let method_name =
2348 format_ident!("stream_{}", self.to_rust_field_name(&endpoint.operation_id));
2349 let event_type =
2350 format_ident!("{}", self.resolve_streaming_event_type(endpoint, analysis)?);
2351
2352 let mut header_setup = Vec::new();
2354 for (name, value) in &endpoint.required_headers {
2355 header_setup.push(quote! {
2356 headers.insert(#name, HeaderValue::from_static(#value));
2357 });
2358 }
2359
2360 if let Some(auth_header) = &endpoint.auth_header {
2363 match auth_header {
2364 crate::streaming::AuthHeader::Bearer(header_name) => {
2365 header_setup.push(quote! {
2366 if let Some(ref api_key) = self.api_key {
2367 headers.insert(#header_name, HeaderValue::from_str(&format!("Bearer {}", api_key))?);
2368 }
2369 });
2370 }
2371 crate::streaming::AuthHeader::ApiKey(header_name) => {
2372 header_setup.push(quote! {
2373 if let Some(ref api_key) = self.api_key {
2374 headers.insert(#header_name, HeaderValue::from_str(api_key)?);
2375 }
2376 });
2377 }
2378 }
2379 } else {
2380 header_setup.push(quote! {
2382 if let Some(ref api_key) = self.api_key {
2383 headers.insert("Authorization", HeaderValue::from_str(&format!("Bearer {}", api_key))?);
2384 }
2385 });
2386 }
2387
2388 header_setup.push(quote! {
2390 for (name, value) in &self.custom_headers {
2391 if let (Ok(header_name), Ok(header_value)) = (reqwest::header::HeaderName::from_bytes(name.as_bytes()), HeaderValue::from_str(value)) {
2392 headers.insert(header_name, header_value);
2393 }
2394 }
2395 });
2396
2397 if !endpoint.optional_headers.is_empty() {
2399 header_setup.push(quote! {
2400 for (key, value) in &self.optional_headers {
2401 if let (Ok(header_name), Ok(header_value)) = (reqwest::header::HeaderName::from_bytes(key.as_bytes()), HeaderValue::from_str(value)) {
2402 headers.insert(header_name, header_value);
2403 }
2404 }
2405 });
2406 }
2407
2408 match endpoint.http_method {
2410 HttpMethod::Get => self.generate_get_streaming_impl(
2411 endpoint,
2412 client_name,
2413 &trait_name,
2414 &method_name,
2415 &event_type,
2416 &header_setup,
2417 ),
2418 HttpMethod::Post => self.generate_post_streaming_impl(
2419 endpoint,
2420 client_name,
2421 &trait_name,
2422 &method_name,
2423 &event_type,
2424 &header_setup,
2425 analysis,
2426 ),
2427 }
2428 }
2429
2430 fn generate_get_streaming_impl(
2432 &self,
2433 endpoint: &crate::streaming::StreamingEndpoint,
2434 client_name: &proc_macro2::Ident,
2435 trait_name: &proc_macro2::Ident,
2436 method_name: &proc_macro2::Ident,
2437 event_type: &proc_macro2::Ident,
2438 header_setup: &[TokenStream],
2439 ) -> Result<TokenStream> {
2440 let path = &endpoint.path;
2441
2442 let mut param_defs = Vec::new();
2444 let mut query_params = Vec::new();
2445
2446 for qp in &endpoint.query_parameters {
2447 let param_name = format_ident!("{}", self.to_rust_field_name(&qp.name));
2448 let param_name_str = &qp.name;
2449
2450 if qp.required {
2451 param_defs.push(quote! { #param_name: &str });
2452 query_params.push(quote! {
2453 url.query_pairs_mut().append_pair(#param_name_str, #param_name);
2454 });
2455 } else {
2456 param_defs.push(quote! { #param_name: Option<&str> });
2457 query_params.push(quote! {
2458 if let Some(v) = #param_name {
2459 url.query_pairs_mut().append_pair(#param_name_str, v);
2460 }
2461 });
2462 }
2463 }
2464
2465 let url_construction = quote! {
2467 let base_url = url::Url::parse(&self.base_url)
2468 .map_err(|e| StreamingError::Connection(format!("Invalid base URL: {}", e)))?;
2469 let path_to_join = #path.trim_start_matches('/');
2470 let mut url = base_url.join(path_to_join)
2471 .map_err(|e| StreamingError::Connection(format!("URL join error: {}", e)))?;
2472 #(#query_params)*
2473 };
2474
2475 let instrument_skip = quote! { #[instrument(skip(self), name = "streaming_get_request")] };
2476
2477 Ok(quote! {
2478 #[async_trait]
2479 impl #trait_name for #client_name {
2480 type Error = StreamingError;
2481
2482 #instrument_skip
2483 async fn #method_name(
2484 &self,
2485 #(#param_defs),*
2486 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error> {
2487 debug!("Starting streaming GET request");
2488
2489 let mut headers = HeaderMap::new();
2490 #(#header_setup)*
2491
2492 #url_construction
2493 let url_str = url.to_string();
2494 debug!("Making streaming GET request to: {}", url_str);
2495
2496 let request_builder = self.http_client
2497 .get(url_str)
2498 .headers(headers);
2499
2500 debug!("Creating SSE stream from request");
2501 let stream = parse_sse_stream::<#event_type>(request_builder).await?;
2502 info!("SSE stream created successfully");
2503 Ok(Box::pin(stream))
2504 }
2505 }
2506 })
2507 }
2508
2509 #[allow(clippy::too_many_arguments)]
2511 fn generate_post_streaming_impl(
2512 &self,
2513 endpoint: &crate::streaming::StreamingEndpoint,
2514 client_name: &proc_macro2::Ident,
2515 trait_name: &proc_macro2::Ident,
2516 method_name: &proc_macro2::Ident,
2517 event_type: &proc_macro2::Ident,
2518 header_setup: &[TokenStream],
2519 analysis: &SchemaAnalysis,
2520 ) -> Result<TokenStream> {
2521 let path = &endpoint.path;
2522
2523 let request_type = self
2525 .find_request_type_for_operation(&endpoint.operation_id, analysis)
2526 .unwrap_or_else(|| "serde_json::Value".to_string());
2527 let request_type_ident = if request_type.contains("::") {
2528 let parts: Vec<&str> = request_type.split("::").collect();
2529 let path_parts: Vec<_> = parts.iter().map(|p| format_ident!("{}", p)).collect();
2530 quote! { #(#path_parts)::* }
2531 } else {
2532 let ident = format_ident!("{}", request_type);
2533 quote! { #ident }
2534 };
2535
2536 let url_construction = quote! {
2538 let base_url = url::Url::parse(&self.base_url)
2539 .map_err(|e| StreamingError::Connection(format!("Invalid base URL: {}", e)))?;
2540 let path_to_join = #path.trim_start_matches('/');
2541 let url = base_url.join(path_to_join)
2542 .map_err(|e| StreamingError::Connection(format!("URL join error: {}", e)))?
2543 .to_string();
2544 };
2545
2546 let stream_param = &endpoint.stream_parameter;
2548 let stream_setup = if stream_param.is_empty() {
2549 quote! {
2550 let streaming_request = request;
2551 }
2552 } else {
2553 quote! {
2554 let mut streaming_request = request;
2556 if let Ok(mut request_value) = serde_json::to_value(&streaming_request) {
2557 if let Some(obj) = request_value.as_object_mut() {
2558 obj.insert(#stream_param.to_string(), serde_json::Value::Bool(true));
2559 }
2560 streaming_request = serde_json::from_value(request_value)?;
2561 }
2562 }
2563 };
2564
2565 Ok(quote! {
2566 #[async_trait]
2567 impl #trait_name for #client_name {
2568 type Error = StreamingError;
2569
2570 #[instrument(skip(self, request), name = "streaming_post_request")]
2571 async fn #method_name(
2572 &self,
2573 request: #request_type_ident,
2574 ) -> Result<Pin<Box<dyn Stream<Item = Result<#event_type, Self::Error>> + Send>>, Self::Error> {
2575 debug!("Starting streaming POST request");
2576
2577 #stream_setup
2578
2579 let mut headers = HeaderMap::new();
2580 #(#header_setup)*
2581
2582 #url_construction
2583 debug!("Making streaming POST request to: {}", url);
2584
2585 let request_builder = self.http_client
2586 .post(&url)
2587 .headers(headers)
2588 .json(&streaming_request);
2589
2590 debug!("Creating SSE stream from request");
2591 let stream = parse_sse_stream::<#event_type>(request_builder).await?;
2592 info!("SSE stream created successfully");
2593 Ok(Box::pin(stream))
2594 }
2595 }
2596 })
2597 }
2598
2599 fn generate_sse_parser_utilities(
2601 &self,
2602 _streaming_config: &crate::streaming::StreamingConfig,
2603 ) -> Result<TokenStream> {
2604 Ok(quote! {
2605 pub async fn parse_sse_stream<T>(
2607 request_builder: reqwest::RequestBuilder
2608 ) -> Result<impl Stream<Item = Result<T, StreamingError>>, StreamingError>
2609 where
2610 T: serde::de::DeserializeOwned + Send + 'static,
2611 {
2612 let mut event_source = reqwest_eventsource::EventSource::new(request_builder).map_err(|e| {
2613 StreamingError::Connection(format!("Failed to create event source: {}", e))
2614 })?;
2615
2616 let stream = event_source.filter_map(|event_result| async move {
2617 match event_result {
2618 Ok(reqwest_eventsource::Event::Open) => {
2619 debug!("SSE connection opened");
2620 None
2621 }
2622 Ok(reqwest_eventsource::Event::Message(message)) => {
2623 if message.event == "ping" {
2625 debug!("Received SSE ping event, skipping");
2626 return None;
2627 }
2628
2629 if message.data.trim().is_empty() {
2631 debug!("Empty SSE data, skipping");
2632 return None;
2633 }
2634
2635 if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(&message.data) {
2637 if let Some(event_type) = json_value.get("event").and_then(|v| v.as_str()) {
2638 if event_type == "ping" {
2639 debug!("Received ping event in JSON data, skipping");
2640 return None;
2641 }
2642 }
2643
2644 match serde_json::from_value::<T>(json_value) {
2646 Ok(parsed_event) => {
2647 Some(Ok(parsed_event))
2648 }
2649 Err(e) => {
2650 if message.data.contains("ping") || message.event.contains("ping") {
2651 debug!("Ignoring ping-related event: {}", message.data);
2652 None
2653 } else {
2654 Some(Err(StreamingError::Parsing(
2655 format!("Failed to parse SSE event: {} (raw: {})", e, message.data)
2656 )))
2657 }
2658 }
2659 }
2660 } else {
2661 Some(Err(StreamingError::Parsing(
2663 format!("SSE event is not valid JSON: {}", message.data)
2664 )))
2665 }
2666 }
2667 Err(e) => {
2668 match e {
2670 reqwest_eventsource::Error::StreamEnded => {
2671 debug!("SSE stream completed normally");
2672 None }
2674 reqwest_eventsource::Error::InvalidStatusCode(status, response) => {
2675 let status_code = status.as_u16();
2677
2678 let error_body = match response.text().await {
2680 Ok(body) => body,
2681 Err(_) => "Failed to read error response body".to_string()
2682 };
2683
2684 error!("SSE connection error - HTTP {}: {}", status_code, error_body);
2685
2686 let detailed_error = format!(
2687 "HTTP {} error: {}",
2688 status_code,
2689 error_body
2690 );
2691
2692 Some(Err(StreamingError::Connection(detailed_error)))
2693 }
2694 _ => {
2695 let error_str = e.to_string();
2696 if error_str.contains("stream closed") {
2697 debug!("SSE stream closed");
2698 None
2699 } else {
2700 error!("SSE connection error: {}", e);
2701 Some(Err(StreamingError::Connection(error_str)))
2702 }
2703 }
2704 }
2705 }
2706 }
2707 });
2708
2709 Ok(stream)
2710 }
2711 })
2712 }
2713
2714 fn generate_reconnection_utilities(
2716 &self,
2717 reconnect_config: &crate::streaming::ReconnectionConfig,
2718 ) -> Result<TokenStream> {
2719 let max_retries = reconnect_config.max_retries;
2720 let initial_delay = reconnect_config.initial_delay_ms;
2721 let max_delay = reconnect_config.max_delay_ms;
2722 let backoff_multiplier = reconnect_config.backoff_multiplier;
2723
2724 Ok(quote! {
2725 #[derive(Debug, Clone)]
2727 pub struct ReconnectionManager {
2728 max_retries: u32,
2729 initial_delay_ms: u64,
2730 max_delay_ms: u64,
2731 backoff_multiplier: f64,
2732 current_attempt: u32,
2733 }
2734
2735 impl ReconnectionManager {
2736 pub fn new() -> Self {
2738 Self {
2739 max_retries: #max_retries,
2740 initial_delay_ms: #initial_delay,
2741 max_delay_ms: #max_delay,
2742 backoff_multiplier: #backoff_multiplier,
2743 current_attempt: 0,
2744 }
2745 }
2746
2747 pub fn should_retry(&self) -> bool {
2749 self.current_attempt < self.max_retries
2750 }
2751
2752 pub fn next_retry_delay(&mut self) -> Duration {
2754 if !self.should_retry() {
2755 return Duration::from_secs(0);
2756 }
2757
2758 let delay_ms = (self.initial_delay_ms as f64
2759 * self.backoff_multiplier.powi(self.current_attempt as i32)) as u64;
2760 let delay_ms = delay_ms.min(self.max_delay_ms);
2761
2762 self.current_attempt += 1;
2763 Duration::from_millis(delay_ms)
2764 }
2765
2766 pub fn reset(&mut self) {
2768 self.current_attempt = 0;
2769 }
2770
2771 pub fn current_attempt(&self) -> u32 {
2773 self.current_attempt
2774 }
2775 }
2776
2777 impl Default for ReconnectionManager {
2778 fn default() -> Self {
2779 Self::new()
2780 }
2781 }
2782 })
2783 }
2784}