Merge branch 'parallel_medias' into 'main'

feat: async upload of medias

See merge request veretcle/oolatoocs!3

Cargo.lock

@@ -444,19 +444,46 @@ dependencies = [
 ]
 
 [[package]]
-name = "futures-channel"
+name = "futures"
-version = "0.3.28"
+version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "955518d47e09b25bbebc7a18df10b81f0c766eaf4c4f1cccef2fca5f2a4fb5f2"
+checksum = "da0290714b38af9b4a7b094b8a37086d1b4e61f2df9122c3cad2577669145335"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-channel"
+version = "0.3.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff4dd66668b557604244583e3e1e1eada8c5c2e96a6d0d6653ede395b78bbacb"
 dependencies = [
  "futures-core",
+ "futures-sink",
 ]
 
 [[package]]
 name = "futures-core"
-version = "0.3.28"
+version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bca583b7e26f571124fe5b7561d49cb2868d79116cfa0eefce955557c6fee8c"
+checksum = "eb1d22c66e66d9d72e1758f0bd7d4fd0bee04cad842ee34587d68c07e45d088c"
+
+[[package]]
+name = "futures-executor"
+version = "0.3.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f4fb8693db0cf099eadcca0efe2a5a22e4550f98ed16aba6c48700da29597bc"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
 
 [[package]]
 name = "futures-io"
@@ -466,9 +493,9 @@ checksum = "8bf34a163b5c4c52d0478a4d757da8fb65cabef42ba90515efee0f6f9fa45aaa"
 
 [[package]]
 name = "futures-macro"
