Merge pull request #577 from swimos/http-lane

Adds support for HTTP lanes.

Author: SirCipher

Cargo.toml

@@ -30,6 +30,7 @@ members = [
     "example_apps/value_downlink",
     "example_apps/map_downlink",
     "example_apps/local_downlink",
+    "example_apps/http_lane",
     "example_apps/tutorial_app",
     "example_apps/tutorial_app/model",
     "example_apps/tutorial_app/generator",
@@ -99,3 +100,6 @@ crossbeam-channel = { version = "0.5" }
 hyper = "0.14"
 lazy_static = "1.4.0"
 percent-encoding = "2.1.0"
+mime = "0.3"
+serde_json = "1.0"
+serde = "1.0"

api/formats/swim_recon/src/parser/mod.rs

@@ -179,6 +179,18 @@ pub fn parse_value(repr: &str, allow_comments: bool) -> Result<Value, ParseError
     parse_recognize(Span::new(repr), allow_comments)
 }
 
+/// Parse exactly one value of type `T` from the input, returning an error if the string does not contain
+/// the representation of exactly one.
+///
+/// * `repr` - The input to parse.
+/// * `allow_comments` - Boolean flag indicating whether or not the parsing should fail on comments.
+pub fn parse_into<T: RecognizerReadable>(
+    repr: &str,
+    allow_comments: bool,
+) -> Result<T, ParseError> {
+    parse_recognize(Span::new(repr), allow_comments)
+}
+
 use crate::comparator::compare_values;
 #[cfg(feature = "async_parser")]
 pub use async_parser::{

api/swim_api/src/agent/mod.rs

@@ -16,25 +16,28 @@ use std::{
     collections::HashMap,
     fmt::{Display, Formatter},
     num::NonZeroUsize,
+    pin::Pin,
+    task::{Context, Poll},
 };
 
 use bytes::Bytes;
-use futures::future::BoxFuture;
+use futures::{future::BoxFuture, ready, Future, FutureExt};
 use swim_model::http::{HttpRequest, HttpResponse};
 use swim_utilities::{
     future::retryable::RetryStrategy,
     io::byte_channel::{ByteReader, ByteWriter},
     non_zero_usize,
     routing::route_uri::RouteUri,
 };
+use thiserror::Error;
 use tokio::sync::{mpsc, oneshot};
 
 use crate::{
     downlink::DownlinkKind,
     error::{
         AgentInitError, AgentRuntimeError, AgentTaskError, DownlinkRuntimeError, OpenStoreError,
     },
-    meta::lane::LaneKind,
+    lane::WarpLaneKind,
     store::StoreKind,
 };
 
@@ -101,21 +104,73 @@ impl Default for StoreConfig {
 
 pub type HttpLaneResponse = HttpResponse<Bytes>;
 
+/// Send half of a single use channel for providing an HTTP response.
+#[derive(Debug)]
+pub struct HttpResponseSender(oneshot::Sender<HttpLaneResponse>);
+
+impl HttpResponseSender {
+    pub fn send(self, response: HttpLaneResponse) -> Result<(), HttpLaneResponse> {
+        self.0.send(response)
+    }
+}
+
+impl Future for HttpResponseReceiver {
+    type Output = Result<HttpLaneResponse, ()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Poll::Ready(ready!(self.as_mut().0.poll_unpin(cx)).map_err(|_| ()))
+    }
+}
+
+pub fn response_channel() -> (HttpResponseSender, HttpResponseReceiver) {
+    let (tx, rx) = oneshot::channel();
+    (HttpResponseSender(tx), HttpResponseReceiver(rx))
+}
+
+/// Receive half of a single use channel for providing an HTTP response.
+#[derive(Debug)]
+pub struct HttpResponseReceiver(oneshot::Receiver<HttpLaneResponse>);
+
+#[derive(Debug, Error, Clone, Copy, PartialEq, Eq, Default)]
+#[error("An HTTP request was dropped before a response was sent.")]
+pub struct ReceiveResponseError;
+
+impl HttpResponseReceiver {
+    pub fn try_recv(&mut self) -> Result<HttpLaneResponse, ReceiveResponseError> {
+        self.0.try_recv().map_err(|_| ReceiveResponseError)
+    }
+}
+
+/// The type of messages sent from the Swim agent runtime to an agent implementation. It includes the
+/// request that was received by the server and a single use channel for the agent implementation to
+/// provide the response.
 #[derive(Debug)]
 pub struct HttpLaneRequest {
     pub request: HttpRequest<Bytes>,
-    pub response_tx: oneshot::Sender<HttpLaneResponse>,
+    response_tx: HttpResponseSender,
 }
 
 impl HttpLaneRequest {
-    pub fn new(
-        request: HttpRequest<Bytes>,
-        response_tx: oneshot::Sender<HttpLaneResponse>,
-    ) -> Self {
-        HttpLaneRequest {
+    /// Create a new instance from an HTTP request and provide the receiver that can be
+    /// used to wait for the response.
+    pub fn new(request: HttpRequest<Bytes>) -> (Self, HttpResponseReceiver) {
+        let (tx, rx) = response_channel();
+        (
+            HttpLaneRequest {
+                request,
+                response_tx: tx,
+            },
+            rx,
+        )
+    }
+
+    /// Split this message into the original request and the channel for sending the response.
+    pub fn into_parts(self) -> (HttpRequest<Bytes>, HttpResponseSender) {
+        let HttpLaneRequest {
             request,
             response_tx,
-        }
+        } = self;
+        (request, response_tx)
     }
 }
 
@@ -136,7 +191,7 @@ pub trait AgentContext: Sync {
     fn add_lane(
         &self,
         name: &str,
-        lane_kind: LaneKind,
+        lane_kind: WarpLaneKind,
         config: LaneConfig,
     ) -> BoxFuture<'static, Result<(ByteWriter, ByteReader), AgentRuntimeError>>;
 

api/swim_api/src/lane/mod.rs

@@ -0,0 +1,78 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt::{Display, Formatter};
+
+use swim_form::structural::Tag;
+
+use crate::{agent::UplinkKind, meta::lane::LaneKind};
+
+/// An enumeration representing the kinds of Warp lanes.
+#[derive(Tag, Debug, PartialEq, Eq, Clone, Copy, Hash)]
+#[form_root(::swim_form)]
+pub enum WarpLaneKind {
+    Command,
+    Demand,
+    DemandMap,
+    Map,
+    JoinMap,
+    JoinValue,
+    Supply,
+    Spatial,
+    Value,
+}
+
+impl From<WarpLaneKind> for LaneKind {
+    fn from(value: WarpLaneKind) -> Self {
+        match value {
+            WarpLaneKind::Command => LaneKind::Command,
+            WarpLaneKind::Demand => LaneKind::Demand,
+            WarpLaneKind::DemandMap => LaneKind::DemandMap,
+            WarpLaneKind::Map => LaneKind::Map,
+            WarpLaneKind::JoinMap => LaneKind::JoinMap,
+            WarpLaneKind::JoinValue => LaneKind::JoinValue,
+            WarpLaneKind::Supply => LaneKind::Supply,
+            WarpLaneKind::Spatial => LaneKind::Spatial,
+            WarpLaneKind::Value => LaneKind::Value,
+        }
+    }
+}
+
+impl WarpLaneKind {
+    pub fn map_like(&self) -> bool {
+        matches!(
+            self,
+            WarpLaneKind::Map
+                | WarpLaneKind::DemandMap
+                | WarpLaneKind::JoinMap
+                | WarpLaneKind::JoinValue
+        )
+    }
+
+    pub fn uplink_kind(&self) -> UplinkKind {
+        match self {
+            WarpLaneKind::Map | WarpLaneKind::DemandMap | WarpLaneKind::JoinMap => UplinkKind::Map,
+            WarpLaneKind::Supply => UplinkKind::Supply,
+            WarpLaneKind::Spatial => todo!("Spatial uplinks not supported."),
+            _ => UplinkKind::Value,
+        }
+    }
+}
+
+impl Display for WarpLaneKind {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let as_str: &str = self.as_ref();
+        write!(f, "{}", as_str)
+    }
+}

api/swim_api/src/lib.rs

@@ -16,6 +16,7 @@ pub mod agent;
 pub mod downlink;
 pub mod error;
 pub mod handlers;
+pub mod lane;
 pub mod meta;
 pub mod net;
 pub mod protocol;

api/swim_api/src/meta/lane.rs

@@ -26,13 +26,10 @@ use swim_form::structural::Tag;
 use swim_form::Form;
 use swim_model::{Text, ValueKind};
 
-use crate::agent::UplinkKind;
-
 /// An enumeration representing the type of a lane.
 #[derive(Tag, Debug, PartialEq, Eq, Clone, Copy, Hash)]
 #[form_root(::swim_form)]
 pub enum LaneKind {
-    Action,
     Command,
     Demand,
     DemandMap,
@@ -42,6 +39,7 @@ pub enum LaneKind {
     Supply,
     Spatial,
     Value,
+    Http,
 }
 
 pub enum ValueLikeLaneKind {
@@ -59,15 +57,6 @@ impl LaneKind {
             LaneKind::Map | LaneKind::DemandMap | LaneKind::JoinMap | LaneKind::JoinValue
         )
     }
-
-    pub fn uplink_kind(&self) -> UplinkKind {
-        match self {
-            LaneKind::Map | LaneKind::DemandMap | LaneKind::JoinMap => UplinkKind::Map,
-            LaneKind::Supply => UplinkKind::Supply,
-            LaneKind::Spatial => todo!("Spatial uplinks not supported."),
-            _ => UplinkKind::Value,
-        }
-    }
 }
 
 /// Lane information metadata that can be retrieved when syncing to
@@ -105,7 +94,6 @@ impl<'a> TryFrom<&'a str> for LaneKind {
 
     fn try_from(value: &'a str) -> Result<Self, Self::Error> {
         match value {
-            "Action" => Ok(LaneKind::Action),
             "Command" => Ok(LaneKind::Command),
             "Demand" => Ok(LaneKind::Demand),
             "DemandMap" => Ok(LaneKind::DemandMap),

api/swim_api/src/protocol/mod.rs

@@ -91,7 +91,7 @@ fn write_recon_with_len<T: StructuralWritable>(dst: &mut BytesMut, body: &T) {
     rewound.put_u64(body_len as u64);
 }
 
-fn write_recon<T: StructuralWritable>(dst: &mut BytesMut, body: &T) -> usize {
+pub fn write_recon<T: StructuralWritable>(dst: &mut BytesMut, body: &T) -> usize {
     let body_offset = write_recon_body(dst, body);
     dst.remaining() - body_offset
 }

api/swim_model/src/http/header.rs

@@ -12,15 +12,23 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::str::FromStr;
+use std::{
+    cmp::Ordering,
+    hash::{Hash, Hasher},
+    str::FromStr,
+};
 
 use bytes::Bytes;
 
 use crate::BytesStr;
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+/// Model for the name of an HTTP header. The representation of this type will either be an enumeration
+/// of the standard header names or a general ASCII string for custom headers.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct HeaderName(Name);
 
+/// Model for the value of an HTTP header. The representation of this type is an array of bytes that
+/// may, or may not, be a valid UTF8 string.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct HeaderValue(HeaderValueInner);
 
@@ -30,12 +38,27 @@ enum HeaderValueInner {
     BytesHeader(Bytes),
 }
 
+/// Mode of an HTTP header, used by [`super::HttpRequest`] and [`super::HttpResponse`].
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Header {
     pub name: HeaderName,
     pub value: HeaderValue,
 }
 
+impl Header {
+    /// Create a header from anything that can be converted into header names and header values.
+    pub fn new<N, V>(name: N, value: V) -> Self
+    where
+        N: Into<HeaderName>,
+        V: Into<HeaderValue>,
+    {
+        Header {
+            name: name.into(),
+            value: value.into(),
+        }
+    }
+}
+
 impl HeaderName {
     pub fn as_str(&self) -> &str {
         self.0.str_value()
@@ -182,6 +205,7 @@ impl From<&str> for HeaderName {
     }
 }
 
+/// An enumeration of standard header names.
 #[repr(u8)]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum StandardHeaderName {
@@ -233,6 +257,33 @@ impl Name {
     }
 }
 
+impl PartialOrd for Name {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        self.str_value().partial_cmp(other.str_value())
+    }
+}
+
+impl Ord for Name {
+    fn cmp(&self, other: &Self) -> Ordering {
+        self.str_value().cmp(other.str_value())
+    }
+}
+
+impl Hash for Name {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.str_value().hash(state)
+    }
+}
+
+impl PartialEq<StandardHeaderName> for HeaderName {
+    fn eq(&self, other: &StandardHeaderName) -> bool {
+        match &self.0 {
+            Name::Standard(h) => h == other,
+            Name::Other(_) => false,
+        }
+    }
+}
+
 impl StandardHeaderName {
     fn try_from_basic(s: &str) -> Option<Self> {
         match s {