~qaul/community

Service/files patch v1 PROPOSED

Dear team,

Sending a patch your way for WIP so far. Dont be too excited.
This is mostly a copypaste of stuff from chat. Nothing original
to see here.

happy-hacking++ to you;

-aj

Amanjeev Sethi (1):
  service/files: wip, release patch so far see squashed commits from
    below.

 Cargo.lock                            |  4 ++
 libqaul/service/files/Cargo.toml      |  4 ++
 libqaul/service/files/src/lib.rs      | 69 +++++++++++++++++++++++----
 libqaul/service/files/src/msg.rs      | 13 +++++
 libqaul/service/files/src/protocol.rs | 54 +++++++++++++++++++++
 libqaul/service/files/src/types.rs    | 48 ++++++-------------
 libqaul/service/files/src/worker.rs   | 61 +++++++++++++++++++++++
 7 files changed, 210 insertions(+), 43 deletions(-)
 create mode 100644 libqaul/service/files/src/msg.rs
 create mode 100644 libqaul/service/files/src/protocol.rs
 create mode 100644 libqaul/service/files/src/worker.rs

-- 
2.23.1
Export patchset (mbox)
How do I use this?

Copy & paste the following snippet into your terminal to import this patchset into git:

curl -s https://lists.sr.ht/~qaul/community/patches/10818/mbox | git am -3
Learn more about email & git
View this thread in the archives

[PATCH 1/1] service/files: wip, release patch so far see squashed commits from below. Export this patch

service/files: wip, adds two methods to Fileshare struct

service/files: wip, Fileshare service accepts advertised files

service/files: wip, moves impls to protocol module

service/files: wip, adds fn advertise_to_user

service/files: wip, hacky copy-paste stuff from chat service but compiles

service/files: wip, adds TODOs for later

service/files: wip, TODOs comment
---
 Cargo.lock                            |  4 ++
 libqaul/service/files/Cargo.toml      |  4 ++
 libqaul/service/files/src/lib.rs      | 69 +++++++++++++++++++++++----
 libqaul/service/files/src/msg.rs      | 13 +++++
 libqaul/service/files/src/protocol.rs | 54 +++++++++++++++++++++
 libqaul/service/files/src/types.rs    | 48 ++++++-------------
 libqaul/service/files/src/worker.rs   | 61 +++++++++++++++++++++++
 7 files changed, 210 insertions(+), 43 deletions(-)
 create mode 100644 libqaul/service/files/src/msg.rs
 create mode 100644 libqaul/service/files/src/protocol.rs
 create mode 100644 libqaul/service/files/src/worker.rs

