feat: stream heartbeat wrapper

[bobrykov]

No description provided.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: d83f618c-6202-4155-8d38-7c87658be970

📥 Commits

Reviewing files that changed from the base of the PR and between 725e791 and cf40be2.

📒 Files selected for processing (2)

• src/stream.rs
• src/stream/heartbeat.rs

✅ Files skipped from review due to trivial changes (1)

• src/stream.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• src/stream/heartbeat.rs

---

📝 Walkthrough

Walkthrough

Adds a new heartbeat submodule exposing HeartbeatStream<S>, a stream wrapper that invokes a user callback on configured heartbeat intervals and returns an ApiError when a hard timeout is exceeded.

Changes

Heartbeat Stream Wrapper

Layer / File(s)	Summary
Heartbeat Type Definitions src/stream/heartbeat.rs	Module docs and imports; introduces HeartbeatData (elapsed, is_timeout), HeartbeatCallback (boxed Fn(HeartbeatData)), and HeartbeatConfig (stores interval, timeout, and callback; public constructor/accessors; Debug).
HeartbeatStream Structure and Constructor src/stream/heartbeat.rs	Defines HeartbeatStream<S> state (last_heartbeat, start, internal tokio::time::Sleep) and new which initializes timers and schedules the timeout deadline using Instant::checked_add with a far-future fallback.
Stream Trait Implementation src/stream/heartbeat.rs	Implements Stream for HeartbeatStream<S>: on poll_next computes elapsed, invokes callback when interval elapsed, polls the scheduled Sleep (or checks start.elapsed()), returns ApiError::Api on timeout, otherwise delegates to inner stream.
Module Documentation and Export src/stream.rs	Adds heartbeat to module docs and pub mod heartbeat; to expose the submodule.
Unit and Tokio Tests src/stream/heartbeat.rs	Adds tests and test helpers validating heartbeat data, config accessors/Debug, inner stream passthrough, heartbeat callback firing (via manual last_heartbeat), and timeout behavior (via manual start backdating).

Sequence Diagram(s)

sequenceDiagram
  participant Caller
  participant HeartbeatStream
  participant HeartbeatCallback
  participant InnerStream
  Caller->>HeartbeatStream: poll_next()
  HeartbeatStream->>HeartbeatStream: compute elapsed, compare heartbeat_interval
  alt Interval elapsed
    HeartbeatStream->>HeartbeatCallback: call(HeartbeatData { elapsed, is_timeout })
  end
  HeartbeatStream->>HeartbeatStream: poll timeout Sleep or check start.elapsed()
  alt Timeout exceeded
    HeartbeatStream->>Caller: return Err(ApiError::Api)
  else Not timed out
    HeartbeatStream->>InnerStream: poll_next()
    InnerStream->>HeartbeatStream: return Ok(StreamEvent) / Err(ApiError)
    HeartbeatStream->>Caller: forward result
  end

Loading

Possibly related PRs

• dch-labs/loopctl#11: Introduces stream types and the original stream module that heartbeat extends.

🚥 Pre-merge checks | ✅ 4

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'feat: stream heartbeat wrapper' accurately summarizes the main change: adding a new heartbeat wrapper functionality to the stream module.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 50.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/stream-heartbeat

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/stream/heartbeat.rs`:
- Around line 396-411: The test fires_heartbeat_on_interval is non-validating
because it creates HeartbeatStream (HeartbeatStream::new) but binds it to
`_stream` and never polls it; update the test to keep the stream (remove the
leading underscore) and actively poll it so heartbeat logic in poll_next runs
and you can assert the callback side-effect: create the callbacks
Arc<Mutex<Vec<HeartbeatData>>> via make_config, construct VecStream and
HeartbeatStream, then use futures::StreamExt::next (or pin the stream and call
poll_next once) after waiting the interval to force heartbeat execution, and
finally assert that callbacks.lock() contains at least one HeartbeatData entry
to verify the heartbeat fired.
- Around line 268-291: The current poll_next (in the stream wrapper that uses
fields last_heartbeat, start, config, on_heartbeat, and inner) only checks
Instant::elapsed() when polled, so add an internal tokio timer (e.g., a
tokio::time::Sleep stored as a field like timeout_sleep:
Option<Pin<Box<Sleep>>>) and poll it alongside the inner stream inside
poll_next; initialize or reset that Sleep to the remaining duration
(config.timeout - start.elapsed()) when the stream starts or after any
activity/heartbeat, poll the Sleep and when it completes return the same
ApiError timeout, and ensure you reset/replace the Sleep whenever
last_heartbeat/start are reset so the wrapper can wake itself even if inner
returns Poll::Pending.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: f2ea6dc4-7524-4219-92c4-af7d71341aa8

📥 Commits

Reviewing files that changed from the base of the PR and between 638749d and a4f5fd0.

📒 Files selected for processing (2)

• src/stream.rs
• src/stream/heartbeat.rs