use super::{Active, Owned};
use std::time::Duration;
const GRACE: Duration = Duration::from_secs(2);
pub(super) async fn discard(owned: &mut Owned) -> anyhow::Result<()> {
tokio::time::timeout(GRACE, async {
while let Some(frame) = owned.pipe.input.recv().await {
match frame {
Err(error) => return Err(error),
Ok(None) => break,
Ok(Some(_)) => {}
}
}
Ok(())
})
.await
.map_err(|_| anyhow::anyhow!("acp_close_drain_timeout"))?
}
pub(super) async fn active(
owned: &mut Owned,
mut active: Active,
result: anyhow::Result<super::End>,
) -> anyhow::Result<i32> {
active.pump.closing = true;
let mut cleanup = owned.child.shutdown(&mut owned.exited).await;
if !active.stdout_closed {
let drained = tokio::time::timeout(GRACE, async {
while let Some(frame) = owned.pipe.input.recv().await {
match frame {
Ok(Some(super::super::wire::Frame::Notification { method, params }))
if method == "session/update" =>
{
active.pump.closing_update(params).await?;
}
Ok(None) => break,
Err(error) => return Err(error),
Ok(Some(_)) => {}
}
}
Ok(())
})
.await
.map_err(|_| anyhow::anyhow!("acp_close_drain_timeout"))
.and_then(std::convert::identity);
cleanup.push(drained);
}
if !owned.pipe_reader_seen {
cleanup.push(
super::super::closing::stop(&mut owned.pipe.reader, "acp_reader")
.await
.map(|_| ()),
);
owned.pipe_reader_seen = true;
}
let reason = match &result {
Ok(end) => end.reason.clone(),
Err(error) => super::End::failed(error).reason,
};
let status = result.as_ref().map_or(1, |end| end.status);
let status = if status == 0 && cleanup.iter().any(Result::is_err) {
1
} else {
status
};
cleanup.push(
tokio::time::timeout(GRACE, active.pump.end(status, &reason))
.await
.map_err(|_| anyhow::anyhow!("publisher_close_input_timeout"))
.and_then(std::convert::identity),
);
drop(active.pump);
cleanup.extend(publisher(owned, &mut active.publisher, active.publisher_seen).await);
if !owned.door_writer_seen {
let door = owned
.link
.as_ref()
.ok_or_else(|| anyhow::anyhow!("door_link_absent"))?;
cleanup.push(
tokio::time::timeout(GRACE, door.output.flush())
.await
.map_err(|_| anyhow::anyhow!("door_close_flush_timeout"))
.and_then(std::convert::identity),
);
}
Ok(super::report(result.map(|end| end.status), cleanup))
}
async fn publisher(
owned: &mut Owned,
publisher: &mut tokio::task::JoinHandle<anyhow::Result<()>>,
seen: bool,
) -> Vec<anyhow::Result<()>> {
let mut cleanup = Vec::new();
if !seen {
if let Ok(result) = tokio::time::timeout(GRACE, &mut *publisher).await {
cleanup.push(
result
.map_err(|_| anyhow::anyhow!("publisher_panicked"))
.and_then(std::convert::identity),
);
} else {
cleanup.push(Err(anyhow::anyhow!("publisher_close_grace_elapsed")));
if let Some(door) = owned.link.as_mut() {
cleanup.push(
super::super::closing::stop(&mut door.writer, "door_writer")
.await
.and_then(|result| result.unwrap_or(Ok(()))),
);
owned.door_writer_seen = true;
}
publisher.abort();
match tokio::time::timeout(GRACE, &mut *publisher).await {
Ok(Ok(result)) => cleanup.push(result),
Ok(Err(error)) if error.is_cancelled() => cleanup.push(Err(anyhow::anyhow!(
"publisher_closed_with_uncommitted_items"
))),
Ok(Err(_)) => cleanup.push(Err(anyhow::anyhow!("publisher_panicked"))),
Err(_) => cleanup.push(Err(anyhow::anyhow!(
"publisher_close_timeout_outcome_unknown"
))),
}
}
}
cleanup
}