cffb5e7e65
That way, we can (hopefully) add more fields in the future without breaking stuff. I don't yet know what such fields could be, though.
922 lines
34 KiB
Rust
922 lines
34 KiB
Rust
// Copyright (c) 2024 Jonas Schäfer <jonas@zombofant.net>
|
|
//
|
|
// This Source Code Form is subject to the terms of the Mozilla Public
|
|
// License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
|
|
use alloc::borrow::Cow;
|
|
use core::{
|
|
future::Future,
|
|
pin::Pin,
|
|
task::{Context, Poll},
|
|
time::Duration,
|
|
};
|
|
use std::io;
|
|
|
|
use futures::{ready, Sink, SinkExt, Stream, StreamExt};
|
|
|
|
use bytes::{Buf, BytesMut};
|
|
|
|
use tokio::{
|
|
io::{AsyncBufRead, AsyncWrite},
|
|
time::Instant,
|
|
};
|
|
|
|
use xso::{
|
|
exports::rxml::{
|
|
self, writer::TrackNamespace, xml_lang::XmlLangStack, xml_ncname, Event, Namespace,
|
|
},
|
|
AsXml, FromEventsBuilder, FromXml, Item,
|
|
};
|
|
|
|
use crate::connect::AsyncReadAndWrite;
|
|
|
|
use super::capture::{log_enabled, log_recv, log_send, CaptureBufRead};
|
|
|
|
use xmpp_parsers::ns::STREAM as XML_STREAM_NS;
|
|
|
|
/// Configuration for timeouts on an XML stream.
|
|
///
|
|
/// The defaults are tuned toward common desktop/laptop use and may not hold
|
|
/// up to extreme conditions (arctic satellite link, mobile internet on a
|
|
/// train in Brandenburg, Germany, and similar) and may be inefficient in
|
|
/// other conditions (stable server link, localhost communication).
|
|
#[derive(Debug, Clone, Copy)]
|
|
pub struct Timeouts {
|
|
/// Maximum silence time before a
|
|
/// [`ReadError::SoftTimeout`][`super::ReadError::SoftTimeout`] is
|
|
/// returned.
|
|
///
|
|
/// Soft timeouts are not fatal, but they must be handled by user code so
|
|
/// that more data is read after at most [`Self::response_timeout`],
|
|
/// starting from the moment the soft timeout is returned.
|
|
pub read_timeout: Duration,
|
|
|
|
/// Maximum silence after a soft timeout.
|
|
///
|
|
/// If the stream is silent for longer than this time after a soft timeout
|
|
/// has been emitted, a hard [`TimedOut`][`io::ErrorKind::TimedOut`]
|
|
/// I/O error is returned and the stream is to be considered dead.
|
|
pub response_timeout: Duration,
|
|
}
|
|
|
|
impl Default for Timeouts {
|
|
fn default() -> Self {
|
|
Self {
|
|
read_timeout: Duration::new(300, 0),
|
|
response_timeout: Duration::new(300, 0),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Timeouts {
|
|
/// Tight timeouts suitable for communicating on a fast LAN or localhost.
|
|
pub fn tight() -> Self {
|
|
Self {
|
|
read_timeout: Duration::new(60, 0),
|
|
response_timeout: Duration::new(15, 0),
|
|
}
|
|
}
|
|
|
|
fn data_to_soft(&self) -> Duration {
|
|
self.read_timeout
|
|
}
|
|
|
|
fn soft_to_warn(&self) -> Duration {
|
|
self.response_timeout / 2
|
|
}
|
|
|
|
fn warn_to_hard(&self) -> Duration {
|
|
self.response_timeout / 2
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Copy)]
|
|
enum TimeoutLevel {
|
|
Soft,
|
|
Warn,
|
|
Hard,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub(super) enum RawError {
|
|
Io(io::Error),
|
|
SoftTimeout,
|
|
}
|
|
|
|
impl From<io::Error> for RawError {
|
|
fn from(other: io::Error) -> Self {
|
|
Self::Io(other)
|
|
}
|
|
}
|
|
|
|
struct TimeoutState {
|
|
/// Configuration for the timeouts.
|
|
timeouts: Timeouts,
|
|
|
|
/// Level of the next timeout which will trip.
|
|
level: TimeoutLevel,
|
|
|
|
/// Sleep timer used for read timeouts.
|
|
// NOTE: even though we pretend we could deal with an !Unpin
|
|
// RawXmlStream, we really can't: box_stream for example needs it,
|
|
// but also all the typestate around the initial stream setup needs
|
|
// to be able to move the stream around.
|
|
deadline: Pin<Box<tokio::time::Sleep>>,
|
|
}
|
|
|
|
impl TimeoutState {
|
|
fn new(timeouts: Timeouts) -> Self {
|
|
Self {
|
|
deadline: Box::pin(tokio::time::sleep(timeouts.data_to_soft())),
|
|
level: TimeoutLevel::Soft,
|
|
timeouts,
|
|
}
|
|
}
|
|
|
|
fn poll(&mut self, cx: &mut Context) -> Poll<TimeoutLevel> {
|
|
ready!(self.deadline.as_mut().poll(cx));
|
|
// Deadline elapsed!
|
|
let to_return = self.level;
|
|
let (next_level, next_duration) = match self.level {
|
|
TimeoutLevel::Soft => (TimeoutLevel::Warn, self.timeouts.soft_to_warn()),
|
|
TimeoutLevel::Warn => (TimeoutLevel::Hard, self.timeouts.warn_to_hard()),
|
|
// Something short-ish so that we fire this over and over until
|
|
// someone finally kills the stream for good.
|
|
TimeoutLevel::Hard => (TimeoutLevel::Hard, Duration::new(1, 0)),
|
|
};
|
|
self.level = next_level;
|
|
self.deadline.as_mut().reset(Instant::now() + next_duration);
|
|
Poll::Ready(to_return)
|
|
}
|
|
|
|
fn reset(&mut self) {
|
|
self.level = TimeoutLevel::Soft;
|
|
self.deadline
|
|
.as_mut()
|
|
.reset(Instant::now() + self.timeouts.data_to_soft());
|
|
}
|
|
}
|
|
|
|
pin_project_lite::pin_project! {
|
|
// NOTE: due to limitations of pin_project_lite, the field comments are
|
|
// no doc comments. Luckily, this struct is only `pub(super)` anyway.
|
|
#[project = RawXmlStreamProj]
|
|
pub(super) struct RawXmlStream<Io> {
|
|
// The parser used for deserialising data.
|
|
#[pin]
|
|
parser: rxml::AsyncReader<CaptureBufRead<Io>>,
|
|
|
|
// Tracker for `xml:lang` data.
|
|
lang_stack: XmlLangStack,
|
|
|
|
// The writer used for serialising data.
|
|
writer: rxml::writer::Encoder<rxml::writer::SimpleNamespaces>,
|
|
|
|
timeouts: TimeoutState,
|
|
|
|
// The default namespace to declare on the stream header.
|
|
stream_ns: &'static str,
|
|
|
|
// Buffer containing serialised data which will then be sent through
|
|
// the inner `Io`. Sending that serialised data happens in
|
|
// `poll_ready` and `poll_flush`, while appending serialised data
|
|
// happens in `start_send`.
|
|
tx_buffer: BytesMut,
|
|
|
|
// Position inside tx_buffer up to which to-be-sent data has already
|
|
// been logged.
|
|
tx_buffer_logged: usize,
|
|
|
|
// This signifies the limit at the point of which the Sink will
|
|
// refuse to accept more data: if the `tx_buffer`'s size grows beyond
|
|
// that high water mark, poll_ready will return Poll::Pending until
|
|
// it has managed to flush enough data down the inner writer.
|
|
//
|
|
// Note that poll_ready will always attempt to progress the writes,
|
|
// which further reduces the chance of hitting this limit unless
|
|
// either the underlying writer gets stuck (e.g. TCP connection
|
|
// breaking in a timeouty way) or a lot of data is written in bulk.
|
|
// In both cases, the backpressure created by poll_ready returning
|
|
// Pending is desirable.
|
|
//
|
|
// However, there is a catch: We don't assert this condition
|
|
// in `start_send` at all. The reason is that we cannot suspend
|
|
// serialisation of an XSO in the middle of writing it: it has to be
|
|
// written in one batch or you have to start over later (this has to
|
|
// do with the iterator state borrowing the data and futures getting
|
|
// cancelled e.g. in tokio::select!). In order to facilitate
|
|
// implementing a `Sink<T: AsXml>` on top of `RawXmlStream`, we
|
|
// cannot be strict about what is going on in `start_send`:
|
|
// `poll_ready` does not know what kind of data will be written (so
|
|
// it could not make a size estimate, even if that was at all
|
|
// possible with AsXml) and `start_send` is not a coroutine. So if
|
|
// `Sink<T: AsXml>` wants to use `RawXmlStream`, it must be able to
|
|
// submit an entire XSO's items in one batch to `RawXmlStream` after
|
|
// it has reported to be ready once. That may easily make the buffer
|
|
// reach its high water mark.
|
|
//
|
|
// So if we checked that condition in `start_send` (as opposed to
|
|
// `poll_ready`), we would cause situations where submitting XSOs
|
|
// failed randomly (with a panic or other errors) and would have to
|
|
// be retried later.
|
|
//
|
|
// While failing with e.g. io::ErrorKind::WouldBlock is something
|
|
// that could be investigated later, it would still require being
|
|
// able to make an accurate estimate of the number of bytes needed to
|
|
// serialise any given `AsXml`, because as pointed out earlier, once
|
|
// we have started, there is no going back.
|
|
//
|
|
// Finally, none of that hurts much because `RawXmlStream` is only an
|
|
// internal API. The high-level APIs will always call `poll_ready`
|
|
// before sending an XSO, which means that we won't *grossly* go over
|
|
// the TX buffer high water mark---unless you send a really large
|
|
// XSO at once.
|
|
tx_buffer_high_water_mark: usize,
|
|
}
|
|
}
|
|
|
|
impl<Io: AsyncBufRead + AsyncWrite> RawXmlStream<Io> {
|
|
fn new_writer(
|
|
stream_ns: &'static str,
|
|
) -> rxml::writer::Encoder<rxml::writer::SimpleNamespaces> {
|
|
let mut writer = rxml::writer::Encoder::new();
|
|
writer
|
|
.ns_tracker_mut()
|
|
.declare_fixed(Some(xml_ncname!("stream")), XML_STREAM_NS.into());
|
|
writer
|
|
.ns_tracker_mut()
|
|
.declare_fixed(None, stream_ns.into());
|
|
writer
|
|
}
|
|
|
|
pub(super) fn new(io: Io, stream_ns: &'static str, timeouts: Timeouts) -> Self {
|
|
let parser = rxml::Parser::default();
|
|
let mut io = CaptureBufRead::wrap(io);
|
|
if log_enabled() {
|
|
io.enable_capture();
|
|
}
|
|
Self {
|
|
parser: rxml::AsyncReader::wrap(io, parser),
|
|
writer: Self::new_writer(stream_ns),
|
|
lang_stack: XmlLangStack::new(),
|
|
timeouts: TimeoutState::new(timeouts),
|
|
tx_buffer_logged: 0,
|
|
stream_ns,
|
|
tx_buffer: BytesMut::new(),
|
|
|
|
// This basically means: "if we already have 2 kiB in our send
|
|
// buffer, do not accept more data".
|
|
// Please see the extensive words at
|
|
//`Self::tx_buffer_high_water_mark` for details.
|
|
tx_buffer_high_water_mark: 2048,
|
|
}
|
|
}
|
|
|
|
pub(super) fn reset_state(self: Pin<&mut Self>) {
|
|
let this = self.project();
|
|
*this.parser.parser_pinned() = rxml::Parser::default();
|
|
*this.lang_stack = XmlLangStack::new();
|
|
*this.writer = Self::new_writer(this.stream_ns);
|
|
}
|
|
|
|
pub(super) fn into_inner(self) -> Io {
|
|
self.parser.into_inner().0.into_inner()
|
|
}
|
|
|
|
/// Box the underlying transport stream.
|
|
///
|
|
/// This removes the specific type of the transport from the XML stream's
|
|
/// type signature.
|
|
pub(super) fn box_stream(self) -> RawXmlStream<Box<dyn AsyncReadAndWrite + Send + 'static>>
|
|
where
|
|
Io: AsyncReadAndWrite + Send + 'static,
|
|
{
|
|
let (io, p) = self.parser.into_inner();
|
|
let mut io = CaptureBufRead::wrap(Box::new(io) as Box<_>);
|
|
if log_enabled() {
|
|
io.enable_capture();
|
|
}
|
|
let parser = rxml::AsyncReader::wrap(io, p);
|
|
RawXmlStream {
|
|
parser,
|
|
lang_stack: XmlLangStack::new(),
|
|
timeouts: self.timeouts,
|
|
writer: self.writer,
|
|
tx_buffer: self.tx_buffer,
|
|
tx_buffer_logged: self.tx_buffer_logged,
|
|
tx_buffer_high_water_mark: self.tx_buffer_high_water_mark,
|
|
stream_ns: self.stream_ns,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<Io: AsyncWrite> RawXmlStream<Io> {
|
|
/// Start sending an entire XSO.
|
|
///
|
|
/// Unlike the `Sink` implementation, this provides nice syntax
|
|
/// highlighting for the serialised data in log outputs (if enabled) *and*
|
|
/// is error safe: if the XSO fails to serialise completely, it will be as
|
|
/// if it hadn't been attempted to serialise it at all.
|
|
///
|
|
/// Note that, like with `start_send`, the caller is responsible for
|
|
/// ensuring that the stream is ready by polling
|
|
/// [`<Self as Sink>::poll_ready`] as needed.
|
|
pub(super) fn start_send_xso<T: AsXml>(self: Pin<&mut Self>, xso: &T) -> io::Result<()> {
|
|
let mut this = self.project();
|
|
let prev_len = this.tx_buffer.len();
|
|
match this.try_send_xso(xso) {
|
|
Ok(()) => Ok(()),
|
|
Err(e) => {
|
|
let curr_len = this.tx_buffer.len();
|
|
this.tx_buffer.truncate(prev_len);
|
|
log::trace!(
|
|
"SEND failed: {}. Rewinding buffer by {} bytes.",
|
|
e,
|
|
curr_len - prev_len
|
|
);
|
|
Err(e)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<Io> RawXmlStream<Io> {
|
|
fn parser_pinned(self: Pin<&mut Self>) -> &mut rxml::Parser {
|
|
self.project().parser.parser_pinned()
|
|
}
|
|
|
|
fn stream_pinned(self: Pin<&mut Self>) -> Pin<&mut CaptureBufRead<Io>> {
|
|
self.project().parser.inner_pinned()
|
|
}
|
|
|
|
pub(super) fn get_stream(&self) -> &Io {
|
|
self.parser.inner().inner()
|
|
}
|
|
}
|
|
|
|
impl<Io: AsyncBufRead> Stream for RawXmlStream<Io> {
|
|
type Item = Result<rxml::Event, RawError>;
|
|
|
|
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
|
let mut this = self.project();
|
|
loop {
|
|
match this.parser.as_mut().poll_read(cx) {
|
|
Poll::Pending => (),
|
|
Poll::Ready(v) => {
|
|
this.timeouts.reset();
|
|
match v.transpose() {
|
|
// Skip the XML declaration, nobody wants to hear about that.
|
|
Some(Ok(rxml::Event::XmlDeclaration(_, _))) => continue,
|
|
Some(Ok(event)) => {
|
|
this.lang_stack.handle_event(&event);
|
|
return Poll::Ready(Some(Ok(event)));
|
|
}
|
|
other => return Poll::Ready(other.map(|x| x.map_err(RawError::Io))),
|
|
}
|
|
}
|
|
};
|
|
|
|
// poll_read returned pending... what do the timeouts have to say?
|
|
match ready!(this.timeouts.poll(cx)) {
|
|
TimeoutLevel::Soft => return Poll::Ready(Some(Err(RawError::SoftTimeout))),
|
|
TimeoutLevel::Warn => (),
|
|
TimeoutLevel::Hard => {
|
|
return Poll::Ready(Some(Err(RawError::Io(io::Error::new(
|
|
io::ErrorKind::TimedOut,
|
|
"read and response timeouts elapsed",
|
|
)))))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<Io: AsyncWrite> RawXmlStreamProj<'_, Io> {
|
|
fn flush_tx_log(&mut self) {
|
|
let range = &self.tx_buffer[*self.tx_buffer_logged..];
|
|
if range.is_empty() {
|
|
return;
|
|
}
|
|
log_send(range);
|
|
*self.tx_buffer_logged = self.tx_buffer.len();
|
|
}
|
|
|
|
fn start_send(&mut self, item: &xso::Item<'_>) -> io::Result<()> {
|
|
self.writer
|
|
.encode_into_bytes(item.as_rxml_item(), self.tx_buffer)
|
|
.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))
|
|
}
|
|
|
|
fn try_send_xso<T: AsXml>(&mut self, xso: &T) -> io::Result<()> {
|
|
let iter = match xso.as_xml_iter() {
|
|
Ok(v) => v,
|
|
Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidInput, e)),
|
|
};
|
|
for item in iter {
|
|
let item = match item {
|
|
Ok(v) => v,
|
|
Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidInput, e)),
|
|
};
|
|
self.start_send(&item)?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn progress_write(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
|
|
self.flush_tx_log();
|
|
while !self.tx_buffer.is_empty() {
|
|
let written = match ready!(self
|
|
.parser
|
|
.as_mut()
|
|
.inner_pinned()
|
|
.poll_write(cx, self.tx_buffer))
|
|
{
|
|
Ok(v) => v,
|
|
Err(e) => return Poll::Ready(Err(e)),
|
|
};
|
|
self.tx_buffer.advance(written);
|
|
*self.tx_buffer_logged = self
|
|
.tx_buffer_logged
|
|
.checked_sub(written)
|
|
.expect("Buffer arithmetic error");
|
|
}
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
}
|
|
|
|
impl<Io: AsyncWrite> RawXmlStream<Io> {
|
|
/// Flush all buffered data and shut down the sender side of the
|
|
/// underlying transport.
|
|
///
|
|
/// Unlike `poll_close` (from the `Sink` impls), this will not close the
|
|
/// receiving side of the underlying the transport. It is advisable to call
|
|
/// `poll_close` eventually after `poll_shutdown` in order to gracefully
|
|
/// handle situations where the remote side does not close the stream
|
|
/// cleanly.
|
|
pub fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
|
ready!(self.as_mut().poll_flush(cx))?;
|
|
let this = self.project();
|
|
this.parser.inner_pinned().poll_shutdown(cx)
|
|
}
|
|
}
|
|
|
|
impl<'x, Io: AsyncWrite> Sink<xso::Item<'x>> for RawXmlStream<Io> {
|
|
type Error = io::Error;
|
|
|
|
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
|
let mut this = self.project();
|
|
match this.progress_write(cx) {
|
|
// No progress on write, but if we have enough space in the buffer
|
|
// it's ok nonetheless.
|
|
Poll::Pending => (),
|
|
// Some progress and it went fine, move on.
|
|
Poll::Ready(Ok(())) => (),
|
|
// Something went wrong -> return the error.
|
|
Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
|
|
}
|
|
if this.tx_buffer.len() < *this.tx_buffer_high_water_mark {
|
|
Poll::Ready(Ok(()))
|
|
} else {
|
|
Poll::Pending
|
|
}
|
|
}
|
|
|
|
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
|
let mut this = self.project();
|
|
ready!(this.progress_write(cx))?;
|
|
this.parser.as_mut().inner_pinned().poll_flush(cx)
|
|
}
|
|
|
|
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
|
let mut this = self.project();
|
|
ready!(this.progress_write(cx))?;
|
|
this.parser.as_mut().inner_pinned().poll_shutdown(cx)
|
|
}
|
|
|
|
fn start_send(self: Pin<&mut Self>, item: xso::Item<'x>) -> Result<(), Self::Error> {
|
|
let mut this = self.project();
|
|
this.start_send(&item)
|
|
}
|
|
}
|
|
|
|
/// Error returned by the [`ReadXso`] future and the [`ReadXsoState`] helper.
|
|
pub(super) enum ReadXsoError {
|
|
/// The outer element was closed before a child element could be read.
|
|
///
|
|
/// This is typically the stream footer in XML stream applications.
|
|
Footer,
|
|
|
|
/// A hard error occurred.
|
|
///
|
|
/// This is either a real I/O error or an error from the XML parser.
|
|
/// Neither are recoverable, because the nesting state is lost and
|
|
/// in addition, XML errors are not recoverable because they indicate a
|
|
/// not well-formed document.
|
|
Hard(io::Error),
|
|
|
|
/// The underlying stream signalled a soft read timeout before a child
|
|
/// element could be read.
|
|
///
|
|
/// Note that soft timeouts which are triggered in the middle of receiving
|
|
/// an element are converted to hard timeouts (i.e. I/O errors).
|
|
///
|
|
/// This masking is intentional, because:
|
|
/// - Returning a [`Self::SoftTimeout`] from the middle of parsing is not
|
|
/// possible without complicating the API.
|
|
/// - There is no reason why the remote side should interrupt sending data
|
|
/// in the middle of an element except if it or the transport has failed
|
|
/// fatally.
|
|
SoftTimeout,
|
|
|
|
/// A parse error occurred.
|
|
///
|
|
/// The XML structure was well-formed, but the data contained did not
|
|
/// match the XSO which was attempted to be parsed. This error is
|
|
/// recoverable: when this error is emitted, the XML stream is at the same
|
|
/// nesting level as it was before the XSO was attempted to be read; all
|
|
/// XML structure which belonged to the XSO which failed to parse has
|
|
/// been consumed. This allows to read more XSOs even if one fails to
|
|
/// parse.
|
|
Parse(xso::error::Error),
|
|
}
|
|
|
|
impl From<io::Error> for ReadXsoError {
|
|
fn from(other: io::Error) -> Self {
|
|
Self::Hard(other)
|
|
}
|
|
}
|
|
|
|
impl From<xso::error::Error> for ReadXsoError {
|
|
fn from(other: xso::error::Error) -> Self {
|
|
Self::Parse(other)
|
|
}
|
|
}
|
|
|
|
/// State for reading an XSO from a `Stream<Item = Result<rxml::Event, ...>>`.
|
|
///
|
|
/// Due to pinning, it is simpler to implement the statemachine in a dedicated
|
|
/// enum and let the actual (pinned) future pass the stream toward this enum's
|
|
/// function.
|
|
///
|
|
/// This is used by both [`ReadXso`] and the [`super::XmlStream`] itself.
|
|
#[derive(Default)]
|
|
pub(super) enum ReadXsoState<T: FromXml> {
|
|
/// The [`rxml::Event::StartElement`] event was not seen yet.
|
|
///
|
|
/// In this state, XML whitespace is ignored (as per RFC 6120 § 11.7), but
|
|
/// other text data is rejected.
|
|
#[default]
|
|
PreData,
|
|
|
|
/// The [`rxml::Event::StartElement`] event was received.
|
|
///
|
|
/// The inner value is the builder for the "return type" of this enum and
|
|
/// the implementation in the [`xso`] crate does all the heavy lifting:
|
|
/// we'll only send events in its general direction.
|
|
// We use the fallible parsing here so that we don't have to do the depth
|
|
// accounting ourselves.
|
|
Parsing(<Result<T, xso::error::Error> as FromXml>::Builder),
|
|
|
|
/// The parsing has completed (successful or not).
|
|
///
|
|
/// This is a final state and attempting to advance the state will panic.
|
|
/// This is in accordance with [`core::future::Future::poll`]'s contract,
|
|
/// for which this enum is primarily used.
|
|
Done,
|
|
}
|
|
|
|
impl<T: FromXml> ReadXsoState<T> {
|
|
/// Progress reading the XSO from the given source.
|
|
///
|
|
/// This attempts to parse a single XSO from the underlying stream,
|
|
/// while discarding any XML whitespace before the beginning of the XSO.
|
|
///
|
|
/// If the XSO is parsed successfully, the method returns Ready with the
|
|
/// parsed value. If parsing fails or an I/O error occurs, an appropriate
|
|
/// error is returned.
|
|
///
|
|
/// If parsing fails, the entire XML subtree belonging to the XSO is
|
|
/// nonetheless processed. That makes parse errors recoverable: After
|
|
/// `poll_advance` has returned Ready with either an Ok result or a
|
|
/// [`ReadXsoError::Parse`] error variant, another XSO can be read and the
|
|
/// XML parsing will be at the same nesting depth as it was before the
|
|
/// first call to `poll_advance`.
|
|
///
|
|
/// Note that this guarantee does not hold for non-parse errors (i.e. for
|
|
/// the other variants of [`ReadXsoError`]): I/O errors as well as
|
|
/// occurrence of the outer closing element are fatal.
|
|
///
|
|
/// The `source` passed to `poll_advance` should be the same on every
|
|
/// call.
|
|
pub(super) fn poll_advance<Io: AsyncBufRead>(
|
|
&mut self,
|
|
mut source: Pin<&mut RawXmlStream<Io>>,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<T, ReadXsoError>> {
|
|
loop {
|
|
// Disable text buffering before the start event. That way, we
|
|
// don't accumulate infinite amounts of XML whitespace caused by
|
|
// whitespace keepalives.
|
|
// (And also, we'll know faster when the remote side sends
|
|
// non-whitespace garbage.)
|
|
let text_buffering = !matches!(self, ReadXsoState::PreData);
|
|
source
|
|
.as_mut()
|
|
.parser_pinned()
|
|
.set_text_buffering(text_buffering);
|
|
|
|
let ev = ready!(source.as_mut().poll_next(cx)).transpose();
|
|
match self {
|
|
ReadXsoState::PreData => {
|
|
log::trace!("ReadXsoState::PreData ev = {:?}", ev);
|
|
match ev {
|
|
Ok(Some(rxml::Event::XmlDeclaration(_, _))) => (),
|
|
Ok(Some(rxml::Event::Text(_, data))) => {
|
|
if xso::is_xml_whitespace(data.as_bytes()) {
|
|
log::trace!("Received {} bytes of whitespace", data.len());
|
|
source.as_mut().stream_pinned().discard_capture();
|
|
continue;
|
|
} else {
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
"non-whitespace text content before XSO",
|
|
)
|
|
.into()));
|
|
}
|
|
}
|
|
Ok(Some(rxml::Event::StartElement(_, name, attrs))) => {
|
|
let source_tmp = source.as_mut();
|
|
let ctx = xso::Context::empty()
|
|
.with_language(source_tmp.lang_stack.current());
|
|
*self = ReadXsoState::Parsing(
|
|
<Result<T, xso::error::Error> as FromXml>::from_events(
|
|
name, attrs, &ctx,
|
|
)
|
|
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?,
|
|
);
|
|
}
|
|
// Amounts to EOF, as we expect to start on the stream level.
|
|
Ok(Some(rxml::Event::EndElement(_))) => {
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(ReadXsoError::Footer));
|
|
}
|
|
Ok(None) => {
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
"eof before XSO started",
|
|
)
|
|
.into()));
|
|
}
|
|
Err(RawError::SoftTimeout) => {
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(ReadXsoError::SoftTimeout));
|
|
}
|
|
Err(RawError::Io(e)) => {
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(ReadXsoError::Hard(e)));
|
|
}
|
|
}
|
|
}
|
|
ReadXsoState::Parsing(builder) => {
|
|
log::trace!("ReadXsoState::Parsing ev = {:?}", ev);
|
|
let ev = match ev {
|
|
Ok(Some(ev)) => ev,
|
|
Ok(None) => {
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(io::Error::new(
|
|
io::ErrorKind::UnexpectedEof,
|
|
"eof during XSO parsing",
|
|
)
|
|
.into()));
|
|
}
|
|
Err(RawError::Io(e)) => {
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(e.into()));
|
|
}
|
|
Err(RawError::SoftTimeout) => {
|
|
// See also [`ReadXsoError::SoftTimeout`] for why
|
|
// we mask the SoftTimeout condition here.
|
|
*self = ReadXsoState::Done;
|
|
return Poll::Ready(Err(io::Error::new(
|
|
io::ErrorKind::TimedOut,
|
|
"read timeout during XSO parsing",
|
|
)
|
|
.into()));
|
|
}
|
|
};
|
|
|
|
let source_tmp = source.as_mut();
|
|
let ctx = xso::Context::empty().with_language(source_tmp.lang_stack.current());
|
|
match builder.feed(ev, &ctx) {
|
|
Err(err) => {
|
|
*self = ReadXsoState::Done;
|
|
source.as_mut().stream_pinned().discard_capture();
|
|
return Poll::Ready(Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
err,
|
|
)
|
|
.into()));
|
|
}
|
|
Ok(Some(Err(err))) => {
|
|
*self = ReadXsoState::Done;
|
|
log_recv(Some(&err), source.as_mut().stream_pinned().take_capture());
|
|
return Poll::Ready(Err(ReadXsoError::Parse(err)));
|
|
}
|
|
Ok(Some(Ok(value))) => {
|
|
*self = ReadXsoState::Done;
|
|
log_recv(None, source.as_mut().stream_pinned().take_capture());
|
|
return Poll::Ready(Ok(value));
|
|
}
|
|
Ok(None) => (),
|
|
}
|
|
}
|
|
|
|
// The error talks about "future", simply because that is
|
|
// where `Self` is used (inside `core::future::Future::poll`).
|
|
ReadXsoState::Done => panic!("future polled after completion"),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Future to read a single XSO from a stream.
|
|
pub(super) struct ReadXso<'x, Io, T: FromXml> {
|
|
/// Stream to read the future from.
|
|
inner: Pin<&'x mut RawXmlStream<Io>>,
|
|
|
|
/// Current state of parsing.
|
|
state: ReadXsoState<T>,
|
|
}
|
|
|
|
impl<'x, Io: AsyncBufRead, T: FromXml> ReadXso<'x, Io, T> {
|
|
/// Start reading a single XSO from a stream.
|
|
pub(super) fn read_from(stream: Pin<&'x mut RawXmlStream<Io>>) -> Self {
|
|
Self {
|
|
inner: stream,
|
|
state: ReadXsoState::PreData,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<Io: AsyncBufRead, T: FromXml> Future for ReadXso<'_, Io, T>
|
|
where
|
|
T::Builder: Unpin,
|
|
{
|
|
type Output = Result<T, ReadXsoError>;
|
|
|
|
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
|
let this = self.get_mut();
|
|
this.state.poll_advance(this.inner.as_mut(), cx)
|
|
}
|
|
}
|
|
|
|
/// Contains metadata from an XML stream header
|
|
#[derive(Default, Debug)]
|
|
pub struct StreamHeader<'x> {
|
|
/// The optional `from` attribute.
|
|
pub from: Option<Cow<'x, str>>,
|
|
|
|
/// The optional `to` attribute.
|
|
pub to: Option<Cow<'x, str>>,
|
|
|
|
/// The optional `id` attribute.
|
|
pub id: Option<Cow<'x, str>>,
|
|
}
|
|
|
|
impl StreamHeader<'_> {
|
|
/// Take the contents and return them as new object.
|
|
///
|
|
/// `self` will be left with all its parts set to `None`.
|
|
pub fn take(&mut self) -> Self {
|
|
Self {
|
|
from: self.from.take(),
|
|
to: self.to.take(),
|
|
id: self.id.take(),
|
|
}
|
|
}
|
|
|
|
pub(super) async fn send<Io: AsyncWrite>(
|
|
self,
|
|
mut stream: Pin<&mut RawXmlStream<Io>>,
|
|
) -> io::Result<()> {
|
|
stream
|
|
.send(Item::XmlDeclaration(rxml::XmlVersion::V1_0))
|
|
.await?;
|
|
stream
|
|
.send(Item::ElementHeadStart(
|
|
Namespace::from(XML_STREAM_NS),
|
|
Cow::Borrowed(xml_ncname!("stream")),
|
|
))
|
|
.await?;
|
|
if let Some(from) = self.from {
|
|
stream
|
|
.send(Item::Attribute(
|
|
Namespace::NONE,
|
|
Cow::Borrowed(xml_ncname!("from")),
|
|
from,
|
|
))
|
|
.await?;
|
|
}
|
|
if let Some(to) = self.to {
|
|
stream
|
|
.send(Item::Attribute(
|
|
Namespace::NONE,
|
|
Cow::Borrowed(xml_ncname!("to")),
|
|
to,
|
|
))
|
|
.await?;
|
|
}
|
|
if let Some(id) = self.id {
|
|
stream
|
|
.send(Item::Attribute(
|
|
Namespace::NONE,
|
|
Cow::Borrowed(xml_ncname!("id")),
|
|
id,
|
|
))
|
|
.await?;
|
|
}
|
|
stream
|
|
.send(Item::Attribute(
|
|
Namespace::NONE,
|
|
Cow::Borrowed(xml_ncname!("version")),
|
|
Cow::Borrowed("1.0"),
|
|
))
|
|
.await?;
|
|
stream.send(Item::ElementHeadEnd).await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
impl StreamHeader<'static> {
|
|
pub(super) async fn recv<Io: AsyncBufRead>(
|
|
mut stream: Pin<&mut RawXmlStream<Io>>,
|
|
) -> io::Result<Self> {
|
|
loop {
|
|
match stream.as_mut().next().await {
|
|
Some(Err(RawError::Io(e))) => return Err(e),
|
|
Some(Err(RawError::SoftTimeout)) => (),
|
|
Some(Ok(Event::StartElement(_, (ns, name), mut attrs))) => {
|
|
if ns != XML_STREAM_NS || name != "stream" {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
"unknown stream header",
|
|
));
|
|
}
|
|
|
|
match attrs.remove(Namespace::none(), "version") {
|
|
Some(v) => {
|
|
if v != "1.0" {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
format!("unsuppored stream version: {}", v),
|
|
));
|
|
}
|
|
}
|
|
None => {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
"required `version` attribute missing",
|
|
))
|
|
}
|
|
}
|
|
|
|
let from = attrs.remove(Namespace::none(), "from");
|
|
let to = attrs.remove(Namespace::none(), "to");
|
|
let id = attrs.remove(Namespace::none(), "id");
|
|
let _ = attrs.remove(Namespace::xml(), "lang");
|
|
|
|
if let Some(((ns, name), _)) = attrs.into_iter().next() {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
format!("unexpected stream header attribute: {{{}}}{}", ns, name),
|
|
));
|
|
}
|
|
|
|
return Ok(StreamHeader {
|
|
from: from.map(Cow::Owned),
|
|
to: to.map(Cow::Owned),
|
|
id: id.map(Cow::Owned),
|
|
});
|
|
}
|
|
Some(Ok(Event::Text(_, _))) | Some(Ok(Event::EndElement(_))) => {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::UnexpectedEof,
|
|
"unexpected content before stream header",
|
|
))
|
|
}
|
|
// We cannot loop infinitely here because the XML parser will
|
|
// prevent more than one XML declaration from being parsed.
|
|
Some(Ok(Event::XmlDeclaration(_, _))) => (),
|
|
None => {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::UnexpectedEof,
|
|
"eof before stream header",
|
|
))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|