type_flow_proc_macros/

lib.rs

use proc_macro::TokenStream;
use std::cmp::Ordering;
use quote::{quote, format_ident};
use syn::{parse_macro_input, Ident, Type, parse::Parse, parse::ParseStream, Token, ItemStruct, FieldsNamed, Fields};

struct ProcessorPipeline {
    pipeline_name: Ident,
    data_type: Type,
    processors: Vec<Type>,
}

impl Parse for ProcessorPipeline {
    fn parse(input: ParseStream) -> syn::Result<Self> {
        let pipeline_name: Ident = input.parse()?;
        input.parse::<Token![,]>()?;
        let data_type: Type = input.parse()?;
        input.parse::<Token![,]>()?;
        let mut processors = Vec::new();
        // Parse the remaining types as processor types
        while !input.is_empty() {
            let processor_type: Type = input.parse()?;
            processors.push(processor_type);
            if input.is_empty() {
                break;
            }
            input.parse::<Token![,]>()?;
        }
        Ok(ProcessorPipeline {
            pipeline_name,
            data_type,
            processors,
        })
    }
}

#[proc_macro]
pub fn stateful_processor_pipeline_with_index(input: TokenStream) -> TokenStream {
    let ProcessorPipeline { pipeline_name, data_type, processors } = parse_macro_input!(input as ProcessorPipeline);
    let struct_fields = processors.iter().enumerate().map(|(idx, processor_type)| {
        // Get a string representation of the type for the field name
        let type_str = get_type_name(processor_type);
        let field_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { #field_name: #processor_type }
    });
    let constructor_params = processors.iter().enumerate().map(|(idx, processor_type)| {
        let type_str = get_type_name(processor_type);
        let param_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { #param_name: #processor_type }
    });
    let field_initializers = processors.iter().enumerate().map(|(idx, processor_type)| {
        let type_str = get_type_name(processor_type);
        let field_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { #field_name }
    });
    let process_implementation = processors.iter().enumerate().map(|(idx, processor_type)| {
        let type_str = get_type_name(processor_type);
        let field_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { data = self.#field_name.process(data); }
    });
    let expanded = quote! {
        pub struct #pipeline_name {
            #(#struct_fields,)*
        }
        impl #pipeline_name {
            pub fn new(#(#constructor_params,)*) -> Self {
                Self {
                    #(#field_initializers,)*
                }
            }
        }
        impl StatefulProcessor<#data_type> for #pipeline_name {
            fn process(&mut self, mut data: #data_type) -> #data_type {
                #(#process_implementation)*
                data
            }
        }
    };
    TokenStream::from(expanded)
}

struct ProcessorInplacePipeline {
    pipeline_name: Ident,
    data_type: Type,
    error_type: Type,
    processors: Vec<Type>,
}

impl Parse for ProcessorInplacePipeline {
    fn parse(input: ParseStream) -> syn::Result<Self> {
        let pipeline_name: Ident = input.parse()?;
        input.parse::<Token![,]>()?;
        let data_type: Type = input.parse()?;
        input.parse::<Token![,]>()?;
        let error_type: Type = input.parse()?;
        input.parse::<Token![,]>()?;
        let mut processors = Vec::new();
        // Parse the remaining types as processor types
        while !input.is_empty() {
            let processor_type: Type = input.parse()?;
            processors.push(processor_type);

            if input.is_empty() {
                break;
            }
            input.parse::<Token![,]>()?;
        }
        Ok(ProcessorInplacePipeline{
            pipeline_name,
            data_type,
            error_type,
            processors,
        })
    }
}


#[proc_macro_attribute]
pub fn implement_processor_swapping(_attr: proc_macro::TokenStream, item: proc_macro::TokenStream) -> proc_macro::TokenStream {
    let input_struct_item = parse_macro_input!(item as ItemStruct);
    let struct_name = &input_struct_item.ident;
    let generics = &input_struct_item.generics;
    let (impl_generics, ty_generics, where_clause) = generics.split_for_impl();
    let type_param_idents: Vec<&Ident> = generics.type_params().map(|tp| &tp.ident).collect();
    let fields = match &input_struct_item.fields {
        Fields::Named(FieldsNamed { named, .. }) => named,
        _ => panic!("#[implement_processor_swapping] only works on structs with named fields."),
    };
    let processor_field_idents: Vec<&Ident> = fields.iter()
        .filter_map(|f| f.ident.as_ref())
        .filter(|&id| id.to_string() != "_marker")
        .collect();
    if type_param_idents.len() != processor_field_idents.len() {
        panic!(
            "#[implement_processor_swapping] detected a mismatch between the number of generic type parameters ({}) and processor fields ({}). Expected them to be equal.",
            type_param_idents.len(),
            processor_field_idents.len()
        );
    }
    // --- Trait Impls ---
    let mut trait_impls = Vec::new();
    if !type_param_idents.is_empty() {
        // Reverse
        let mut rev_types = type_param_idents.clone();
        rev_types.reverse();
        let mut rev_source_fields = processor_field_idents.clone();
        rev_source_fields.reverse();
        let rev_assignments = processor_field_idents.iter().zip(rev_source_fields.iter()).map(|(dest, src)| {
            quote!{ #dest: self.#src }
        });
        trait_impls.push(quote!{
            impl #impl_generics type_flow_traits::Reverse for #struct_name #ty_generics #where_clause {
                type Output = #struct_name<#(#rev_types),*>;
                fn reverse(self) -> Self::Output {
                    Self::Output {
                        _marker: std::marker::PhantomData,
                        #(#rev_assignments),*
                    }
                }
            }
        });
        // ShiftLeft
        let mut sl_types = type_param_idents.clone();
        sl_types.rotate_left(1);
        let mut sl_source_fields = processor_field_idents.clone();
        sl_source_fields.rotate_left(1);
        let sl_assignments = processor_field_idents.iter().zip(sl_source_fields.iter()).map(|(dest, src)| {
            quote!{ #dest: self.#src }
        });
        trait_impls.push(quote!{
            impl #impl_generics type_flow_traits::ShiftLeft for #struct_name #ty_generics #where_clause {
                type ShiftedLeft = #struct_name<#(#sl_types),*>;
                fn shift_left(self) -> Self::ShiftedLeft {
                    Self::ShiftedLeft {
                        _marker: std::marker::PhantomData,
                        #(#sl_assignments),*
                    }
                }
            }
        });
        // ShiftRight
        let mut sr_types = type_param_idents.clone();
        sr_types.rotate_right(1);
        let mut sr_source_fields = processor_field_idents.clone();
        sr_source_fields.rotate_right(1);
        let sr_assignments = processor_field_idents.iter().zip(sr_source_fields.iter()).map(|(dest, src)| {
            quote!{ #dest: self.#src }
        });
        trait_impls.push(quote!{
            impl #impl_generics type_flow_traits::ShiftRight for #struct_name #ty_generics #where_clause {
                type ShiftedRight = #struct_name<#(#sr_types),*>;
                fn shift_right(self) -> Self::ShiftedRight {
                    Self::ShiftedRight {
                        _marker: std::marker::PhantomData,
                        #(#sr_assignments),*
                    }
                }
            }
        });
        // SwapStartEnd
        let mut sse_types = type_param_idents.clone();
        let sse_length = sse_types.len();
        if sse_length > 1 { sse_types.swap(0, sse_length - 1); }
        let mut sse_source_fields = processor_field_idents.clone();
        let sse_source_length = sse_source_fields.len();
        if sse_source_length > 1 { sse_source_fields.swap(0, sse_source_length - 1); }
        let sse_assignments = processor_field_idents.iter().zip(sse_source_fields.iter()).map(|(dest, src)| {
            quote!{ #dest: self.#src }
        });
        trait_impls.push(quote!{
            impl #impl_generics type_flow_traits::SwapStartEnd for #struct_name #ty_generics #where_clause {
                type Output = #struct_name<#(#sse_types),*>;
                fn swap(self) -> Self::Output {
                    Self::Output {
                        _marker: std::marker::PhantomData,
                        #(#sse_assignments),*
                    }
                }
            }
        });
    }
    let num_processors = type_param_idents.len();
    for i in 0..num_processors {
        let mut process_indexes_before_pivot = Vec::new();
        let mut process_indexes_after_pivot = Vec::new();
        let mut process_indexes_before_interleave_index = Vec::new();
        let mut process_indexes_after_interleave_index = Vec::new();
        for j in 0..num_processors {
            if i != j {
                let mut swapped_generic_params_for_type = type_param_idents.clone();
                swapped_generic_params_for_type.swap(i, j);
                let field_i_name = processor_field_idents[i];
                let field_j_name = processor_field_idents[j];
                let mut current_field_initializers = Vec::new();
                for k in 0..num_processors {
                    let current_field_name = processor_field_idents[k];
                    if k == i {
                        current_field_initializers.push(quote! { #current_field_name: self.#field_j_name });
                    } else if k == j {
                        current_field_initializers.push(quote! { #current_field_name: self.#field_i_name });
                    } else {
                        current_field_initializers.push(quote! { #current_field_name: self.#current_field_name });
                    }
                }
                // SwapArbitraryProcessors
                trait_impls.push(quote! {
                    impl #impl_generics type_flow_traits::SwapArbitraryProcessors<#i, #j> for #struct_name #ty_generics #where_clause {
                        type SwappedOutput = #struct_name<#(#swapped_generic_params_for_type),*>;
                        fn swap_processors(self) -> Self::SwappedOutput {
                            #struct_name {
                                _marker: std::marker::PhantomData,
                                #( #current_field_initializers ),*
                            }
                        }
                    }
                });
                // PivotSpin picks an index of a processor to keep in its current location and spins all the other processors around its index.
                // PivotSwap picks an index of a processor to keep in its current location and swaps the two sections to either side, preserving their order within their section.
                match j.cmp(&i) {
                    Ordering::Equal => {},
                    Ordering::Less => { process_indexes_before_pivot.push(j); },
                    Ordering::Greater => { process_indexes_after_pivot.push(j); },
                }
            }
            // InterleavePivotSpin picks an index between processors and reorders the processors in the pipeline by spinning around the interleaved index.
            // InterleavePivotSwap picks an index between processors and reorders the processors in the pipeline by swapping the two sections to either side of the interleaved index, preserving their order within their section.
            if i != num_processors - 1 {
                let even_j_index = j * 2;
                let odd_i_index = i * 2 + 1;
                match even_j_index.cmp(&odd_i_index) {
                    Ordering::Equal => {},
                    Ordering::Less => { process_indexes_before_interleave_index.push(j); },
                    Ordering::Greater => { process_indexes_after_interleave_index.push(j); },
                }
            }
        }
        //Implement the Pivot Interleave Spin and swap traits
        //for swap you need reverse order of the indexes because you are using them like a stack
        let mut swap_indexes_before_pivot = process_indexes_before_pivot.clone();
        swap_indexes_before_pivot.reverse();
        let mut swap_indexes_after_pivot = process_indexes_after_pivot.clone();
        swap_indexes_after_pivot.reverse();
        let mut swap_assignments = Vec::new();
        let mut spin_assignments = Vec::new();
        let pivot_side_length_difference = process_indexes_before_pivot.len() as isize - process_indexes_after_pivot.len() as isize;
        if 0 == pivot_side_length_difference {
            for _ in 0..process_indexes_after_pivot.len() {
                swap_assignments.push(swap_indexes_after_pivot.pop().unwrap());
                spin_assignments.push(process_indexes_after_pivot.pop().unwrap());
            }
            swap_assignments.push(swap_indexes_before_pivot.len());
            spin_assignments.push(process_indexes_before_pivot.len());
            for _ in 0..process_indexes_before_pivot.len() {
                swap_assignments.push(swap_indexes_before_pivot.pop().unwrap());
                spin_assignments.push(process_indexes_before_pivot.pop().unwrap());
            }
        } else if 0 > pivot_side_length_difference {
            let mut swap_hanging_off_front = Vec::new();
            let mut spin_hanging_off_front = Vec::new();
            for _ in 0..pivot_side_length_difference.abs() as usize {
                swap_hanging_off_front.push(swap_indexes_after_pivot.pop().unwrap());
                spin_hanging_off_front.push(process_indexes_after_pivot.pop().unwrap());
            }
            swap_hanging_off_front.reverse();
            spin_hanging_off_front.reverse();
            for _ in 0..process_indexes_after_pivot.len() {
                swap_assignments.push(swap_indexes_after_pivot.pop().unwrap());
                spin_assignments.push(process_indexes_after_pivot.pop().unwrap());
            }
            swap_assignments.push(swap_indexes_before_pivot.len());
            spin_assignments.push(process_indexes_before_pivot.len());
            for _ in 0..process_indexes_before_pivot.len() {
                swap_assignments.push(swap_indexes_before_pivot.pop().unwrap());
                spin_assignments.push(process_indexes_before_pivot.pop().unwrap());
            }
            for _ in 0..spin_hanging_off_front.len() {
                swap_assignments.push(swap_hanging_off_front.pop().unwrap());
                spin_assignments.push(spin_hanging_off_front.pop().unwrap());
            }
        } else {
            let mut swap_hanging_off_back = Vec::new();
            let mut spin_hanging_off_back = Vec::new();
            swap_indexes_before_pivot.reverse();
            process_indexes_before_pivot.reverse();
            for _ in 0..pivot_side_length_difference.abs() as usize {
                swap_hanging_off_back.push(swap_indexes_before_pivot.pop().unwrap());
                spin_hanging_off_back.push(process_indexes_before_pivot.pop().unwrap());
            }
            swap_indexes_before_pivot.reverse();
            process_indexes_before_pivot.reverse();
            for _ in 0..spin_hanging_off_back.len() {
                swap_assignments.push(swap_hanging_off_back.pop().unwrap());
                spin_assignments.push(spin_hanging_off_back.pop().unwrap());
            }
            for _ in 0..process_indexes_after_pivot.len() {
                swap_assignments.push(swap_indexes_after_pivot.pop().unwrap());
                spin_assignments.push(process_indexes_after_pivot.pop().unwrap());
            }
            swap_assignments.push(swap_assignments.len());
            spin_assignments.push(spin_assignments.len());
            for _ in 0..process_indexes_before_pivot.len() {
                swap_assignments.push(swap_indexes_before_pivot.pop().unwrap());
                spin_assignments.push(process_indexes_before_pivot.pop().unwrap());
            }
        }
        if num_processors != spin_assignments.len() {
            panic!("swap_assignments.len() != num_processors");
        }
        let mut spun_generic_params_for_type = Vec::new();
        let mut swapped_generic_params_for_type = Vec::new();
        let mut spun_field_initializers = Vec::new();
        let mut swapped_field_initializes = Vec::new();
        for k in 0..num_processors {
            spun_generic_params_for_type.push(type_param_idents[spin_assignments[k]]);
            swapped_generic_params_for_type.push(type_param_idents[swap_assignments[k]]);
            let current_field_name = processor_field_idents[k];
            let spun_name = processor_field_idents[spin_assignments[k]];
            let swapped_name = processor_field_idents[swap_assignments[k]];
            spun_field_initializers.push(quote! { #current_field_name: self.#spun_name });
            swapped_field_initializes.push(quote! { #current_field_name: self.#swapped_name });
        }
        trait_impls.push(quote! {
            impl #impl_generics type_flow_traits::PivotSpin<#i> for #struct_name #ty_generics #where_clause {
                type PivotedSpinOutput = #struct_name<#(#spun_generic_params_for_type),*>;
                fn pivot_spin(self) -> Self::PivotedSpinOutput {
                    #struct_name {
                        _marker: std::marker::PhantomData,
                        #( #spun_field_initializers ),*
                    }
                }
            }
        });
        if i < num_processors - 1 && i > 0 {
            trait_impls.push(quote! {
                impl #impl_generics type_flow_traits::PivotSwap<#i> for #struct_name #ty_generics #where_clause {
                    type PivotedSwapOutput = #struct_name<#(#swapped_generic_params_for_type),*>;
                    fn pivot_swap(self) -> Self::PivotedSwapOutput {
                        #struct_name {
                            _marker: std::marker::PhantomData,
                            #( #swapped_field_initializes ),*
                        }
                    }
                }
            });
        }
        if i != num_processors - 1 {
            let total_length = process_indexes_before_interleave_index.len() + process_indexes_after_interleave_index.len();
            if num_processors != total_length {
                panic!("Wrong Total Length For Interleave");
            }

            let mut swap_indexes_before_interleave = process_indexes_before_interleave_index.clone();
            swap_indexes_before_interleave.reverse();
            let mut swap_indexes_after_interleave = process_indexes_after_interleave_index.clone();
            swap_indexes_after_interleave.reverse();
            let mut swap_assignments_interleave = Vec::new();
            let mut spin_assignments_interleave = Vec::new();
            let interleave_side_length_difference = swap_indexes_before_interleave.len() as isize - swap_indexes_after_interleave.len() as isize;
            if 0 == interleave_side_length_difference {
                for _ in 0..process_indexes_after_interleave_index.len() {
                    swap_assignments_interleave.push(swap_indexes_after_interleave.pop().unwrap());
                    spin_assignments_interleave.push(process_indexes_after_interleave_index.pop().unwrap());
                }
                for _ in 0..process_indexes_before_interleave_index.len() {
                    swap_assignments_interleave.push(swap_indexes_before_interleave.pop().unwrap());
                    spin_assignments_interleave.push(process_indexes_before_interleave_index.pop().unwrap());
                }
            }else if 0 > interleave_side_length_difference {
                let mut swap_hanging_off_front = Vec::new();
                let mut spin_hanging_off_front = Vec::new();
                for _ in 0..interleave_side_length_difference.abs() as usize {
                    swap_hanging_off_front.push(swap_indexes_after_interleave.pop().unwrap());
                    spin_hanging_off_front.push(process_indexes_after_interleave_index.pop().unwrap());
                }
                swap_hanging_off_front.reverse();
                spin_hanging_off_front.reverse();
                for _ in 0..process_indexes_after_interleave_index.len() {
                    swap_assignments_interleave.push(swap_indexes_after_interleave.pop().unwrap());
                    spin_assignments_interleave.push(process_indexes_after_interleave_index.pop().unwrap());
                }
                for _ in 0..process_indexes_before_interleave_index.len() {
                    swap_assignments_interleave.push(swap_indexes_before_interleave.pop().unwrap());
                    spin_assignments_interleave.push(process_indexes_before_interleave_index.pop().unwrap());
                }
                for _ in 0..spin_hanging_off_front.len() {
                    swap_assignments_interleave.push(swap_hanging_off_front.pop().unwrap());
                    spin_assignments_interleave.push(spin_hanging_off_front.pop().unwrap());
                }
            } else {
                let mut swap_hanging_off_back = Vec::new();
                let mut spin_hanging_off_back = Vec::new();
                swap_indexes_before_interleave.reverse();
                process_indexes_before_interleave_index.reverse();
                for _ in 0..interleave_side_length_difference.abs() as usize {
                    swap_hanging_off_back.push(swap_indexes_before_interleave.pop().unwrap());
                    spin_hanging_off_back.push(process_indexes_before_interleave_index.pop().unwrap());
                }
                swap_indexes_before_interleave.reverse();
                process_indexes_before_interleave_index.reverse();
                for _ in 0..spin_hanging_off_back.len() {
                    swap_assignments_interleave.push(swap_hanging_off_back.pop().unwrap());
                    spin_assignments_interleave.push(spin_hanging_off_back.pop().unwrap());
                }
                for _ in 0..process_indexes_after_interleave_index.len() {
                    swap_assignments_interleave.push(swap_indexes_after_interleave.pop().unwrap());
                    spin_assignments_interleave.push(process_indexes_after_interleave_index.pop().unwrap());
                }
                for _ in 0..process_indexes_before_interleave_index.len() {
                    swap_assignments_interleave.push(swap_indexes_before_interleave.pop().unwrap());
                    spin_assignments_interleave.push(process_indexes_before_interleave_index.pop().unwrap());
                }
            }
            if num_processors != spin_assignments_interleave.len() || num_processors != swap_assignments_interleave.len(){
                panic!("swap_assignments_interleave.len() != num_processors");
            }
            let mut spun_interleave_generic_params_for_type = Vec::new();
            let mut swapped_interleave_generic_params_for_type = Vec::new();
            let mut spun_interleave_field_initializers = Vec::new();
            let mut swapped_interleave_field_initializes = Vec::new();
            for k in 0..num_processors {
                spun_interleave_generic_params_for_type.push(type_param_idents[spin_assignments_interleave[k]]);
                swapped_interleave_generic_params_for_type.push(type_param_idents[swap_assignments_interleave[k]]);
                let current_field_name = processor_field_idents[k];
                let spun_name = processor_field_idents[spin_assignments_interleave[k]];
                let swapped_name = processor_field_idents[swap_assignments_interleave[k]];
                spun_interleave_field_initializers.push(quote! { #current_field_name: self.#spun_name });
                swapped_interleave_field_initializes.push(quote! { #current_field_name: self.#swapped_name });
            }
            trait_impls.push(quote! {
                impl #impl_generics type_flow_traits::InterleavePivotSpin<#i> for #struct_name #ty_generics #where_clause {
                    type InterleavePivotSpinOutput = #struct_name<#(#spun_interleave_generic_params_for_type),*>;
                    fn interleaved_pivot_spin(self) -> Self::InterleavePivotSpinOutput {
                        #struct_name {
                            _marker: std::marker::PhantomData,
                            #( #spun_interleave_field_initializers ),*
                        }
                    }
                }
            });
            trait_impls.push(quote! {
                impl #impl_generics type_flow_traits::InterleavePivotSwap<#i> for #struct_name #ty_generics #where_clause {
                    type InterleavePivotSwapOutput = #struct_name<#(#swapped_interleave_generic_params_for_type),*>;
                    fn interleaved_pivot_swap(self) -> Self::InterleavePivotSwapOutput {
                        #struct_name {
                            _marker: std::marker::PhantomData,
                            #( #swapped_interleave_field_initializes ),*
                        }
                    }
                }
            });
        }
    }
    let expanded = quote! {
        #input_struct_item
        #(#trait_impls)*
    };
    proc_macro::TokenStream::from(expanded)
}

struct TypeFLowInplaceStatefulProcessorPipeline {
    pipeline_name: Ident,
    data_type: Type,
    error_type: Type,
    number_of_processors: usize,
}
impl Parse for TypeFLowInplaceStatefulProcessorPipeline {
    fn parse(input: ParseStream) -> syn::Result<Self> {
        let pipeline_name: Ident = input.parse()?;
        input.parse::<Token![,]>()?;
        let data_type: Type = input.parse()?;
        input.parse::<Token![,]>()?;
        let error_type: Type = input.parse()?;
        input.parse::<Token![,]>()?;
        let lit_int: syn::LitInt = input.parse()?;
        let number_of_processors = lit_int.base10_parse::<usize>()?;
        Ok(TypeFLowInplaceStatefulProcessorPipeline{
            pipeline_name,
            data_type,
            error_type,
            number_of_processors,
        })
    }
}

#[proc_macro]
pub fn type_flow_inplace_stateful_processor_pipeline_by_count(input: TokenStream) -> TokenStream {
    let TypeFLowInplaceStatefulProcessorPipeline { pipeline_name, data_type, error_type, number_of_processors } = parse_macro_input!(input as TypeFLowInplaceStatefulProcessorPipeline);
    let generic_params: Vec<_> = (0..number_of_processors)
        .map(|i| format_ident!("P{}", i))
        .collect();
    let field_names: Vec<_> = (0..number_of_processors)
        .map(|i| format_ident!("p{}", i))
        .collect();
    let fields_with_types: Vec<_> = field_names.iter().zip(generic_params.iter())
        .map(|(field, param)| quote! { #field: #param })
        .collect();
    let processor_args = fields_with_types.into_iter().reduce(|acc, item| {
        quote! { #acc, #item }
    });
    let expanded = quote! {
        type_flow_macros::type_flow_inplace_stateful_processor_pipeline!(
            #pipeline_name,
            #data_type,
            #error_type,
            #processor_args
        );
    };
    TokenStream::from(expanded)
}


#[proc_macro]
pub fn inplace_stateful_processor_pipeline_with_index(input: TokenStream) -> TokenStream { 
    let ProcessorInplacePipeline { pipeline_name, data_type, error_type, processors } = parse_macro_input!(input as ProcessorInplacePipeline);
    let struct_fields = processors.iter().enumerate().map(|(idx, processor_type)| {
        let type_str = get_type_name(processor_type);
        let field_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { #field_name: #processor_type }
    });
    let constructor_params = processors.iter().enumerate().map(|(idx, processor_type)| {
        let type_str = get_type_name(processor_type);
        let param_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { #param_name: #processor_type }
    });
    let field_initializers = processors.iter().enumerate().map(|(idx, processor_type)| {
        let type_str = get_type_name(processor_type);
        let field_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { #field_name }
    });
    let process_implementation = processors.iter().enumerate().map(|(idx, processor_type)| {
        let type_str = get_type_name(processor_type);
        let field_name = format_ident!("processor_{}_{}", type_str, idx);
        quote! { self.#field_name.process(data)?; }
    });
    let expanded = quote! {
        pub struct #pipeline_name {
            #(#struct_fields,)*
        }
        impl #pipeline_name {
            pub fn new(#(#constructor_params,)*) -> Self {
                Self {
                    #(#field_initializers,)*
                }
            }
        }
        impl crate::InPlaceStatefulProcessor<#data_type, #error_type> for #pipeline_name {
            fn process(&mut self, data: &mut #data_type) -> Result<(), #error_type> {
                #(#process_implementation)*
                Ok(())
            }
        }
    };
    TokenStream::from(expanded)
}

// Helper function to extract a usable identifier from a Type
fn get_type_name(ty: &syn::Type) -> String {
    match ty {
        syn::Type::Path(type_path) if !type_path.path.segments.is_empty() => {
            // Get the last segment of the path (e.g., for std::string::String, get "String")
            let segment = type_path.path.segments.last().unwrap();
            let name = segment.ident.to_string();

            // Filter out any characters that aren't valid in an identifier
            name.chars()
                .map(|c| if c.is_alphanumeric() { c } else { '_' })
                .collect()
        },
        // Handle other types (arrays, references, etc.) by using a generic name
        _ => "unknown_type".to_string(),
    }
}