#![cfg(all(feature = "storage", feature = "serde"))]
use core::net::Ipv4Addr;
use core::num::{NonZeroU32, NonZeroU64};
use core::time::Duration;
use pipecircuit::storage::warp_table::observed::{
SourceId, ViewAdmission, ViewApplyOutcome, ViewChange, ViewDefinition, ViewDelivery, ViewId,
ViewReplica, ViewRevision,
};
use pipecircuit::storage::warp_table::{Capacity, WarpTable, WarpTableRow};
use serde::{Deserialize, Serialize};
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
#[derive(WarpTableRow)]
#[index(primary = u32)]
struct Actor {
position: u32,
}
struct FullCopy;
impl ViewDefinition<Actor> for FullCopy {
type Value = u32;
fn project(&self, row: ActorRef<'_>) -> Option<u32> {
Some(*row.position)
}
}
#[derive(Serialize, Deserialize)]
enum Packet {
Begin {
through: ViewRevision,
rows: usize,
},
Rows {
through: ViewRevision,
changes: Vec<ViewChange<u32, u32>>,
},
End {
through: ViewRevision,
},
Changes {
base: ViewRevision,
through: ViewRevision,
changes: Vec<ViewChange<u32, u32>>,
},
Finish,
}
const FRAME_BYTES: usize = 4096;
const IO_TIMEOUT: Duration = Duration::from_secs(5);
fn send(stream: &mut TcpStream, packet: &Packet) -> Result<(), &'static str> {
let mut frame = [0_u8; FRAME_BYTES];
let encoded = postcard::to_slice(packet, &mut frame).map_err(|_| "encode")?;
let length = u32::try_from(encoded.len()).map_err(|_| "length")?;
stream
.write_all(&length.to_be_bytes())
.map_err(|_| "write length")?;
for chunk in encoded.chunks(3) {
stream.write_all(chunk).map_err(|_| "write payload")?;
}
Ok(())
}
fn receive(stream: &mut TcpStream) -> Result<Packet, &'static str> {
let mut length = [0; 4];
stream.read_exact(&mut length).map_err(|_| "read length")?;
let length = usize::try_from(u32::from_be_bytes(length)).map_err(|_| "length")?;
let mut frame = [0_u8; FRAME_BYTES];
let payload = frame.get_mut(..length).ok_or("frame bound")?;
stream.read_exact(payload).map_err(|_| "read payload")?;
postcard::from_bytes(payload).map_err(|_| "decode")
}
#[test]
fn a_foreign_tcp_receiver_keeps_a_full_copy_through_coalesced_updates() -> Result<(), &'static str>
{
let capacity = Capacity::new(NonZeroU32::new(4).ok_or("capacity")?);
let source_id = SourceId::new(NonZeroU64::MIN);
let view_id = ViewId::new(NonZeroU64::MIN);
let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).map_err(|_| "listen")?;
let mut writer =
TcpStream::connect(listener.local_addr().map_err(|_| "address")?).map_err(|_| "connect")?;
writer
.set_write_timeout(Some(IO_TIMEOUT))
.map_err(|_| "write timeout")?;
let receiver = std::thread::spawn(move || -> Result<Vec<(u32, u32)>, &'static str> {
let (mut stream, _) = listener.accept().map_err(|_| "accept")?;
stream
.set_read_timeout(Some(IO_TIMEOUT))
.map_err(|_| "read timeout")?;
let mut replica = ViewReplica::<u32, u32>::try_new(source_id, view_id, capacity)
.map_err(|_| "replica")?;
loop {
match receive(&mut stream)? {
Packet::Begin { through, rows } => {
assert_eq!(
replica.begin_baseline(through, rows),
Ok(ViewApplyOutcome::BaselineRequired)
);
}
Packet::Rows { through, changes } => replica
.baseline_chunk(through, &changes)
.map_err(|_| "chunk")?,
Packet::End { through } => assert_eq!(
replica.finish_baseline(through),
Ok(ViewApplyOutcome::Applied)
),
Packet::Changes {
base,
through,
changes,
} => assert_eq!(
replica.apply_changes(base, through, &changes),
Ok(ViewApplyOutcome::Applied)
),
Packet::Finish => {
let mut values: Vec<_> =
replica.rows().map(|row| (row.id, *row.value)).collect();
values.sort_unstable();
return Ok(values);
}
}
}
});
let table = WarpTable::<Actor>::try_new(capacity).map_err(|_| "table")?;
let mut source = table.observe(source_id).map_err(|_| "source")?;
for id in 0..4 {
assert!(source.insert_with_id(id, Actor { position: 0 }).is_ok());
}
let mut view = source
.maintain_with_retention(
view_id,
FullCopy,
Capacity::new(NonZeroU32::new(16).ok_or("retention")?),
)
.map_err(|_| "view")?;
let mut buffer = view.delivery_buffer().map_err(|_| "buffer")?;
let mut subscription = view.subscribe();
view.transmit(&mut subscription, &mut buffer, |delivery| {
let ViewDelivery::Baseline { through, rows } = delivery else {
return Err("baseline");
};
send(
&mut writer,
&Packet::Begin {
through,
rows: rows.len(),
},
)?;
for row in rows {
send(
&mut writer,
&Packet::Rows {
through,
changes: vec![ViewChange::Upsert {
id: row.id,
incarnation: row.incarnation,
value: *row.value,
}],
},
)?;
}
send(&mut writer, &Packet::End { through })?;
Ok(ViewAdmission::Admitted)
})
.map_err(|_| "initial transmission")?;
for position in 1..=4 {
for id in 0..4 {
assert!(
source
.update(id, |mut row| *row.position_mut() = position)
.is_ok()
);
}
assert!(view.refresh(&source).is_ok());
}
view.transmit(&mut subscription, &mut buffer, |delivery| {
let ViewDelivery::Changes {
base,
through,
changes,
} = delivery
else {
return Err("incremental delivery");
};
assert_eq!(changes.len(), 4);
send(
&mut writer,
&Packet::Changes {
base,
through,
changes: changes.to_vec(),
},
)?;
Ok(ViewAdmission::Admitted)
})
.map_err(|_| "incremental transmission")?;
send(&mut writer, &Packet::Finish)?;
let result = receiver.join().map_err(|_| "receiver panicked")??;
assert_eq!(result, [(0, 4), (1, 4), (2, 4), (3, 4)]);
Ok(())
}