Request
The payload of the channel request: the tag byte 8 and the request of the channel.
[8]- No parameter. The channel request carries nothing after the tag. A server ignores every byte that follows the tag.
The channel request is defined by diverge-provider-sdk/src/endpoints/containers/agents/run/client/channel_request/frame.rs:
//! What a client's channel request frame carries for an agent container run.
use std::error::Error;
use std::fmt;
use crate::decode::Decode;
use crate::encode::{Encode, Writer};
use crate::shared::containers::{enqueue, run_loop};
use crate::shared::containers::{postgres, read, write_path};
/// What a caller asks a provider for while an agent container runs.
///
/// A payload leads with one byte saying which, and the rest is that
/// variant's own bytes.
///
/// | tag | asks for |
/// |-----|----------|
/// | `0` | [`Stop`](Self::Stop) |
/// | `1` | [`Filetree`](Self::Filetree) |
/// | `2` | [`Read`](Self::Read) |
/// | `3` | [`Write`](Self::Write) |
/// | `4` | [`Postgres`](Self::Postgres) |
/// | `5` | [`RunLoop`](Self::RunLoop) |
/// | `6` | [`AgentSchema`](Self::AgentSchema) |
/// | `7` | [`Enqueue`](Self::Enqueue) |
/// | `8` | [`Dequeue`](Self::Dequeue) |
///
/// The first five are the same in every container scope, in the same
/// order, so a reader of one is a reader of all; what follows is this
/// family's own exchange. All of them but the first reach INTO the
/// container, which is the thing a caller cannot dial: it runs on the
/// provider. That is the whole reason these channels open outward
/// from the client rather than the other way.
#[derive(Debug, Clone, PartialEq)]
pub enum Frame {
/// Stop the container. Tag `0`.
///
/// # It has no answer, and does not need one
///
/// Nothing comes back on this channel. What comes back is the end
/// of the SCOPE — a
/// [`ResponseFinish`](crate::frame::server::ServerFrame::ResponseFinish),
/// which already means nothing bearing this scope follows on any
/// channel. Finishing this one first would be a smaller way of
/// saying the same thing, moments earlier.
///
/// # What it adds over closing the connection
///
/// The scope IS the container's life, so dropping the connection
/// stops it too. The difference is that a provider cannot tell a
/// deliberate exit from a network that stopped answering, and has
/// to wait to find out. This is unambiguous and immediate: a
/// caller that says so is not gone, it is finished.
///
/// # What it does to everyone else
///
/// Ends them. Connectors hold scopes on a container that no longer
/// exists, so those scopes finish too — a connection cannot
/// outlive the thing it joined.
Stop,
/// The container's filesystem, watched. Tag `1`.
///
/// Carries nothing — the variant is bare — and the provider answers
/// with a snapshot and then every change, for as long as the
/// channel lives. See
/// [`filetree`](crate::shared::containers::filetree).
Filetree,
/// One file, read out of the container. Tag `2`.
///
/// See [`read`](crate::shared::containers::read) for why this is
/// one file and never a directory.
Read(read::request::Request),
/// One file, written into the container. Tag `3`.
///
/// Carries no content. The provider answers by opening a channel
/// of its own asking for it — see
/// [`write_path`](crate::shared::containers::write_path) for why
/// it travels that direction, and
/// [`write_bytes`](crate::shared::containers::write_bytes) for
/// what comes back.
Write(write_path::request::Request),
/// The caller's half of a database connection. Tag `4`.
///
/// Opened once the caller has taken the provider's half, quoting
/// the same connection; what comes back is everything the
/// container wrote. See
/// [`postgres`](crate::shared::containers::postgres) for the pair.
Postgres(postgres::request::Postgres),
/// Run a loop. Tag `5`.
///
/// The family's own exchange. Carries the prompt — the agent was
/// on the request that made the container, and never changes —
/// and answers with the loop's chunks. See
/// [`run_loop`](crate::shared::containers::run_loop).
RunLoop(run_loop::request::Request),
/// What the agent may be. Tag `6`.
///
/// Carries nothing — the variant is bare — and the provider answers
/// with the JSON Schema of the agent value. See
/// [`agent_schema`](crate::shared::containers::agent_schema).
AgentSchema,
/// A message for the running loop's queue. Tag `7`.
///
/// Answered once — by an
/// [`enqueue::response::Frame`](crate::shared::containers::enqueue::response::Frame)
/// naming the message's fate, whenever that is known — and then
/// the finish. See [`enqueue`](crate::shared::containers::enqueue).
Enqueue(enqueue::request::Request),
/// Withdraw every message still waiting in the queue. Tag `8`.
///
/// Carries nothing — the variant is bare. Answered once — by a
/// [`dequeue::response::Frame`](crate::shared::containers::dequeue::response::Frame)
/// saying whether the queue held anything — and then the finish.
/// Each message it withdraws is ALSO answered, on its own enqueue
/// channel. See [`dequeue`](crate::shared::containers::dequeue).
Dequeue,
}
/// Tag for [`Frame::Stop`].
const STOP: u8 = 0;
/// Tag for [`Frame::Filetree`].
const FILETREE: u8 = 1;
/// Tag for [`Frame::Read`].
const READ: u8 = 2;
/// Tag for [`Frame::Write`].
const WRITE: u8 = 3;
/// Tag for [`Frame::Postgres`].
const POSTGRES: u8 = 4;
/// Tag for [`Frame::RunLoop`].
const RUN_LOOP: u8 = 5;
/// Tag for [`Frame::AgentSchema`].
const AGENT_SCHEMA: u8 = 6;
/// Tag for [`Frame::Enqueue`].
const ENQUEUE: u8 = 7;
/// Tag for [`Frame::Dequeue`].
const DEQUEUE: u8 = 8;
impl Encode for Frame {
/// The ordinary JSON failure, from whichever half has one.
type Error = serde_json::Error;
fn encode(&self, out: &mut Writer<'_>) -> Result<(), Self::Error> {
match self {
Frame::Stop => {
out.extend_from_slice(&[STOP]);
Ok(())
}
Frame::Filetree => {
out.extend_from_slice(&[FILETREE]);
Ok(())
}
Frame::Read(request) => {
out.extend_from_slice(&[READ]);
request.encode(out)
}
Frame::Write(request) => {
out.extend_from_slice(&[WRITE]);
request.encode(out)
}
Frame::Postgres(request) => {
out.extend_from_slice(&[POSTGRES]);
request.encode(out).map_err(|error| match error {})
}
Frame::RunLoop(request) => {
out.extend_from_slice(&[RUN_LOOP]);
request.encode(out)
}
Frame::AgentSchema => {
out.extend_from_slice(&[AGENT_SCHEMA]);
Ok(())
}
Frame::Enqueue(request) => {
out.extend_from_slice(&[ENQUEUE]);
request.encode(out)
}
Frame::Dequeue => {
out.extend_from_slice(&[DEQUEUE]);
Ok(())
}
}
}
}
impl Decode<'_> for Frame {
/// Several ways to fail, and the parses among them name which.
type Error = FrameError;
fn decode(bytes: &[u8]) -> Result<Self, Self::Error> {
let (tag, rest) = bytes.split_first().ok_or(FrameError::Empty)?;
match *tag {
STOP => Ok(Frame::Stop),
FILETREE => Ok(Frame::Filetree),
READ => read::request::Request::decode(rest)
.map(Frame::Read)
.map_err(FrameError::Read),
WRITE => write_path::request::Request::decode(rest)
.map(Frame::Write)
.map_err(FrameError::Write),
POSTGRES => postgres::request::Postgres::decode(rest)
.map(Frame::Postgres)
.map_err(FrameError::Postgres),
RUN_LOOP => run_loop::request::Request::decode(rest)
.map(Frame::RunLoop)
.map_err(FrameError::RunLoop),
AGENT_SCHEMA => Ok(Frame::AgentSchema),
ENQUEUE => enqueue::request::Request::decode(rest)
.map(Frame::Enqueue)
.map_err(FrameError::Enqueue),
DEQUEUE => Ok(Frame::Dequeue),
tag => Err(FrameError::UnknownTag(tag)),
}
}
}
/// An agent container run channel request that could not be read.
#[derive(Debug)]
pub enum FrameError {
/// No bytes at all, so not even a tag.
Empty,
/// A tag that is none of this frame's.
UnknownTag(u8),
/// The read request did not parse.
Read(serde_json::Error),
/// The write request did not parse.
Write(serde_json::Error),
/// The connection id was not four bytes.
Postgres(postgres::request::PostgresError),
/// The loop's prompt did not parse as JSON.
RunLoop(serde_json::Error),
/// The enqueued message did not parse as JSON.
Enqueue(serde_json::Error),
}
impl fmt::Display for FrameError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
FrameError::Empty => {
f.write_str("agents run channel request frame is empty")
}
FrameError::UnknownTag(tag) => {
write!(f, "unknown agents run channel request tag {tag}")
}
FrameError::Read(error) => {
write!(f, "read request did not parse: {error}")
}
FrameError::Write(error) => {
write!(f, "write request did not parse: {error}")
}
FrameError::Postgres(error) => write!(f, "{error}"),
FrameError::RunLoop(error) => {
write!(f, "run loop request did not parse: {error}")
}
FrameError::Enqueue(error) => {
write!(f, "enqueue request did not parse: {error}")
}
}
}
}
impl Error for FrameError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
FrameError::Read(error)
| FrameError::Write(error)
| FrameError::RunLoop(error)
| FrameError::Enqueue(error) => Some(error),
FrameError::Postgres(error) => Some(error),
FrameError::Empty | FrameError::UnknownTag(_) => None,
}
}
}