fn upload(
        &self,
        ctx: RpcContext,
        stream: RequestStream<UploadRequest>,
        sink: ClientStreamingSink<UploadResponse>,
    ) {
        let label = "upload";
        let timer = Instant::now_coarse();
        let import = Arc::clone(&self.importer);
        let bounded_stream = mpsc::spawn(stream, &self.threads, self.cfg.stream_channel_window);

        ctx.spawn(
            self.threads.spawn(
                bounded_stream
                    .into_future()
                    .map_err(|(e, _)| Error::from(e))
                    .and_then(move |(chunk, stream)| {
                        // The first message of the stream contains metadata
                        // of the file.
                        let meta = match chunk {
                            Some(ref chunk) if chunk.has_meta() => chunk.get_meta(),
                            _ => return Err(Error::InvalidChunk),
                        };
                        let file = import.create(meta)?;
                        Ok((file, stream))
                    })
                    .and_then(move |(file, stream)| {
                        stream
                            .map_err(Error::from)
                            .fold(file, |mut file, chunk| {
                                let start = Instant::now_coarse();
                                let data = chunk.get_data();
                                if data.is_empty() {
                                    return future::err(Error::InvalidChunk);
                                }
                                if let Err(e) = file.append(data) {
                                    return future::err(e);
                                }
                                IMPORT_UPLOAD_CHUNK_BYTES.observe(data.len() as f64);
                                IMPORT_UPLOAD_CHUNK_DURATION.observe(start.elapsed_secs());
                                future::ok(file)
                            })
                            .and_then(|mut file| file.finish())
                    })
                    .map(|_| UploadResponse::new())
                    .then(move |res| send_rpc_response!(res, sink, label, timer)),
            ),
        )
    }
/// 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(
        &self,
        ctx: RpcContext,
        stream: RequestStream<UploadRequest>,
        sink: ClientStreamingSink<UploadResponse>,
    ) {
        let label = "upload";
        let timer = Instant::now_coarse();
        let import = Arc::clone(&self.importer);
        let bounded_stream = mpsc::spawn(stream, &self.threads, self.cfg.stream_channel_window);
        ctx.spawn(
            self.threads.spawn(
                bounded_stream
                    .into_future()
                    .map_err(|(e, _)| Error::from(e))
                    .and_then(move |(chunk, stream)| {
                        // The first message of the stream contains metadata
                        // of the file.
                        let meta = match chunk {
                            Some(ref chunk) if chunk.has_meta() => chunk.get_meta(),
                            _ => return Err(Error::InvalidChunk),
                        };
                        let file = import.create(meta)?;
                        Ok((file, stream))
                    })
                    .and_then(move |(file, stream)| {
                        stream
                            .map_err(Error::from)
                            .fold(file, |mut file, chunk| {
                                let start = Instant::now_coarse();
                                let data = chunk.get_data();
                                if data.is_empty() {
                                    return future::err(Error::InvalidChunk);
                                }
                                if let Err(e) = file.append(data) {
                                    return future::err(e);
                                }
                                IMPORT_UPLOAD_CHUNK_BYTES.observe(data.len() as f64);
                                IMPORT_UPLOAD_CHUNK_DURATION.observe(start.elapsed_secs());
                                future::ok(file)
                            })
                            .and_then(|mut file| file.finish())
                    })
                    .map(|_| UploadResponse::new())
                    .then(move |res| send_rpc_response!(res, sink, label, timer)),
            ),
        )
    }
fn upload ( & self , ctx : RpcContext , stream : RequestStream < UploadRequest > , sink : ClientStreamingSink < UploadResponse > , ) { let label = " " ; let timer = Instant :: now_coarse ( ) ; let import = Arc :: clone ( & self . importer ) ; let bounded_stream = mpsc :: spawn ( stream , & self . threads , self . cfg . stream_channel_window ) ; ctx . spawn ( self . threads . spawn ( bounded_stream . into_future ( ) . map_err ( | ( e , _ ) | Error :: from ( e ) ) . and_then ( move | ( chunk , stream ) | { // The first message of the stream contains metadata // of the file. let meta = match chunk { Some ( ref chunk ) if chunk . has_meta ( ) => chunk . get_meta ( ) , _ => return Err ( Error :: InvalidChunk ) , } ; let file = import . create ( meta ) ? ; Ok ( ( file , stream ) ) } ) . and_then ( move | ( file , stream ) | { stream . map_err ( Error :: from ) . fold ( file , | mut file , chunk | { let start = Instant :: now_coarse ( ) ; let data = chunk . get_data ( ) ; if data . is_empty ( ) { return future :: err ( Error :: InvalidChunk ) ; } if let Err ( e ) = file . append ( data ) { return future :: err ( e ) ; } IMPORT_UPLOAD_CHUNK_BYTES . observe ( data . len ( ) as f64 ) ; IMPORT_UPLOAD_CHUNK_DURATION . observe ( start . elapsed_secs ( ) ) ; future :: ok ( file ) } ) . and_then ( | mut file | file . finish ( ) ) } ) . map ( | _ | UploadResponse :: new ( ) ) . then ( move | res | send_rpc_response ! ( res sink label timer ) ) , ) , ) }
Receive SST from client and save the file for later ingesting.
Receive SST from client and save the file for later ingesting .
&self
ctx
RpcContext
stream
RequestStream<UploadRequest>
sink
ClientStreamingSink<UploadResponse>