Struct Stream 

pub struct Stream { /* private fields */ }

Priority data stream entity

Implementations

impl Stream

pub fn new( session_id: Id<SessionMarker>, source_data: JsonData, config: StreamConfig, ) -> Stream

Create new stream

pub fn id(&self) -> Id<StreamMarker>

Get stream ID

pub fn session_id(&self) -> Id<SessionMarker>

Get session ID

pub fn state(&self) -> &StreamState

Get current state

pub fn config(&self) -> &StreamConfig

Get configuration

pub fn stats(&self) -> &StreamStats

Get statistics

pub fn created_at(&self) -> DateTime<Utc>

Get creation timestamp

pub fn updated_at(&self) -> DateTime<Utc>

Get last update timestamp

pub fn completed_at(&self) -> Option<DateTime<Utc>>

Get completion timestamp

pub fn source_data(&self) -> Option<&JsonData>

Get source data

pub fn metadata(&self) -> &HashMap<String, String>

Get metadata

pub fn add_metadata(&mut self, key: String, value: String)

Add metadata

pub fn start_streaming(&mut self) -> Result<(), DomainError>

Start streaming (transition to Streaming state)

pub fn complete(&mut self) -> Result<(), DomainError>

Complete stream successfully

pub fn fail(&mut self, error: String) -> Result<(), DomainError>

Fail stream with error

pub fn cancel(&mut self) -> Result<(), DomainError>

Cancel stream

pub fn create_skeleton_frame(&mut self) -> Result<Frame, DomainError>

Generate skeleton frame for the stream

pub fn create_patch_frames( &mut self, priority_threshold: Priority, max_frames: usize, ) -> Result<Vec<Frame>, DomainError>

Create batch of patch frames based on priority

pub fn extract_prioritized_patches( &self, priority_threshold: Priority, ) -> Result<Vec<(FramePatch, Priority)>, DomainError>

Compute prioritized patches for this stream’s current source_data, without mutating any state — the expensive half of Self::create_patch_frames (a full traversal of source_data, computing a priority for every leaf value).

Split out from Self::commit_patch_frames so a caller that needs its own concurrency control around the mutating half (e.g. a repository holding a per-session lock for the read-modify-write) can run this traversal lock-free and only take the lock for the commit step — which is itself O(patches.len()), not cheap or max_frames-bounded; see Self::commit_patch_frames’s docs. Safe to call without holding any lock: source_data is set once at stream creation and never mutated afterward.

pub fn commit_patch_frames( &mut self, patches: Vec<(FramePatch, Priority)>, max_frames: usize, ) -> Result<Vec<Frame>, DomainError>

Turn already-extracted prioritized patches (from Self::extract_prioritized_patches) into frames and commit their bookkeeping (next_sequence, stats) — the other half of Self::create_patch_frames.

max_frames only bounds the number of frames returned, not the work done to produce them: every patch in patches is still cloned into some frame, so this call’s cost is O(patches.len()), proportional to the total number of patches extracted, not to max_frames.

Re-checks the streaming-state precondition itself: if the stream transitioned out of Streaming between the caller’s earlier Self::extract_prioritized_patches call and this one, this fails cleanly instead of committing frames for a stream that can no longer accept them.

Not strictly all-or-nothing: chunks are finalized one at a time via Self::finalize_patch_frame, so a construction failure partway through would leave earlier chunks already committed. Currently unobservable — Frame::patch only errors on an empty patch vector, and chunking never produces an empty chunk here.

pub fn finalize_patch_frame( &mut self, priority: Priority, frame_patches: Vec<FramePatch>, ) -> Result<Frame, DomainError>

Build and commit one patch frame from an already-chunked group of patches (see Self::chunk_patches_for_commit).

This is the only place patch-frame next_sequence/stats bookkeeping happens: assigns the next sequence number, constructs the Frame, and records it. Callers that decide which chunked candidates survive before committing (e.g. a cross-stream priority truncation across multiple streams) should call this only for the surviving subset, so discarded candidates never consume a sequence number or inflate stats.

Examples

use pjson_rs_domain::entities::Stream;
use pjson_rs_domain::entities::frame::FramePatch;
use pjson_rs_domain::value_objects::{JsonData, JsonPath, Priority, SessionId};

let mut stream = Stream::new(
    SessionId::new(),
    JsonData::Object(Default::default()),
    Default::default(),
);
stream.start_streaming().unwrap();

let patch = FramePatch::set(JsonPath::root(), JsonData::Bool(true));
let frame = stream
    .finalize_patch_frame(Priority::HIGH, vec![patch])
    .unwrap();

assert_eq!(frame.sequence(), 1);
assert_eq!(stream.stats().total_frames, 1);

pub fn create_completion_frame( &mut self, checksum: Option<String>, ) -> Result<Frame, DomainError>

Create completion frame

pub fn is_active(&self) -> bool

Check if stream is active

pub fn is_finished(&self) -> bool

Check if stream is finished

pub fn duration(&self) -> Option<TimeDelta>

Get stream duration

pub fn progress(&self) -> f64

Calculate stream progress (0.0 to 1.0)

pub fn update_config(&mut self, config: StreamConfig) -> Result<(), DomainError>

Update configuration

pub fn chunk_patches_for_commit( patches: Vec<(FramePatch, Priority)>, max_frames: usize, ) -> Vec<(Priority, Vec<FramePatch>)>

Group prioritized patches into per-frame chunks without constructing any Frame or mutating stream state.

Each returned chunk’s priority is the maximum priority of the patches it contains, so per-frame ordering downstream reflects the most important content the frame will carry. Pure and side-effect-free so a caller (e.g. a cross-stream priority batch) can chunk candidates from several streams, decide which survive truncation, and only then commit the survivors via Self::finalize_patch_frame — see that method’s docs.

Examples

use pjson_rs_domain::entities::Stream;
use pjson_rs_domain::entities::frame::FramePatch;
use pjson_rs_domain::value_objects::{JsonData, JsonPath, Priority, SessionId};

let mut stream = Stream::new(
    SessionId::new(),
    JsonData::Object(Default::default()),
    Default::default(),
);
stream.start_streaming().unwrap();

let patches = vec![
    (
        FramePatch::set(JsonPath::root(), JsonData::Bool(true)),
        Priority::LOW,
    ),
    (
        FramePatch::set(JsonPath::root(), JsonData::Bool(false)),
        Priority::HIGH,
    ),
];

// Chunk without mutating the stream — a caller can inspect priorities
// and pick survivors before any sequence number is spent.
let mut chunks = Stream::chunk_patches_for_commit(patches, 2);
assert_eq!(chunks.len(), 2);
chunks.sort_by_key(|(priority, _)| std::cmp::Reverse(*priority));

// Only commit the highest-priority chunk.
let (priority, frame_patches) = chunks.remove(0);
let frame = stream
    .finalize_patch_frame(priority, frame_patches)
    .unwrap();
assert_eq!(frame.priority(), Priority::HIGH);
assert_eq!(stream.stats().total_frames, 1);

Trait Implementations

impl Clone for Stream

fn clone(&self) -> Stream

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Stream

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for Stream

fn deserialize<__D>( __deserializer: __D, ) -> Result<Stream, <__D as Deserializer<'de>>::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl Serialize for Stream

fn serialize<__S>( &self, __serializer: __S, ) -> Result<<__S as Serializer>::Ok, <__S as Serializer>::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.