2021-08-28 16:50:55 +00:00
|
|
|
use std::convert::TryFrom;
|
2021-08-27 14:19:14 +00:00
|
|
|
use std::io;
|
|
|
|
use std::path::PathBuf;
|
|
|
|
use std::sync::Arc;
|
|
|
|
|
|
|
|
use syndicate::actor::*;
|
2021-08-28 16:50:55 +00:00
|
|
|
use syndicate::convert::*;
|
|
|
|
use syndicate::during::entity;
|
2021-08-27 14:19:14 +00:00
|
|
|
use syndicate::error::Error;
|
|
|
|
use syndicate::relay;
|
2021-08-28 16:50:55 +00:00
|
|
|
use syndicate::schemas::dataspace::Observe;
|
|
|
|
use syndicate::value::NestedValue;
|
2021-08-27 14:19:14 +00:00
|
|
|
|
|
|
|
use tokio::net::UnixListener;
|
|
|
|
use tokio::net::UnixStream;
|
|
|
|
|
|
|
|
use crate::protocol::run_connection;
|
2021-08-28 16:50:55 +00:00
|
|
|
use crate::schemas::internal_services;
|
2021-08-27 14:19:14 +00:00
|
|
|
|
2021-08-28 16:50:55 +00:00
|
|
|
pub fn on_demand(t: &mut Activation, ds: Arc<Cap>, gateway: Arc<Cap>) {
|
|
|
|
t.spawn(syndicate::name!("on_demand", module = module_path!()), move |t| {
|
|
|
|
let monitor = entity(())
|
|
|
|
.on_asserted_facet({
|
|
|
|
let ds = Arc::clone(&ds);
|
|
|
|
move |_, t, captures| {
|
|
|
|
let ds = Arc::clone(&ds);
|
|
|
|
let gateway = Arc::clone(&gateway);
|
2021-08-30 10:06:40 +00:00
|
|
|
t.spawn_link(syndicate::name!(parent: None, "relay", addr = ?captures),
|
2021-08-28 16:50:55 +00:00
|
|
|
|t| run(t, ds, gateway, captures));
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
})
|
|
|
|
.create_cap(t);
|
|
|
|
ds.assert(t, &Observe {
|
2021-08-31 14:19:29 +00:00
|
|
|
pattern: syndicate_macros::pattern!{<require-service $(<relay-listener <unix _>>)>},
|
2021-08-28 16:50:55 +00:00
|
|
|
observer: monitor,
|
|
|
|
});
|
|
|
|
Ok(())
|
2021-09-01 15:31:01 +00:00
|
|
|
});
|
2021-08-27 14:19:14 +00:00
|
|
|
}
|
|
|
|
|
2021-08-28 16:50:55 +00:00
|
|
|
fn run(
|
|
|
|
t: &'_ mut Activation,
|
|
|
|
ds: Arc<Cap>,
|
2021-08-27 14:19:14 +00:00
|
|
|
gateway: Arc<Cap>,
|
2021-08-28 16:50:55 +00:00
|
|
|
captures: AnyValue,
|
2021-08-27 14:19:14 +00:00
|
|
|
) -> ActorResult {
|
2021-08-28 16:50:55 +00:00
|
|
|
let spec = internal_services::UnixRelayListener::try_from(&from_any_value(
|
|
|
|
&captures.value().to_sequence()?[0])?)?;
|
|
|
|
let path_str = spec.addr.path.clone();
|
|
|
|
{
|
|
|
|
let spec = from_io_value(&spec)?;
|
|
|
|
ds.assert(t, syndicate_macros::template!("<service-running =spec>"));
|
2021-08-27 14:19:14 +00:00
|
|
|
}
|
2021-08-28 16:50:55 +00:00
|
|
|
let parent_span = tracing::Span::current();
|
|
|
|
t.linked_task(syndicate::name!("listener"), async move {
|
|
|
|
let listener = bind_unix_listener(&PathBuf::from(path_str)).await?;
|
|
|
|
tracing::info!("listening");
|
|
|
|
loop {
|
|
|
|
let (stream, _addr) = listener.accept().await?;
|
|
|
|
let peer = stream.peer_cred()?;
|
|
|
|
let gateway = Arc::clone(&gateway);
|
|
|
|
Actor::new().boot(
|
|
|
|
syndicate::name!(parent: parent_span.clone(), "conn",
|
2021-08-30 10:06:40 +00:00
|
|
|
pid = ?peer.pid().unwrap_or(-1),
|
2021-08-28 16:50:55 +00:00
|
|
|
uid = peer.uid()),
|
|
|
|
|t| Ok(t.linked_task(
|
|
|
|
tracing::Span::current(),
|
|
|
|
{
|
|
|
|
let facet = t.facet.clone();
|
|
|
|
async move {
|
2021-08-30 10:06:40 +00:00
|
|
|
tracing::info!(protocol = %"unix");
|
2021-08-28 16:50:55 +00:00
|
|
|
let (i, o) = stream.into_split();
|
|
|
|
run_connection(facet,
|
|
|
|
relay::Input::Bytes(Box::pin(i)),
|
|
|
|
relay::Output::Bytes(Box::pin(o)),
|
|
|
|
gateway)
|
|
|
|
}
|
|
|
|
})));
|
|
|
|
}
|
|
|
|
});
|
|
|
|
Ok(())
|
2021-08-27 14:19:14 +00:00
|
|
|
}
|
|
|
|
|
2021-08-28 16:50:55 +00:00
|
|
|
async fn bind_unix_listener(path: &PathBuf) -> Result<UnixListener, Error> {
|
2021-08-27 14:19:14 +00:00
|
|
|
match UnixListener::bind(path) {
|
|
|
|
Ok(s) => Ok(s),
|
|
|
|
Err(e) if e.kind() == io::ErrorKind::AddrInUse => {
|
|
|
|
// Potentially-stale socket file sitting around. Try
|
|
|
|
// connecting to it to see if it is alive, and remove it
|
|
|
|
// if not.
|
|
|
|
match UnixStream::connect(path).await {
|
|
|
|
Ok(_probe) => Err(e)?, // Someone's already there! Give up.
|
|
|
|
Err(f) if f.kind() == io::ErrorKind::ConnectionRefused => {
|
|
|
|
// Try to steal the socket.
|
|
|
|
tracing::info!("Cleaning stale socket");
|
|
|
|
std::fs::remove_file(path)?;
|
|
|
|
Ok(UnixListener::bind(path)?)
|
|
|
|
}
|
2021-08-30 10:06:40 +00:00
|
|
|
Err(error) => {
|
|
|
|
tracing::error!(?error, "Problem while probing potentially-stale socket");
|
2021-08-27 14:19:14 +00:00
|
|
|
return Err(e)? // signal the *original* error, not the probe error
|
|
|
|
}
|
|
|
|
}
|
|
|
|
},
|
|
|
|
Err(e) => Err(e)?,
|
|
|
|
}
|
|
|
|
}
|