liminal-rs 0.16.2

//! An ordinary single durable publication stores the causal parent its live
//! delivery carries (R3b independent review, row 13:
//! `durable_publish.rs:305@ddf6fae` stored `causal_context: None`).

use std::error::Error;
use std::sync::Arc;

use super::{ChannelConfig, ChannelHandle, ChannelMode, ChannelSupervisor, Schema};
use crate::causal::{CausalContext, MessageId};
use crate::durability::bridge::block_on;
use crate::durability::{DurableStore, MessageEnvelope, open_ephemeral};
use crate::envelope::PublisherId;

#[test]
fn a_single_publication_stores_the_parent_its_live_delivery_carries() -> Result<(), Box<dyn Error>>
{
    let supervisor = ChannelSupervisor::new()?;
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let config = ChannelConfig::new(
        "single.parent".to_owned(),
        Schema::new(serde_json::json!({"type": "object"}))?,
        ChannelMode::Durable,
    );
    let channel =
        ChannelHandle::new_durable_with_supervisor(config, Arc::clone(&store), supervisor.clone())?;
    let subscription = channel.subscribe()?;
    let parent = MessageId::new();

    channel.publish_with_context(
        br#"{"child":1}"#,
        PublisherId::new("parented"),
        Some(CausalContext::from_parent_chain(vec![parent])),
    )?;

    let live = subscription
        .try_next()?
        .ok_or("no live delivery by the time publish returned")?;
    assert_eq!(
        live.causal_context.and_then(|context| context.parent),
        Some(parent),
        "the live delivery lost its parent"
    );
    let row =
        block_on(store.read_at("single.parent:0", 0))??.ok_or("the published row is absent")?;
    let stored = MessageEnvelope::deserialize(&row.payload)?;
    assert_eq!(
        stored.causal_context.and_then(|context| context.parent_id),
        Some(parent.as_uuid().to_string()),
        "the stored row lost the parent its live delivery carried"
    );
    supervisor.shutdown();
    Ok(())
}