use super::channel_source::Source;
use super::publish_link::{Link, open};
use zeroize::Zeroizing;
pub(super) fn publish(
source: &Source,
subject: &str,
text: &str,
key: &str,
) -> Result<u64, String> {
let frame_bytes = 64_usize
.checked_add(subject.len())
.and_then(|n| n.checked_add(text.len()))
.and_then(|n| n.checked_add(key.len()))
.ok_or("the publish frame length overflowed")?;
if frame_bytes > source.record_bytes {
return Err(format!(
"the outgoing wire frame exceeds record_bytes {}",
source.record_bytes
));
}
let credential = source.credential()?;
let mut socket = open(&source.address, source.timeout, source.record_bytes)?;
let mut connect = Zeroizing::new(Vec::new());
connect.extend_from_slice(&[0, 1, 0, 0, 0, 1, 0, 2]);
field(&mut connect, &credential)?;
write(&mut socket, 1, 0, 0, &connect, source.record_bytes)?;
let (kind, flags, stream, bytes) = read(&mut socket, source.record_bytes)?;
if kind == 3 && flags == 0 && stream == 0 {
return Err(refusal(&bytes)?);
}
if kind != 2 || flags != 0 || stream != 0 || bytes.len() != 8 {
return Err("the publish connection received an invalid ConnectAck".to_owned());
}
if bytes
.get(..4)
.is_none_or(|version| !matches!(version, [0, 1, 0, 0..=2]))
{
return Err("the server selected an unsupported publish protocol version".to_owned());
}
let mut envelope = vec![0; 32];
field(&mut envelope, &[0, 0])?;
field(&mut envelope, text.as_bytes())?;
let mut publish = Vec::new();
field(&mut publish, subject.as_bytes())?;
field(&mut publish, &envelope)?;
field(&mut publish, key.as_bytes())?;
write(&mut socket, 9, 2, 1, &publish, source.record_bytes)?;
let (kind, flags, stream, bytes) = read(&mut socket, source.record_bytes)?;
if stream != 1 {
return Err("the publish response names a different stream".to_owned());
}
match kind {
10 if flags & !0x03 == 0 && bytes.len() == if flags & 0x02 == 0 { 8 } else { 16 } => {
let id: [u8; 8] = bytes
.get(..8)
.ok_or("the PublishAck id is absent")?
.try_into()
.map_err(|_| "the PublishAck id is invalid")?;
Ok(u64::from_be_bytes(id))
}
11 if flags == 0 => Err(refusal(&bytes)?),
_ => Err(format!(
"the publish received neither PublishAck nor PublishError (type {kind}, flags {flags}, {} bytes)",
bytes.len()
)),
}
}
fn field(target: &mut Vec<u8>, bytes: &[u8]) -> Result<(), String> {
target.extend_from_slice(
&u32::try_from(bytes.len())
.map_err(|_| "a wire field exceeds u32")?
.to_be_bytes(),
);
target.extend_from_slice(bytes);
Ok(())
}
fn write(
socket: &mut Link,
kind: u8,
flags: u8,
stream: u32,
bytes: &[u8],
bound: usize,
) -> Result<(), String> {
if bytes.len().saturating_add(10) > bound {
return Err(format!(
"the outgoing wire frame exceeds record_bytes {bound}"
));
}
let mut frame = Zeroizing::new(vec![kind, flags]);
frame.extend_from_slice(&stream.to_be_bytes());
frame.extend_from_slice(
&u32::try_from(bytes.len())
.map_err(|_| "a wire frame exceeds u32")?
.to_be_bytes(),
);
frame.extend_from_slice(bytes);
socket.send(&frame)
}
fn read(socket: &mut Link, bound: usize) -> Result<(u8, u8, u32, Vec<u8>), String> {
let mut header = [0; 10];
socket.read_exact(&mut header)?;
let stream = u32::from_be_bytes(
header[2..6]
.try_into()
.map_err(|_| "the wire stream is invalid")?,
);
let size = usize::try_from(u32::from_be_bytes(
header[6..10]
.try_into()
.map_err(|_| "the wire length is invalid")?,
))
.map_err(|_| "the wire length cannot fit usize")?;
if size.saturating_add(10) > bound {
return Err(format!(
"the incoming wire frame exceeds record_bytes {bound}"
));
}
let mut bytes = vec![0; size];
socket.read_exact(&mut bytes)?;
socket.frame_ended()?;
Ok((header[0], header[1], stream, bytes))
}
fn refusal(bytes: &[u8]) -> Result<String, String> {
match bytes.get(2) {
Some(0) if bytes.len() == 3 => Ok("the server refusal carried no detail".to_owned()),
Some(1) => {
let length = u32::from_be_bytes(
bytes
.get(3..7)
.ok_or("the refusal length is absent")?
.try_into()
.map_err(|_| "the refusal length is invalid")?,
);
let size =
usize::try_from(length).map_err(|_| "the refusal length cannot fit usize")?;
if bytes.len() != size.saturating_add(7) {
return Err("the refusal fields are invalid".to_owned());
}
std::str::from_utf8(bytes.get(7..).ok_or("the refusal detail is absent")?)
.map(str::to_owned)
.map_err(|_| "the refusal detail is not UTF-8".to_owned())
}
_ => Err("the server refusal is malformed".to_owned()),
}
}