diff --git a/Cargo.lock b/Cargo.lock
index 319ea295..d3d6579e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1877,10 +1877,14 @@ dependencies = [
name = "qaul-files"
version = "0.1.0"
dependencies = [
 "async-std",
 "libqaul",
 "mime",
 "mime_guess",
 "ratman-identity",
 "serde",
 "tracing",
 "tracing-futures",
]

[[package]]
diff --git a/libqaul/service/files/Cargo.toml b/libqaul/service/files/Cargo.toml
index b45944ce..88d28556 100644
--- a/libqaul/service/files/Cargo.toml
+++ b/libqaul/service/files/Cargo.toml
@@ -9,4 +9,8 @@ license = "AGPL-3.0"
identity = { path = "../../../ratman/identity", package = "ratman-identity" }
libqaul = { path = "../../" }
mime = "0.3"
mime_guess = "2.0.1"
serde = { version = "1.0" }
async-std = "=1.5"
tracing = "0.1"
tracing-futures = "0.2"
diff --git a/libqaul/service/files/src/lib.rs b/libqaul/service/files/src/lib.rs
index 3a801e83..8c0b89ff 100644
--- a/libqaul/service/files/src/lib.rs
+++ b/libqaul/service/files/src/lib.rs
@@ -1,36 +1,85 @@
//! `qaul.net` filesharing service

use std::sync::Arc;
use async_std::{sync::Arc, task};
use mime::Mime;

use libqaul::messages::{Message, MsgQuery};
use libqaul::users::UserAuth;
use libqaul::Identity;
use libqaul::{error::Result, Qaul};
use libqaul::services::ServiceEvent;

pub use crate::types::{File, FileFilter, FileId, Subscription, Files};

mod msg;
mod protocol;
pub mod types;
mod worker;

// these are original TODOs
// TODO: Partial files
// TODO: file progress
// TODO: Download links with tokens

const ASC_NAME: &'static str = "net.qaul.filesharing";
const ASC_NAME: &'static str = "net.qaul.fileshare";

/// Filesharing service state
#[derive(Clone)]
pub struct Filesharing {
pub struct Fileshare {
    qaul: Arc<Qaul>,
    advertised: Arc<Vec<FileId>>,
}

impl Filesharing {
    /// Initialise the filesharing service
impl Fileshare {
    /// Initialise the file-sharing service
    ///
    /// In order to initialise, a valid and running
    /// `Qaul` reference needs to be provided.
    pub fn new(qaul: Arc<Qaul>) -> Result<Self> {
        Ok(Self { qaul })
    pub fn new(qaul: Arc<Qaul>, advertised: Arc<Vec<FileId>>) -> Result<Arc<Self>> {
        let this = Arc::new(Self {qaul, advertised});
        let sender = Arc::new(worker::run_asnc(Arc::clone(&this)));

        this.qaul.services().register(ASC_NAME, move |cmd| {
            let sender = Arc::clone(&sender);
            task::block_on(async move {
                match cmd {
                    ServiceEvent::Open(auth) => sender.send(worker::Command::Start(auth)).await,
                    ServiceEvent::Close(auth) => sender.send(worker::Command::Stop(auth)).await,
                }
            });
        });

        Ok(this)
    }

    /// Advertise a file into a network
    pub fn advertise(
        &self,
        file_name: String,
        file_id: FileId,
        file_size: usize,
        file_type: Mime,
    ) -> Result<Arc<Vec<FileId>>> {
        // TODO: Check if ok that it returns the `advertised` vector

        unimplemented!()
    }

    //     /// Announce a new file into a network
    //     pub fn announce<S>(&self, name: S) -> Result<FileId> {
    // Advertise to a single user
    pub fn advertise_to_user(
        &self,
        file_name: String,
        file_size: usize,
        file_type: Mime,
        send_to: UserAuth,
    ) -> Result<Arc<Vec<FileId>>> {
        unimplemented!()
    }

    //     }
    /// Request a file with a given file id
    pub fn request(&self, file_id: FileId) -> Result<File> {
        unimplemented!()
    }
}

// impl<'q> Filesharing<'q> {
diff --git a/libqaul/service/files/src/msg.rs b/libqaul/service/files/src/msg.rs
new file mode 100644
index 00000000..c906f508
--- /dev/null
+++ b/libqaul/service/files/src/msg.rs
@@ -0,0 +1,13 @@
use libqaul::messages::Message;
use crate::File;

impl From<Message> for File {
    fn from(msg: Message) -> Self {
        Self {
            name: Some(msg.id.to_string()), // TODO: how to get name here?
            id: msg.id,
            data: Some(msg.payload),
            owner: msg.sender,
        }
    }
}
\ No newline at end of file
diff --git a/libqaul/service/files/src/protocol.rs b/libqaul/service/files/src/protocol.rs
new file mode 100644
index 00000000..de40e836
--- /dev/null
+++ b/libqaul/service/files/src/protocol.rs
@@ -0,0 +1,54 @@
//! The file protocol implementation
//!
//! Underlying types used are defined in `types.rs`, interactions are
//! defined here for clarity.  Following is a textual explanation of
//! the dynamics of the protocol, what parts are implemented here, and
//! what parts are implemented via libqaul.

use libqaul::error::Result;
use libqaul::Identity;
use libqaul::users::UserAuth;

use crate::types::File;
use crate::types::FileFilter;
use crate::types::FileId;
use crate::types::FileMeta;
use crate::types::Files;

impl<'qaul> Files<'qaul> {
    /// Query the local file store for a specific constraint
    pub fn query<I>(&self, user: UserAuth, filter: FileFilter) -> Result<I>
        where
            I: Iterator<Item=FileMeta>,
    {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }

    /// List all available files
    pub fn list<I>(&self, user: UserAuth) -> Result<I>
        where
            I: Iterator<Item=FileMeta>,
    {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }

    /// Stream one particular file from storage
    pub async fn get(&self, user: UserAuth, file: FileId) -> Result<File> {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }

    /// Adds a new file to the local user's storage
    pub fn add(&self, user: UserAuth, name: &str, file: File) -> Result<FileId> {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }

    /// Delete a file from the local user store
    pub fn delete(&self, user: UserAuth, name: FileId) -> Result<()> {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }
}
diff --git a/libqaul/service/files/src/types.rs b/libqaul/service/files/src/types.rs
index 4ffb8e93..8350460d 100644
--- a/libqaul/service/files/src/types.rs
+++ b/libqaul/service/files/src/types.rs
@@ -3,13 +3,17 @@ use serde::{Deserialize, Serialize};
use libqaul::error::Result;
use libqaul::Identity;
use libqaul::users::UserAuth;
use libqaul::messages::{Message, MsgId};
use libqaul::helpers::{Subscription as Sub};

pub type FileId = Identity;

/// Local file abstraction
pub struct File {
    pub name: String,
    pub name: Option<String>,
    pub id: FileId,
    pub data: Option<Vec<u8>>,
    pub owner: Identity,
}

/// Describe a file's lifecycle
@@ -70,40 +74,18 @@ pub struct Files<'chain> {
    pub(crate) q: &'chain crate::Qaul,
}

impl<'qaul> Files<'qaul> {
    /// Query the local file store for a specific constraint
    pub fn query<I>(&self, user: UserAuth, filter: FileFilter) -> Result<I>
        where
            I: Iterator<Item=FileMeta>,
    {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }

    /// List all available files
    pub fn list<I>(&self, user: UserAuth) -> Result<I>
        where
            I: Iterator<Item=FileMeta>,
    {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }

    /// Stream one particular file from storage
    pub async fn get(&self, user: UserAuth, file: FileId) -> Result<File> {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    }
/// A subscription handler that pushes out updates
pub struct Subscription {
    pub(crate) inner: Sub<Message>,
}

    /// Adds a new file to the local user's storage
    pub fn add(&self, user: UserAuth, name: &str, file: File) -> Result<FileId> {
        // self.q.auth.trusted(user)?;
        unimplemented!()
impl Subscription {
    pub(crate) fn new(inner: Sub<Message>) -> Self {
        Self { inner }
    }

    /// Delete a file from the local user store
    pub fn delete(&self, user: UserAuth, name: FileId) -> Result<()> {
        // self.q.auth.trusted(user)?;
        unimplemented!()
    /// Get the next chat message
    pub async fn next(&self) -> File {
        self.inner.next().await.into()
    }
}
diff --git a/libqaul/service/files/src/worker.rs b/libqaul/service/files/src/worker.rs
new file mode 100644
index 00000000..7752cd12
--- /dev/null
+++ b/libqaul/service/files/src/worker.rs
@@ -0,0 +1,61 @@
use crate::{ASC_NAME, Fileshare, Subscription};
use libqaul::{users::UserAuth, Identity, helpers::TagSet};
use async_std::{
    sync::{channel, Arc, RwLock, Sender},
    task,
};
use std::collections::BTreeSet;
use tracing::{debug, info, trace};

pub(crate) enum Command {
    Start(UserAuth),
    Stop(UserAuth),
}

type RunMap = Arc<RwLock<BTreeSet<Identity>>>;

pub(crate) fn run_asnc(file_serv: Arc<Fileshare>) -> Sender<Command> {
    let (tx, rx) = channel(1);

    task::spawn(async move {
        let map: RunMap = Default::default();
        while let Some(cmd) = rx.recv().await {
            let map = Arc::clone(&map);
            match cmd {
                Command::Start(auth) => {
                    trace!("Receiving libqaul user {} START event!", auth.0);
                    map.write().await.insert(auth.0);
                    task::spawn(run_user(auth, Arc::clone(&file_serv), Arc::clone(&map)));
                }
                Command::Stop(auth) => {
                    trace!("Receiving libqaul user {} STOP event!", auth.0);
                    map.write().await.remove(&auth.0);
                }
            }
        }

        // Stop all remaining workers
        info!("Deallocating subscription workers");
        map.write().await.clear();
    });

    tx
}

pub(crate) async fn run_user(user: UserAuth, file_serv: Arc<Fileshare>, run: RunMap) {
    let sub = Subscription::new(
        file_serv.qaul
            .messages()
            .subscribe(user.clone(), ASC_NAME, TagSet::empty())
            .await
            .unwrap(),
    );
    trace!("Creating message subscription!");

    while run.read().await.contains(&user.0) {
        let file = sub.next().await;
        if file.owner == user.0 && continue {}
    }

    // TODO: what the hell should this do?
}
-- 
2.23.1