| 270 | } |
| 271 | |
| 272 | pub async fn create_streamer(output: &str) -> Result<SeaProducer> { |
| 273 | if let Some(dir) = Path::new(output).parent() { |
| 274 | create_dir_all(dir) |
| 275 | .await |
| 276 | .with_context(|| format!("Fail to create directory: `{}`", dir.display()))?; |
| 277 | } |
| 278 | let file_id = FileId::new(output); |
| 279 | let mut options = SeaConnectOptions::default(); |
| 280 | options.set_file_connect_options(|options| { |
| 281 | options.set_create_only(true); |
| 282 | options.set_end_with_eos(true); |
| 283 | }); |
| 284 | let uri = file_id.to_streamer_uri().context("Fail to get URI")?; |
| 285 | let streamer = SeaStreamer::connect(uri, options) |
| 286 | .await |
| 287 | .context("Fail to connect streamer")?; |
| 288 | let producer = streamer |
| 289 | .create_generic_producer(Default::default()) |
| 290 | .await |
| 291 | .context("Fail to create producer")?; |
| 292 | |
| 293 | Ok(producer) |
| 294 | } |
| 295 | |
| 296 | fn print_to_stdout(bytes: sea_streamer::file::Bytes) -> Result<()> { |
| 297 | print!( |