如何在Rust中直接从curl读取缓冲行

huangapple
go评论98阅读模式
英文:

How to read buffered lines directly from curl in Rust

问题

我正在使用curl库来下载一个经过gzip压缩的文件：

  1. use curl::easy::Easy;
  3. fn download_file() -> MyResult<()> {
  4.     let mut file = File::create("file.txt.gz")?;
  5.     let mut easy = Easy::new();
  6.     easy.url("https://example.com/file.txt.gz")?;
  7.     easy.write_function(move |data| {
  8.         file.write_all(data).unwrap();
  9.         return Ok(data.len());
  10.     })?;
  11.     easy.perform()?;
  12.     return Ok(());
  13. }

我正在使用flate2库来解压文件，并逐行读取：

  1. use flate2::read::GzDecoder;
  3. fn import_file() -> MyResult<()> {
  4.     let file = File::open("file.txt.gz")?;
  5.     let reader = BufReader::new(GzDecoder::new(file));
  6.     for line in reader.lines() {
  7.         let line = line?;
  8.         println!("{}", line);
  9.     }
  10.     return Ok(());
  11. }

我想要将这些步骤合并，但是GzDecoder需要一个实现Read trait的对象，我不确定如何将其连接到Easy。我怀疑这需要在单独的线程上执行下载和导入步骤：

  1. use curl::easy::Easy;
  2. use flate2::read::GzDecoder;
  4. fn download_and_import_file() -> MyResult<()> {
  5.     let handle = thread::spawn(|| {
  6.         let mut easy = Easy::new();
  7.         easy.url("https://example.com/file.txt.gz").unwrap();
  8.         easy.write_function(move |data| {
  9.             // （这里应该放什么？）
  10.             return Ok(data.len());
  11.         }).unwrap();
  12.         easy.perform().unwrap();
  13.     });
  14.     let reader = // （这里应该放什么？）
  15.     let reader = BufReader::new(GzDecoder::new(reader));
  16.     for line in reader.lines() {
  17.         let line = line?;
  18.         println!("{}", line);
  19.     }
  20.     handle.join().unwrap();
  21.     return Ok(());
  22. }
英文:

