graph_fs 0.1.5

A CLI tool that helps query a local or remote file system by a GraphQL interface
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use std::io::Error;
use std::path::Path;

use crate::Session;
use actix_web::web;
use actix_web::HttpResponse;
use async_stream::__private::AsyncStream;
use async_stream::try_stream;
use bytes::Bytes;
use std::io::Write;
use uuid::Uuid;

use actix_multipart::Multipart;
use futures::{StreamExt, TryStreamExt};
// Local util
pub async fn save_local_file(
    mut payload: Multipart,
    file_path: &Path,
) -> Result<Option<(bool, String)>, Error> {
    let mut upload_path: String = "".to_string();

    // Iterate over multipart stream
    while let Ok(Some(mut field)) = payload.try_next().await {
        let content_type = field.content_disposition();
        let filename = content_type.get_filename().unwrap_or("");
        let uid = Uuid::new_v4().to_string();
        let new_name = format!("{}-{}", uid, filename);
        let filepath = file_path.join(new_name);
        upload_path = filepath.to_str().unwrap().to_string();

        // File::create is blocking operation, use threadpool
        let mut f = web::block(|| std::fs::File::create(filepath))
            .await
            .unwrap()?;

        // Field in turn is stream of *Bytes* object
        while let Some(chunk) = field.next().await {
            let data = chunk.unwrap();
            // Filesystem operations are blocking, we have to use threadpool
            f = web::block(move || f.write_all(&data).map(|_| f))
                .await
                .unwrap()?;
        }
    }
    println!("file {}", upload_path);

    Ok(Some((true, upload_path)))
}

pub async fn save_remote_file(
    mut payload: Multipart,
    sessiopn: &Session,
    file_path: &Path,
) -> Result<Option<(bool, String)>, Error> {
    let mut return_write: Vec<Vec<u8>> = vec![];
    let mut file_name = String::new();

    while let Ok(Some(mut field)) = payload.try_next().await {
        let content_type = field.content_disposition();
        let filename = content_type.get_filename().unwrap();
        let uid = Uuid::new_v4().to_string();
        let new_name = format!("{}-{}", uid, filename);
        file_name = file_path.join(new_name).to_str().unwrap().to_string();

        // Field in turn is stream of *Bytes* object
        while let Some(chunk) = field.next().await {
            return_write.push(chunk.unwrap()[..].to_vec());
        }
    }
    let flattened = return_write.into_iter().flatten().collect::<Vec<u8>>();
    let mut remote_file =
        sessiopn.scp_send(Path::new(&file_name), 0o644, flattened.len() as u64, None)?;

    // Write buffer to remote FS
    remote_file.write_all(&flattened)?;
    remote_file.send_eof()?;
    remote_file.wait_eof()?;
    remote_file.close()?;
    remote_file.wait_close()?;
    Ok(Some((true, file_name)))
}

// Stream buffer to client
pub fn buffer_response(mut x: Vec<u8>) -> HttpResponse {
    // If buffer is less than 4kb, just return buffer
    if x.len() <= 4096 {
        HttpResponse::Ok().body(x)
    } else {
        let stream: AsyncStream<Result<Bytes, Error>, _> = try_stream! {
            // Stream large buffer files
            loop{
                if x.len()>4096{
                    let u:Vec<u8>=x.drain(0..4096).collect();

                    yield Bytes::copy_from_slice(&u[..4096]);
                }else{
                    let u:Vec<u8>=std::mem::take(&mut x);
                    if u.is_empty(){
                        break
                    }
                    yield Bytes::copy_from_slice(&u[..u.len()]);

                }
            }
        };
        HttpResponse::Ok().streaming(stream)
    }
}