| 683 | } |
| 684 | |
| 685 | async fn build_image( |
| 686 | &self, |
| 687 | options: BuildImageOptions, |
| 688 | tar_archive: Vec<u8>, |
| 689 | ) -> Result<Box<dyn futures_util::Stream<Item = Result<String, String>> + Send + Unpin>, String> |
| 690 | { |
| 691 | use futures_util::StreamExt; |
| 692 | use tokio::sync::mpsc; |
| 693 | |
| 694 | let (tx, rx) = mpsc::unbounded_channel(); |
| 695 | let docker = self.docker.clone(); |
| 696 | |
| 697 | // Spawn a task to handle the streaming |
| 698 | tokio::spawn(async move { |
| 699 | use bytes::Bytes; |
| 700 | use http_body_util::{Either, Full}; |
| 701 | |
| 702 | let body = Either::Left(Full::new(Bytes::from(tar_archive))); |
| 703 | let mut stream = docker.build_image(options, None, Some(body)); |
| 704 | |
| 705 | while let Some(build_info) = stream.next().await { |
| 706 | match build_info { |
| 707 | Ok(info) => { |
| 708 | if let Some(error_detail) = info.error_detail { |
| 709 | let error_msg = error_detail |
| 710 | .message |
| 711 | .unwrap_or_else(|| "Unknown error".to_string()); |
| 712 | let _ = tx.send(Err(format!("Docker build error: {error_msg}"))); |
| 713 | break; |
| 714 | } else if let Some(stream_msg) = info.stream |
| 715 | && !stream_msg.is_empty() |
| 716 | && tx.send(Ok(stream_msg)).is_err() |
| 717 | { |
| 718 | break; // Receiver dropped |
| 719 | } |
| 720 | } |
| 721 | Err(e) => { |
| 722 | let _ = tx.send(Err(format!("Failed to build image: {e}"))); |
| 723 | break; |
| 724 | } |
| 725 | } |
| 726 | } |
| 727 | }); |
| 728 | |
| 729 | // Convert receiver to stream |
| 730 | let receiver_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx); |
| 731 | Ok(Box::new(Box::pin(receiver_stream))) |
| 732 | } |
| 733 | |
| 734 | async fn inspect_container(&self, id: &str) -> Result<String, String> { |
| 735 | let container = self |