I'm using the curl crate to download a gzipped file:

  1. use curl::easy::Easy;
  3. fn download_file() -&gt; MyResult&lt;()&gt; {
  4.     let mut file = File::create(&quot;file.txt.gz&quot;)?;
  5.     let mut easy = Easy::new();
  6.     easy.url(&quot;https://example.com/file.txt.gz&quot;)?;
  7.     easy.write_function(move |data| {
  8.         file.write_all(data).unwrap();
  9.         return Ok(data.len());
  10.     })?;
  11.     easy.perform()?;
  12.     return Ok(());
  13. }

I'm using the flate2 crate to decompress the file, and read one line at a time:

  1. use flate2::read::GzDecoder;
  3. fn import_file() -&gt; MyResult&lt;()&gt; {
  4.     let file = File::open(&quot;file.txt.gz&quot;)?;
  5.     let reader = BufReader::new(GzDecoder::new(file));
  6.     for line in reader.lines() {
  7.         let line = line?;
  8.         println!(&quot;{line}&quot;);
  9.     }
  10.     return Ok(());
  11. }

I would like to combine these steps, but GzDecoder requires an object implementing the Read trait, and I'm not sure how to hook that up to Easy. I suspect this will require doing the download and import steps on separate threads:

  1. use curl::easy::Easy;
  2. use flate2::read::GzDecoder;
  4. fn download_and_import_file() -&gt; MyResult&lt;()&gt; {
  5.     let handle = thread::spawn(|| {
  6.         let mut easy = Easy::new();
  7.         easy.url(&quot;https://example.com/file.txt.gz&quot;).unwrap();
  8.         easy.write_function(move |data| {
  9.             // (what goes here?)
  10.             return Ok(data.len());
  11.         }).unwrap();
  12.         easy.perform().unwrap();
  13.     });
  14.     let reader = // (what goes here?)
  15.     let reader = BufReader::new(GzDecoder::new(reader));
  16.     for line in reader.lines() {
  17.         let line = line?;
  18.         println!(&quot;{line}&quot;);
  19.     }
  20.     handle.join().unwrap();
  21.     return Ok(());
  22. }

答案1

得分: 2

以下是您要求的代码部分的翻译：

  1. use std::collections::VecDeque;
  2. use std::io::{self, Read};
  3. use std::sync::atomic::{AtomicBool, Ordering};
  4. use std::sync::{Arc, Condvar, Mutex};
  6. #[derive(Clone, Default)]
  7. pub struct InMemoryStream {
  8.     inner: Arc<InMemoryStreamInner>,
  9. }
  11. #[derive(Default)]
  12. struct InMemoryStreamInner {
  13.     data: Mutex<VecDeque<u8>>,
  14.     condvar: Condvar,
  15.     closed: AtomicBool,
  16. }
  18. impl InMemoryStream {
  19.     pub fn write(&self, data: &[u8]) {
  20.         self.inner.data.lock().unwrap().extend(data);
  21.         self.inner.condvar.notify_one();
  22.     }
  24.     pub fn close(&self, data: &[u8]) {
  25.         self.inner.closed.store(true, Ordering::SeqCst);
  26.         self.inner.condvar.notify_all();
  27.     }
  28. }
  30. impl Read for InMemoryStream {
  31.     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
  32.         let mut lock = self.inner.data.lock().unwrap();
  33.         let read = lock.read(buf)?;
  34.         if read > 0 {
  35.             return Ok(read);
  36.         }
  37.         if self.inner.closed.load(Ordering::SeqCst) {
  38.             return Ok(0);
  39.         }
  41.         let mut lock = self.inner.condvar.wait(lock).unwrap();
  42.         lock.read(buf)
  43.     }
  44. }

然后使用它如下：

  1. fn download_and_import_file() -> MyResult<()> {
  2.     let stream = InMemoryStream::default();
  3.     let handle = thread::spawn({
  4.         let stream = stream.clone();
  5.         let mut stream2 = stream.clone();
  6.         || {
  7.             let mut easy = Easy::new();
  8.             easy.url("https://example.com/file.txt.gz").unwrap();
  9.             easy.write_function(move |data| {
  10.                 stream.write(data);
  11.                 return Ok(data.len());
  12.             })
  13.             .unwrap();
  14.             easy.perform().unwrap();
  15.             stream2.close();
  16.         }
  17.     });
  18.     let reader = BufReader::new(GzDecoder::new(stream));
  19.     for line in reader.lines() {
  20.         let line = line?;
  21.         println!("{line}");
  22.     }
  23.     handle.join().unwrap();
  24.     return Ok(());
  25. }

请注意，我已经将HTML编码符号（例如<和>）翻译回原始的Rust代码中的角括号。

英文:

You can create a type that serves as an in-memory Read implementation which you can write into:

  1. use std::collections::VecDeque;
  2. use std::io::{self, Read};
  3. use std::sync::atomic::{AtomicBool, Ordering};
  4. use std::sync::{Arc, Condvar, Mutex};
  6. #[derive(Clone, Default)]
  7. pub struct InMemoryStream {
  8.     inner: Arc&lt;InMemoryStreamInner&gt;,
  9. }
  11. #[derive(Default)]
  12. struct InMemoryStreamInner {
  13.     data: Mutex&lt;VecDeque&lt;u8&gt;&gt;,
  14.     condvar: Condvar,
  15.     closed: AtomicBool,
  16. }
  18. impl InMemoryStream {
  19.     pub fn write(&amp;self, data: &amp;[u8]) {
  20.         self.inner.data.lock().unwrap().extend(data);
  21.         self.inner.condvar.notify_one();
  22.     }
  24.     pub fn close(&amp;self, data: &amp;[u8]) {
  25.         self.inner.closed.store(true, Ordering::SeqCst);
  26.         self.inner.condvar.notify_all();
  27.     }
  28. }
  30. impl Read for InMemoryStream {
  31.     fn read(&amp;mut self, buf: &amp;mut [u8]) -&gt; io::Result&lt;usize&gt; {
  32.         let mut lock = self.inner.data.lock().unwrap();
  33.         let read = lock.read(buf)?;
  34.         if read &gt; 0 {
  35.             return Ok(read);
  36.         }
  37.         if self.inner.closed.load(Ordering::SeqCst) {
  38.             return Ok(0);
  39.         }
  41.         let mut lock = self.inner.condvar.wait(lock).unwrap();
  42.         lock.read(buf)
  43.     }
  44. }

Then use it as follows:

  1. fn download_and_import_file() -&gt; MyResult&lt;()&gt; {
  2.     let stream = InMemoryStream::default();
  3.     let handle = thread::spawn({
  4.         let stream = stream.clone();
  5.         let mut stream2 = stream.clone();
  6.         || {
  7.             let mut easy = Easy::new();
  8.             easy.url(&quot;https://example.com/file.txt.gz&quot;).unwrap();
  9.             easy.write_function(move |data| {
  10.                 stream.write(data);
  11.                 return Ok(data.len());
  12.             })
  13.             .unwrap();
  14.             easy.perform().unwrap();
  15.             stream2.close();
  16.         }
  17.     });
  18.     let reader = BufReader::new(GzDecoder::new(stream));
  19.     for line in reader.lines() {
  20.         let line = line?;
  21.         println!(&quot;{line}&quot;);
  22.     }
  23.     handle.join().unwrap();
  24.     return Ok(());
  25. }

huangapple
  • 本文由 发表于 2023年7月23日 18:06:56
  • 转载请务必保留本文链接：https://go.coder-hub.com/76747671.html
  • libcurl
  • rust
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定