-version = "0.3.28"
+version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89ca545a94061b6365f2c7355b4b32bd20df3ff95f02da9329b34ccc3bd6ee72"
+checksum = "53b153fd91e4b0147f4aced87be237c98248656bb01050b96bf3ee89220a8ddb"
 dependencies = [
  "proc-macro2",
  "quote 1.0.33",
@@ -477,22 +504,23 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.28"
+version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f43be4fe21a13b9781a69afa4985b0f6ee0e1afab2c6f454a8cf30e2b2237b6e"
+checksum = "e36d3378ee38c2a36ad710c5d30c2911d752cb941c00c72dbabfb786a7970817"
 
 [[package]]
 name = "futures-task"
-version = "0.3.28"
+version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "76d3d132be6c0e6aa1534069c705a74a5997a356c0dc2f86a47765e5617c5b65"
+checksum = "efd193069b0ddadc69c46389b740bbccdd97203899b48d09c5f7969591d6bae2"
 
 [[package]]
 name = "futures-util"
-version = "0.3.28"
+version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26b01e40b772d54cf6c6d721c1d1abd0647a0106a12ecaa1c186273392a69533"
+checksum = "a19526d624e703a3179b3d322efec918b6246ea0fa51d41124525f00f1cc8104"
 dependencies = [
+ "futures-channel",
  "futures-core",
  "futures-io",
  "futures-macro",
@@ -1058,11 +1086,12 @@ checksum = "dd8b5dd2ae5ed71462c540258bedcb51965123ad7e7ccf4b9a8cafaa4a63576d"
 
 [[package]]
 name = "oolatoocs"
-version = "1.0.0"
+version = "1.1.0"
 dependencies = [
  "clap",
  "dissolve",
  "env_logger",
+ "futures",
  "log 0.4.20",
  "megalodon",
  "oauth1-request",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "oolatoocs"
-version = "1.0.0"
+version = "1.1.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -9,6 +9,7 @@ edition = "2021"
 clap = "^4"
 dissolve = "0.2.2"
 env_logger = "^0.10"
+futures = "0.3.29"
 log = "^0.4"
 megalodon = "^0.11"
 oauth1-request = "^0.6"

src/config.rs

@@ -8,7 +8,7 @@ pub struct Config {
     pub twitter: TwitterConfig,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct TwitterConfig {
     pub consumer_key: String,
     pub consumer_secret: String,

src/lib.rs

@@ -20,8 +20,11 @@ mod twitter;
 #[allow(unused_imports)]
 use twitter::{post_tweet, upload_chunk_media, upload_simple_media};
 
+use futures::{stream, StreamExt};
+use log::{error, warn};
 use megalodon::entities::attachment::AttachmentType;
 use rusqlite::Connection;
+use std::error::Error;
 
 #[tokio::main]
 pub async fn run(config: &Config) {
@@ -40,42 +43,45 @@ pub async fn run(config: &Config) {
         let Ok(tweet_content) = strip_everything(&toot.content, &toot.tags) else {
             continue; // skip in case we can’t strip something
         };
+
         let mut medias: Vec<u64> = vec![];
 
         // if we wanted to cut toot in half, now would be the right time to do so
 
-        // treats media
+        let media_attachments = toot.media_attachments.clone();
-        for media in toot.media_attachments {
+        let mut stream = stream::iter(media_attachments)
-            let id = match media.r#type {
+            .map(|media| {
-                AttachmentType::Image => {
+                let twitter_config = config.twitter.clone();
-                    let Ok(id) =
+                tokio::task::spawn(async move {
-                        upload_simple_media(&config.twitter, &media.url, &media.description).await
+                    match media.r#type {
-                    else {
+                        AttachmentType::Image => {
-                        continue;
+                            upload_simple_media(&twitter_config, &media.url, &media.description)
-                    };
+                                .await
-                    id
+                        }
-                }
+                        AttachmentType::Gifv => {
-                AttachmentType::Gifv => {
+                            upload_chunk_media(&twitter_config, &media.url, "tweet_gif").await
-                    let Ok(id) = upload_chunk_media(&config.twitter, &media.url, "tweet_gif").await
+                        }
-                    else {
+                        AttachmentType::Video => {
-                        continue;
+                            upload_chunk_media(&twitter_config, &media.url, "tweet_video").await
-                    };
+                        }
-                    id
+                        _ => Err::<u64, Box<dyn Error + Send + Sync>>(
-                }
+                            OolatoocsError::new(&format!(
-                AttachmentType::Video => {
+                                "Cannot treat this type of media: {}",
-                    let Ok(id) =
+                                &media.url
-                        upload_chunk_media(&config.twitter, &media.url, "tweet_video").await
+                            ))
-                    else {
+                            .into(),
-                        continue;
+                        ),
-                    };
+                    }
-                    id
+                })
-                }
+            })
-                _ => {
+            .buffered(4);
-                    continue;
-                }
-            };
 
-            medias.push(id);
+        while let Some(result) = stream.next().await {
+            match result {
+                Ok(Ok(v)) => medias.push(v),
+                Ok(Err(e)) => warn!("Cannot treat media: {}", e),
+                Err(e) => error!("Something went wrong when joining the main thread: {}", e),
+            }
         }
 
         // threads if necessary

src/twitter.rs

@@ -104,7 +104,7 @@ pub async fn upload_simple_media(
     config: &TwitterConfig,
     u: &str,
     d: &Option<String>,
-) -> Result<u64, Box<dyn Error>> {
+) -> Result<u64, Box<dyn Error + Send + Sync>> {
     // initiate request parameters
     let empty_request = EmptyRequest {}; // Why? Because fuck you, that’s why!
     let token = get_token(config);
@@ -152,7 +152,11 @@ pub async fn upload_simple_media(
 }
 
 /// This function updates the metadata given the current media_id and token
-async fn metadata_create(config: &TwitterConfig, id: u64, m: &str) -> Result<(), Box<dyn Error>> {
+async fn metadata_create(
+    config: &TwitterConfig,
+    id: u64,
+    m: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
     let token = get_token(config);
     let empty_request = EmptyRequest {};
 
@@ -191,7 +195,7 @@ pub async fn upload_chunk_media(
     config: &TwitterConfig,
     u: &str,
     t: &str,
-) -> Result<u64, Box<dyn Error>> {
+) -> Result<u64, Box<dyn Error + Send + Sync>> {
     let empty_request = EmptyRequest {};
     let token = get_token(config);