use super::{
DeliveredId, DisconnectReason, Frame, WorkerActivityDescriptor, WorkerRegisterOutcome,
WorkerRegistration,
};
use crate::protocol::{
CausalContext, MessageEnvelope, MessageId, ProtocolVersion, SchemaId, decode, encode,
encoded_len,
};
const FIXTURE: &str = include_str!("../../../../../docs/wire/channel-frames.json");
#[test]
fn channel_frames_match_the_committed_rust_encoder_bytes() -> Result<(), Box<dyn std::error::Error>>
{
let fixture: serde_json::Value = serde_json::from_str(FIXTURE)?;
let rows = fixture["frames"]
.as_array()
.ok_or("fixture frames must be an array")?;
let frames = frames();
assert_eq!(rows.len(), frames.len());
for (row, (name, frame)) in rows.iter().zip(frames) {
assert_eq!(row["name"].as_str(), Some(name));
let hex = row["frame_hex"]
.as_str()
.ok_or("frame hex must be a string")?;
let expected = hex_bytes(hex)?;
let mut actual = vec![0; encoded_len(&frame)?];
let written = encode(&frame, &mut actual)?;
assert_eq!(written, actual.len());
assert_eq!(actual, expected, "{name} bytes drifted");
let (decoded, consumed) = decode(&expected)?;
assert_eq!(consumed, expected.len());
assert_eq!(decoded, frame, "{name} meaning drifted");
}
Ok(())
}
fn hex_bytes(hex: &str) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
if !hex.is_ascii() || !hex.len().is_multiple_of(2) {
return Err("fixture hex must be even ASCII".into());
}
(0..hex.len())
.step_by(2)
.map(|i| u8::from_str_radix(&hex[i..i + 2], 16).map_err(Into::into))
.collect()
}
fn frames() -> Vec<(&'static str, Frame)> {
let envelope = MessageEnvelope::new(
SchemaId::new([0x11; 32]),
CausalContext {
parent_id: Some(MessageId::new("7:9")),
vector_clock_entry: Some(17),
},
vec![0, 255, 65],
);
let mut out = Vec::new();
out.extend(control_frames());
out.extend(subscription_frames());
out.extend(publish_frames(&envelope));
out.extend(conversation_frames(&envelope));
out.extend(pressure_and_push_frames());
out.extend(worker_frames());
out.extend(delivery_and_close_frames(&envelope));
out
}
fn control_frames() -> Vec<(&'static str, Frame)> {
vec![
(
"Connect",
Frame::Connect {
flags: 0,
min_version: ProtocolVersion::new(1, 0),
max_version: ProtocolVersion::new(1, 3),
auth_token: b"\x02fixture-pass".to_vec(),
},
),
(
"ConnectAck",
Frame::ConnectAck {
flags: 0,
selected_version: ProtocolVersion::new(1, 3),
capabilities: 1,
},
),
(
"ConnectError",
Frame::ConnectError {
flags: 0,
reason_code: 2,
message: Some("authentication refused".to_owned()),
},
),
("Disconnect", Frame::Disconnect { flags: 0 }),
]
}
fn subscription_frames() -> Vec<(&'static str, Frame)> {
vec![
(
"Subscribe",
Frame::Subscribe {
flags: 0,
stream_id: 1,
channel: "fixture.records".to_owned(),
accepted_schemas: vec![SchemaId::new([0x11; 32])],
max_in_flight: 8,
},
),
(
"SubscribeAck",
Frame::SubscribeAck {
flags: 0,
stream_id: 1,
subscription_id: 7,
selected_schema: SchemaId::new([0x11; 32]),
},
),
(
"SubscribeError",
Frame::SubscribeError {
flags: 0,
stream_id: 1,
reason_code: 3,
message: Some("subscribe refused".to_owned()),
},
),
(
"Unsubscribe",
Frame::Unsubscribe {
flags: 0,
stream_id: 1,
subscription_id: 7,
},
),
]
}
fn publish_frames(envelope: &MessageEnvelope) -> Vec<(&'static str, Frame)> {
vec![
(
"Publish",
Frame::Publish {
flags: 0,
stream_id: 1,
channel: "fixture.records".to_owned(),
envelope: envelope.clone(),
idempotency_key: None,
},
),
(
"Publish.WithIdempotencyKey",
Frame::Publish {
flags: 2,
stream_id: 1,
channel: "fixture.records".to_owned(),
envelope: envelope.clone(),
idempotency_key: Some("fixture-key".to_owned()),
},
),
(
"PublishAck",
Frame::PublishAck {
flags: 1,
stream_id: 1,
message_id: 9,
durable_position: None,
},
),
(
"PublishAck.DurablePosition",
Frame::PublishAck {
flags: 3,
stream_id: 1,
message_id: 9,
durable_position: Some(12),
},
),
(
"PublishError",
Frame::PublishError {
flags: 0,
stream_id: 1,
reason_code: 4,
message: Some("publish refused".to_owned()),
},
),
]
}
fn conversation_frames(envelope: &MessageEnvelope) -> Vec<(&'static str, Frame)> {
vec![
(
"ConversationOpen",
Frame::ConversationOpen {
flags: 0,
stream_id: 1,
conversation_id: 7,
subject: "fixture.echo".to_owned(),
},
),
(
"ConversationMessage",
Frame::ConversationMessage {
flags: 1,
stream_id: 1,
conversation_id: 7,
envelope: envelope.clone(),
},
),
(
"ConversationClose",
Frame::ConversationClose {
flags: 0,
stream_id: 1,
conversation_id: 7,
reason_code: Some(5),
message: Some("worker gone".to_owned()),
},
),
(
"ConversationError",
Frame::ConversationError {
flags: 0,
stream_id: 1,
conversation_id: 7,
reason_code: 6,
message: Some("call refused".to_owned()),
},
),
]
}
fn pressure_and_push_frames() -> Vec<(&'static str, Frame)> {
vec![
(
"Accept",
Frame::Accept {
flags: 0,
stream_id: 1,
referenced_message_id: MessageId::new("7:9"),
},
),
(
"Defer",
Frame::Defer {
flags: 0,
stream_id: 1,
referenced_message_id: MessageId::new("7:9"),
reason: Some("deferred".to_owned()),
},
),
(
"Reject",
Frame::Reject {
flags: 0,
stream_id: 1,
referenced_message_id: MessageId::new("7:9"),
reason: Some("rejected".to_owned()),
},
),
("Ping", Frame::Ping { flags: 0 }),
("Pong", Frame::Pong { flags: 0 }),
(
"Push",
Frame::Push {
flags: 0,
stream_id: 1,
correlation_id: 9,
payload: vec![0, 255, 65],
},
),
(
"PushReply",
Frame::PushReply {
flags: 0,
stream_id: 1,
correlation_id: 9,
payload: vec![0, 255, 65],
},
),
]
}
fn worker_frames() -> Vec<(&'static str, Frame)> {
vec![
(
"WorkerRegister",
Frame::WorkerRegister {
flags: 0,
registration: WorkerRegistration {
namespaces: vec!["fixture".to_owned()],
task_queue: "echo".to_owned(),
node: Some("local".to_owned()),
activity_types: vec!["echo".to_owned()],
identity: "fixture-worker".to_owned(),
activities: vec![WorkerActivityDescriptor {
name: "echo".to_owned(),
input_schema_json: r#"{"type":"object"}"#.to_owned(),
output_schema_json: r#"{"type":"object"}"#.to_owned(),
}],
},
},
),
(
"WorkerRegisterAck.Accepted",
Frame::WorkerRegisterAck {
flags: 0,
outcome: WorkerRegisterOutcome::Accepted,
},
),
(
"WorkerRegisterAck.Rejected",
Frame::WorkerRegisterAck {
flags: 0,
outcome: WorkerRegisterOutcome::Rejected {
reason: "worker refused".to_owned(),
},
},
),
]
}
fn delivery_and_close_frames(envelope: &MessageEnvelope) -> Vec<(&'static str, Frame)> {
vec![
(
"Deliver",
Frame::Deliver {
flags: 0,
stream_id: 1,
delivery_seq: 1,
envelope: envelope.clone(),
message: None,
},
),
(
"Deliver.MessageId",
Frame::Deliver {
flags: 1,
stream_id: 1,
delivery_seq: 1,
envelope: envelope.clone(),
message: Some(DeliveredId { boot: 7, id: 9 }),
},
),
(
"DisconnectReason.GrantRevoked",
Frame::DisconnectReason {
flags: 0,
reason: DisconnectReason::GrantRevoked,
},
),
(
"DisconnectReason.RevocationLogReset",
Frame::DisconnectReason {
flags: 0,
reason: DisconnectReason::RevocationLogReset,
},
),
]
}