fn upload(
        &mut self,
        ctx: RpcContext<'_>,
        stream: RequestStream<UploadRequest>,
        sink: ClientStreamingSink<UploadResponse>,
    ) {
        if !check_common_name(self.security_mgr.cert_allowed_cn(), &ctx) {
            return;
        }
        let label = "upload";
        let timer = Instant::now_coarse();
        let import = self.importer.clone();
        let (rx, buf_driver) =
            create_stream_with_buffer(stream.compat(), self.cfg.stream_channel_window);
        let mut rx = rx.map_err(Error::from);

        let handle_task = async move {
            let res = async move {
                let first_chunk = rx.try_next().await?;
                let meta = match first_chunk {
                    Some(ref chunk) if chunk.has_meta() => chunk.get_meta(),
                    _ => return Err(Error::InvalidChunk),
                };
                let file = import.create(meta)?;
                let mut file = rx
                    .try_fold(file, |mut file, chunk| async move {
                        let start = Instant::now_coarse();
                        let data = chunk.get_data();
                        if data.is_empty() {
                            return Err(Error::InvalidChunk);
                        }
                        file.append(data)?;
                        IMPORT_UPLOAD_CHUNK_BYTES.observe(data.len() as f64);
                        IMPORT_UPLOAD_CHUNK_DURATION.observe(start.elapsed_secs());
                        Ok(file)
                    })
                    .await?;
                file.finish().map(|_| UploadResponse::default())
            }
            .await;
            send_rpc_response!(res, sink, label, timer);
        };

        let thread_handle = self.threads.clone();
        let ctx_task = async move {
            thread_handle.spawn_ok(buf_driver);
            thread_handle.spawn_ok(handle_task);
        };
        ctx.spawn(Compat::new(ctx_task.unit_error().boxed()));
    }
/// Receive SST from client and save the file for later ingesting.
Receive SST from client and save the file for later ingesting.
Receive SST from client and save the file for later ingesting .
fn upload(
        &mut self,
        ctx: RpcContext<'_>,
        stream: RequestStream<UploadRequest>,
        sink: ClientStreamingSink<UploadResponse>,
    ) {
        if !check_common_name(self.security_mgr.cert_allowed_cn(), &ctx) {
            return;
        }
        let label = "upload";
        let timer = Instant::now_coarse();
        let import = self.importer.clone();
        let (rx, buf_driver) =
            create_stream_with_buffer(stream.compat(), self.cfg.stream_channel_window);
        let mut rx = rx.map_err(Error::from);
        let handle_task = async move {
            let res = async move {
                let first_chunk = rx.try_next().await?;
                let meta = match first_chunk {
                    Some(ref chunk) if chunk.has_meta() => chunk.get_meta(),
                    _ => return Err(Error::InvalidChunk),
                };
                let file = import.create(meta)?;
                let mut file = rx
                    .try_fold(file, |mut file, chunk| async move {
                        let start = Instant::now_coarse();
                        let data = chunk.get_data();
                        if data.is_empty() {
                            return Err(Error::InvalidChunk);
                        }
                        file.append(data)?;
                        IMPORT_UPLOAD_CHUNK_BYTES.observe(data.len() as f64);
                        IMPORT_UPLOAD_CHUNK_DURATION.observe(start.elapsed_secs());
                        Ok(file)
                    })
                    .await?;
                file.finish().map(|_| UploadResponse::default())
            }
            .await;
            send_rpc_response!(res, sink, label, timer);
        };
        let thread_handle = self.threads.clone();
        let ctx_task = async move {
            thread_handle.spawn_ok(buf_driver);
            thread_handle.spawn_ok(handle_task);
        };
        ctx.spawn(Compat::new(ctx_task.unit_error().boxed()));
    }
fn upload ( & mut self , ctx : RpcContext < ' _ > , stream : RequestStream < UploadRequest > , sink : ClientStreamingSink < UploadResponse > , ) { if ! check_common_name ( self . security_mgr . cert_allowed_cn ( ) , & ctx ) { return ; } let label = " " ; let timer = Instant :: now_coarse ( ) ; let import = self . importer . clone ( ) ; let ( rx , buf_driver ) = create_stream_with_buffer ( stream . compat ( ) , self . cfg . stream_channel_window ) ; let mut rx = rx . map_err ( Error :: from ) ; let handle_task = async move { let res = async move { let first_chunk = rx . try_next ( ) . await ? ; let meta = match first_chunk { Some ( ref chunk ) if chunk . has_meta ( ) => chunk . get_meta ( ) , _ => return Err ( Error :: InvalidChunk ) , } ; let file = import . create ( meta ) ? ; let mut file = rx . try_fold ( file , | mut file , chunk | async move { let start = Instant :: now_coarse ( ) ; let data = chunk . get_data ( ) ; if data . is_empty ( ) { return Err ( Error :: InvalidChunk ) ; } file . append ( data ) ? ; IMPORT_UPLOAD_CHUNK_BYTES . observe ( data . len ( ) as f64 ) ; IMPORT_UPLOAD_CHUNK_DURATION . observe ( start . elapsed_secs ( ) ) ; Ok ( file ) } ) . await ? ; file . finish ( ) . map ( | _ | UploadResponse :: default ( ) ) } . await ; send_rpc_response ! ( res sink label timer ) ; } ; let thread_handle = self . threads . clone ( ) ; let ctx_task = async move { thread_handle . spawn_ok ( buf_driver ) ; thread_handle . spawn_ok ( handle_task ) ; } ; ctx . spawn ( Compat :: new ( ctx_task . unit_error ( ) . boxed ( ) ) ) ; }
Receive SST from client and save the file for later ingesting.
Receive SST from client and save the file for later ingesting .
&mut self
ctx
RpcContext<'_>
stream
RequestStream<UploadRequest>
sink
ClientStreamingSink<UploadResponse>