290 lines
10 KiB
Rust
290 lines
10 KiB
Rust
#![cfg(feature = "http")]
|
|
//! The same stateful scenario runs through direct DTOs and the HTTP facade.
|
|
use std::collections::BTreeMap;
|
|
use wip_client::direct::{Client as DirectClient, Request, Response};
|
|
use wip_client::{
|
|
CallOutcome, Client, ClientLimits, Completion, InterfaceObservation, ObjectObservation,
|
|
RequestId, SecurityContext, SessionId,
|
|
};
|
|
use wip_http::{
|
|
Limits, decode_call_operation_request, decode_fetch_interface_request, decode_observe_request,
|
|
encode_call_operation_response, encode_fetch_interface_response, encode_observe_response,
|
|
encode_protocol_error_response,
|
|
};
|
|
use wip_protocol::{
|
|
CallOperationResponse, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor,
|
|
InterfaceReference, Object, ObjectObservation as ProtocolObservation, OperationDeclaration,
|
|
ProtocolError, ProtocolErrorCode, ProtocolInteraction, ReturnDeclaration, TypeExpr, Value,
|
|
};
|
|
|
|
fn limits() -> Limits {
|
|
Limits::new(64 * 1024, 64 * 1024, 64).unwrap()
|
|
}
|
|
fn reference() -> InterfaceReference {
|
|
InterfaceReference {
|
|
scope: "/scope".into(),
|
|
name: "math".into(),
|
|
}
|
|
}
|
|
fn descriptor() -> InterfaceDescriptor {
|
|
InterfaceDescriptor {
|
|
format: INTERFACE_FORMAT_V1.into(),
|
|
documentation: None,
|
|
types: vec![],
|
|
operations: vec![OperationDeclaration {
|
|
name: "run".into(),
|
|
documentation: None,
|
|
parameters: vec![],
|
|
returns: ReturnDeclaration {
|
|
documentation: None,
|
|
r#type: TypeExpr::Integer,
|
|
},
|
|
}],
|
|
}
|
|
}
|
|
fn observation(name: &str, identity: &str) -> ProtocolObservation {
|
|
ProtocolObservation {
|
|
object: Object {
|
|
name: name.into(),
|
|
description: None,
|
|
interfaces: vec![reference()],
|
|
r#ref: Some(identity.into()),
|
|
validator: None,
|
|
},
|
|
children: None,
|
|
}
|
|
}
|
|
fn interface(identity: &str) -> FetchInterfaceResponse {
|
|
FetchInterfaceResponse {
|
|
interface: reference(),
|
|
scope_ref: Some(identity.into()),
|
|
validator: None,
|
|
descriptor: descriptor(),
|
|
}
|
|
}
|
|
|
|
enum Channel {
|
|
Direct(DirectClient),
|
|
Http(Client),
|
|
}
|
|
impl Channel {
|
|
fn core(&self) -> &DirectClient {
|
|
match self {
|
|
Self::Direct(client) => client,
|
|
Self::Http(client) => client,
|
|
}
|
|
}
|
|
fn observe(&mut self, session: &SessionId, path: &str) -> (RequestId, Request) {
|
|
match self {
|
|
Self::Direct(client) => {
|
|
let p = client.refresh_object(session, path).unwrap();
|
|
(p.id, p.request)
|
|
}
|
|
Self::Http(client) => {
|
|
let p = client.refresh_object(session, path).unwrap();
|
|
(
|
|
p.id,
|
|
Request::Observe(decode_observe_request(p.request.body(), limits()).unwrap()),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
fn fetch(&mut self, session: &SessionId) -> (RequestId, Request) {
|
|
match self {
|
|
Self::Direct(client) => {
|
|
let p = client.prepare_interface(session, reference()).unwrap();
|
|
(p.id, p.request)
|
|
}
|
|
Self::Http(client) => {
|
|
let p = client.prepare_interface(session, reference()).unwrap();
|
|
(
|
|
p.id,
|
|
Request::FetchInterface(
|
|
decode_fetch_interface_request(p.request.body(), limits()).unwrap(),
|
|
),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
fn call(&mut self, session: &SessionId) -> (RequestId, Request) {
|
|
match self {
|
|
Self::Direct(client) => {
|
|
let p = client
|
|
.prepare_call(session, "/scope/item", &reference(), "run", BTreeMap::new())
|
|
.unwrap();
|
|
client.mark_dispatched(p.id).unwrap();
|
|
(p.id, p.request)
|
|
}
|
|
Self::Http(client) => {
|
|
let p = client
|
|
.prepare_call(session, "/scope/item", &reference(), "run", BTreeMap::new())
|
|
.unwrap();
|
|
client.mark_dispatched(p.id).unwrap();
|
|
(
|
|
p.id,
|
|
Request::CallOperation(
|
|
decode_call_operation_request(p.request.body(), &descriptor(), limits())
|
|
.unwrap(),
|
|
),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
fn complete(&mut self, pending: (RequestId, Request), result: Response) -> Completion {
|
|
let (id, request) = pending;
|
|
match self {
|
|
Self::Direct(client) => client.complete(id, result).unwrap(),
|
|
Self::Http(client) => {
|
|
let wire = match (request, result) {
|
|
(Request::Observe(request), Response::Observe(Ok(value))) => {
|
|
encode_observe_response(&request, &value, limits())
|
|
}
|
|
(Request::FetchInterface(request), Response::FetchInterface(Ok(value))) => {
|
|
encode_fetch_interface_response(&request, &value, limits())
|
|
}
|
|
(Request::CallOperation(request), Response::CallOperation(Ok(value))) => {
|
|
encode_call_operation_response(&request, &descriptor(), &value, limits())
|
|
}
|
|
(_, Response::Observe(Err(error))) => encode_protocol_error_response(
|
|
ProtocolInteraction::Observe,
|
|
&error,
|
|
limits(),
|
|
),
|
|
(_, Response::FetchInterface(Err(error))) => encode_protocol_error_response(
|
|
ProtocolInteraction::FetchInterface,
|
|
&error,
|
|
limits(),
|
|
),
|
|
(_, Response::CallOperation(Err(error))) => encode_protocol_error_response(
|
|
ProtocolInteraction::CallOperation,
|
|
&error,
|
|
limits(),
|
|
),
|
|
_ => panic!("test interaction mismatch"),
|
|
}
|
|
.unwrap();
|
|
client.complete(id, wire).unwrap()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, PartialEq)]
|
|
struct Snapshot {
|
|
objects: Vec<ObjectObservation>,
|
|
interface: Option<InterfaceObservation>,
|
|
outcomes: Vec<CallOutcome>,
|
|
requests: Vec<Request>,
|
|
}
|
|
fn scenario(http: bool) -> Snapshot {
|
|
let state_limits = ClientLimits::new(4, 16, 8, 8, 8).unwrap();
|
|
let (mut channel, session) = if http {
|
|
let mut client = Client::new(state_limits, limits());
|
|
let session = client
|
|
.open_session("https://example.test/wip", SecurityContext::new("alice"))
|
|
.unwrap();
|
|
(Channel::Http(client), session)
|
|
} else {
|
|
let mut client = DirectClient::new(state_limits);
|
|
let session = client
|
|
.open_session("local world", SecurityContext::new("alice"))
|
|
.unwrap();
|
|
(Channel::Direct(client), session)
|
|
};
|
|
let mut requests = vec![];
|
|
for (path, name, identity) in [
|
|
("/scope", "scope", "scope-v1"),
|
|
("/scope/item", "item", "item-v1"),
|
|
] {
|
|
let p = channel.observe(&session, path);
|
|
requests.push(p.1.clone());
|
|
channel.complete(p, Response::Observe(Ok(observation(name, identity))));
|
|
}
|
|
let p = channel.fetch(&session);
|
|
requests.push(p.1.clone());
|
|
channel.complete(p, Response::FetchInterface(Ok(interface("scope-v1"))));
|
|
let p = channel.call(&session);
|
|
requests.push(p.1.clone());
|
|
channel.complete(
|
|
p,
|
|
Response::CallOperation(Ok(CallOperationResponse {
|
|
result: Value::Integer(42),
|
|
validator: Some(vec![2]),
|
|
})),
|
|
);
|
|
let p = channel.call(&session);
|
|
requests.push(p.1.clone());
|
|
channel.complete(
|
|
p,
|
|
Response::CallOperation(Err(ProtocolError {
|
|
code: ProtocolErrorCode::InterfaceMismatch,
|
|
message: "changed".into(),
|
|
})),
|
|
);
|
|
let p = channel.observe(&session, "/scope/item");
|
|
channel.complete(p, Response::Observe(Ok(observation("item", "item-v1"))));
|
|
let p = channel.fetch(&session);
|
|
channel.complete(p, Response::FetchInterface(Ok(interface("scope-v1"))));
|
|
let old = channel.fetch(&session);
|
|
let p = channel.observe(&session, "/scope");
|
|
channel.complete(p, Response::Observe(Ok(observation("scope", "scope-v2"))));
|
|
let old_id = old.0;
|
|
assert_eq!(
|
|
channel.complete(old, Response::FetchInterface(Ok(interface("scope-v1")))),
|
|
Completion::StaleResponseRejected(old_id)
|
|
);
|
|
let p = channel.fetch(&session);
|
|
channel.complete(p, Response::FetchInterface(Ok(interface("scope-v2"))));
|
|
let p = channel.call(&session);
|
|
requests.push(p.1.clone());
|
|
channel.complete(
|
|
p,
|
|
Response::CallOperation(Err(ProtocolError {
|
|
code: ProtocolErrorCode::OperationOutcomeUnknown,
|
|
message: "unknown".into(),
|
|
})),
|
|
);
|
|
assert_eq!(channel.core().in_flight_len(), 0);
|
|
Snapshot {
|
|
objects: channel.core().known_space(&session).unwrap(),
|
|
interface: channel.core().interface(&session, &reference()).cloned(),
|
|
outcomes: channel
|
|
.core()
|
|
.call_history(&session)
|
|
.unwrap()
|
|
.iter()
|
|
.map(|record| record.outcome.clone())
|
|
.collect(),
|
|
requests,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn direct_and_http_agree_on_requests_state_calls_scope_invalidation_and_uncertainty() {
|
|
assert_eq!(scenario(false), scenario(true));
|
|
}
|
|
|
|
#[test]
|
|
fn http_endpoint_and_identical_direct_identity_do_not_share_cache() {
|
|
let mut client = Client::new(ClientLimits::new(4, 16, 8, 8, 8).unwrap(), limits());
|
|
let http = client
|
|
.open_session("https://example.test/wip", SecurityContext::new("alice"))
|
|
.unwrap();
|
|
let direct = DirectClient::open_session(
|
|
&mut client,
|
|
"https://example.test/wip/",
|
|
SecurityContext::new("alice"),
|
|
)
|
|
.unwrap();
|
|
assert_ne!(http, direct);
|
|
let request = DirectClient::refresh_object(&mut client, &direct, "/scope").unwrap();
|
|
DirectClient::complete(
|
|
&mut client,
|
|
request.id,
|
|
Response::Observe(Ok(observation("scope", "local"))),
|
|
)
|
|
.unwrap();
|
|
assert!(client.object(&http, "/scope").is_none());
|
|
assert!(client.refresh_object(&direct, "/scope").is_err());
|
|
}
|