/// given SmartModule invocation and context, generate execution context
given SmartModule invocation and context, generate execution context
given SmartModule invocation and context generate execution context
async fn extract_smartmodule_context(
        invocation: SmartModuleInvocation,
        version: i16,
        ctx: &DefaultSharedGlobalContext,
    ) -> Result<Self, ErrorCode> {
        // check for right consumer stream exists, this only happens for join type
        let right_consumer_stream = match invocation.kind {
            // for join, create consumer stream
            SmartModuleKind::Join(ref topic) => {
                let consumer = ctx.leaders().partition_consumer(topic.to_owned(), 0).await;
                Some(
                    consumer
                        .stream(fluvio::Offset::beginning())
                        .await
                        .map_err(|err| {
                            error!("error fetching join data {}", err);
                            ErrorCode::DerivedStreamJoinFetchError
                        })?
                        .boxed(),
                )
            }
            SmartModuleKind::JoinStream {
                topic: ref _topic,
                derivedstream: ref derivedstream_name,
            } => {
                // first ensure derivedstream exists
                if let Some(derivedstream) = ctx.derivedstream_store().spec(derivedstream_name) {
                    // find input which has topic
                    match derivedstream.spec.input {
                        DerivedStreamInputRef::Topic(topic) => {
                            let consumer = ctx
                                .leaders()
                                .partition_consumer(topic.name.to_owned(), 0)
                                .await;
                            // need to build stream arg
                            let mut builder = ConsumerConfig::builder();
                            builder.derivedstream(Some(DerivedStreamInvocation {
                                stream: derivedstream_name.to_owned(),
                                params: invocation.params.clone(),
                            }));
                            let consume_config = builder.build().map_err(|err| {
                                error!("error building consumer config {}", err);
                                ErrorCode::Other(format!("error building consumer config {}", err))
                            })?;
                            Some(
                                consumer
                                    .stream_with_config(fluvio::Offset::beginning(), consume_config)
                                    .await
                                    .map_err(|err| {
                                        error!("error fetching join data {}", err);
                                        ErrorCode::DerivedStreamJoinFetchError
                                    })?
                                    .boxed(),
                            )
                        }
                        DerivedStreamInputRef::DerivedStream(child_smart) => {
                            return Err(ErrorCode::DerivedStreamRecursion(
                                derivedstream_name.to_owned(),
                                child_smart.name,
                            ));
                        }
                    }
                } else {
                    return Err(ErrorCode::DerivedStreamNotFound(
                        derivedstream_name.to_owned(),
                    ));
                }
            }
            _ => None,
        };
        // then get smartmodule context
        let payload = match invocation.wasm {
            SmartModuleInvocationWasm::Predefined(name) => {
                if let Some(smartmodule) = ctx.smartmodule_localstore().spec(&name) {
                    let wasm = SmartModuleWasmCompressed::Gzip(smartmodule.wasm.payload);
                    LegacySmartModulePayload {
                        wasm,
                        kind: invocation.kind,
                        params: invocation.params,
                    }
                } else {
                    return Err(ErrorCode::SmartModuleNotFound { name });
                }
            }
            SmartModuleInvocationWasm::AdHoc(bytes) => {
                let wasm = SmartModuleWasmCompressed::Gzip(bytes);
                LegacySmartModulePayload {
                    wasm,
                    kind: invocation.kind,
                    params: invocation.params,
                }
            }
        };
        Ok(Self {
            smartmodule_instance: Self::payload_to_smartmodule(payload, version, ctx)?,
            right_consumer_stream,
        })
    }
async fn extract_smartmodule_context ( invocation : SmartModuleInvocation , version : i16 , ctx : & DefaultSharedGlobalContext , ) -> Result < Self , ErrorCode > { // check for right consumer stream exists, this only happens for join type let right_consumer_stream = match invocation . kind { // for join, create consumer stream SmartModuleKind :: Join ( ref topic ) => { let consumer = ctx . leaders ( ) . partition_consumer ( topic . to_owned ( ) , 0 ) . await ; Some ( consumer . stream ( fluvio :: Offset :: beginning ( ) ) . await . map_err ( | err | { error ! ( " " err ) ; ErrorCode :: DerivedStreamJoinFetchError } ) ? . boxed ( ) , ) } SmartModuleKind :: JoinStream { topic : ref _topic , derivedstream : ref derivedstream_name , } => { // first ensure derivedstream exists if let Some ( derivedstream ) = ctx . derivedstream_store ( ) . spec ( derivedstream_name ) { // find input which has topic match derivedstream . spec . input { DerivedStreamInputRef :: Topic ( topic ) => { let consumer = ctx . leaders ( ) . partition_consumer ( topic . name . to_owned ( ) , 0 ) . await ; // need to build stream arg let mut builder = ConsumerConfig :: builder ( ) ; builder . derivedstream ( Some ( DerivedStreamInvocation { stream : derivedstream_name . to_owned ( ) , params : invocation . params . clone ( ) , } ) ) ; let consume_config = builder . build ( ) . map_err ( | err | { error ! ( " " err ) ; ErrorCode :: Other ( format ! ( " " err ) ) } ) ? ; Some ( consumer . stream_with_config ( fluvio :: Offset :: beginning ( ) , consume_config ) . await . map_err ( | err | { error ! ( " " err ) ; ErrorCode :: DerivedStreamJoinFetchError } ) ? . boxed ( ) , ) } DerivedStreamInputRef :: DerivedStream ( child_smart ) => { return Err ( ErrorCode :: DerivedStreamRecursion ( derivedstream_name . to_owned ( ) , child_smart . name , ) ) ; } } } else { return Err ( ErrorCode :: DerivedStreamNotFound ( derivedstream_name . to_owned ( ) , ) ) ; } } _ => None , } ; // then get smartmodule context let payload = match invocation . wasm { SmartModuleInvocationWasm :: Predefined ( name ) => { if let Some ( smartmodule ) = ctx . smartmodule_localstore ( ) . spec ( & name ) { let wasm = SmartModuleWasmCompressed :: Gzip ( smartmodule . wasm . payload ) ; LegacySmartModulePayload { wasm , kind : invocation . kind , params : invocation . params , } } else { return Err ( ErrorCode :: SmartModuleNotFound { name } ) ; } } SmartModuleInvocationWasm :: AdHoc ( bytes ) => { let wasm = SmartModuleWasmCompressed :: Gzip ( bytes ) ; LegacySmartModulePayload { wasm , kind : invocation . kind , params : invocation . params , } } } ; Ok ( Self { smartmodule_instance : Self :: payload_to_smartmodule ( payload , version , ctx ) ? , right_consumer_stream , } ) }
given SmartModule invocation and context, generate execution context
given SmartModule invocation and context generate execution context
invocation
SmartModuleInvocation
version
i16
ctx
&DefaultSharedGlobalContext