share operator by btilds · Pull Request #757 · AlexInLog/ReactivePlusPlus

btilds · 2026-08-16T23:56:54Z

Adds share() as a shortcut for the multicast<Subject>() | ref_count() pipeline. It is templated over the subject with publish_subject as the default, mirroring multicast(), so share<some_subject>() works too.

Because it composes the existing template_multicast_t and ref_count(), the resulting observable type is identical to what publish() | ref_count() produces today, making it a drop-in replacement at existing call sites.

Covered by a new share test case in test_connectable_observable.cpp and a doxygen sample.

Summary by CodeRabbit

New Features
- Added a share operator for multicasting observable values across multiple subscribers.
- Automatically connects on the first subscription and disconnects after the last subscriber unsubscribes.
- Supports configurable subject types, with publish behavior by default.
- Added usage documentation and examples.
Tests
- Added coverage for shared subscriptions, lazy connections, subscriber lifetimes, and reconnection behavior.

Adds `share()` as a shortcut for the `multicast<Subject>() | ref_count()` pipeline. It is templated over the subject with `publish_subject` as the default, mirroring `multicast()`, so `share<some_subject>()` works too. Because it composes the existing `template_multicast_t` and `ref_count()`, the resulting observable type is identical to what `publish() | ref_count()` produces today, making it a drop-in replacement at existing call sites. Covered by a new `share` test case in test_connectable_observable.cpp and a doxygen sample.

coderabbitai · 2026-08-16T23:57:14Z

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 9062dfa6-d2cb-46ad-8fc6-d9d24cb8d361

📥 Commits

Reviewing files that changed from the base of the PR and between d54535d and d9eda58.

📒 Files selected for processing (5)

src/examples/rpp/doxygen/share.cpp
src/rpp/rpp/operators.hpp
src/rpp/rpp/operators/fwd.hpp
src/rpp/rpp/operators/share.hpp
src/tests/rpp/test_connectable_observable.cpp

Included review availability: Your plan includes up to 2 reviews per rolling hour; 1 remains after this review.

📝 Walkthrough

Walkthrough

The pull request adds a configurable share operator. It combines multicast behavior with reference counting, includes tests for subscription lifecycle behavior, and adds a Doxygen example.

Changes

Share operator

Layer / File(s)	Summary
Share API and implementation `src/rpp/rpp/operators/fwd.hpp`, `src/rpp/rpp/operators/share.hpp`, `src/rpp/rpp/operators.hpp`	The public `share` API uses `publish_subject` by default and implements sharing through `multicast` and `ref_count`.
Share subscription behavior tests `src/tests/rpp/test_connectable_observable.cpp`	Tests cover lazy subscription, one shared source subscription, observer-specific unsubscription, completion, and reconnection.
Executable documentation example `src/examples/rpp/doxygen/share.cpp`	The example demonstrates two observers receiving values from a shared observable.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Merge Risk: ⚪ Minimal · up to d9eda

This PR adds a localized share() convenience operator with accompanying tests and documentation; no actionable merge-blocking risk remains beyond normal checks and review.

Sequence Diagram(s)

sequenceDiagram
  participant ObserverA
  participant ObserverB
  participant ShareOperator
  participant Source
  ObserverA->>ShareOperator: subscribe
  ShareOperator->>Source: subscribe on first observer
  ObserverB->>ShareOperator: subscribe
  Source-->>ShareOperator: emit value
  ShareOperator-->>ObserverA: deliver value
  ShareOperator-->>ObserverB: deliver value
  ObserverA-->>ShareOperator: unsubscribe
  ObserverB-->>ShareOperator: remain subscribed
  ObserverB-->>ShareOperator: unsubscribe
  ShareOperator->>Source: dispose after last observer

Suggested reviewers: alexinlog

Poem

A rabbit shares one stream of cheer,
Two listeners hop and gather near.
One source subscription starts the show,
Each receives the values that flow.
When all hop off, the stream lets go.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely identifies the main change: adding the share operator.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
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.

✨ Finishing Touches

🧪 Generate unit tests (beta)

Create PR with unit tests

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

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

AlexInLog

LGTM! thanks

btilds · 2026-08-19T02:33:13Z

LGTM! thanks

Seems like some ci tests failed; mind looking into it?

AlexInLog · 2026-08-19T03:48:24Z

LGTM! thanks

Seems like some ci tests failed; mind looking into it?

Yeah, I'm trying. Looks like qt is outdated

github-actions · 2026-08-20T06:50:00Z

BENCHMARK RESULTS (AUTOGENERATED)

ci-ubuntu-gcc

General

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
Subscribe empty callbacks to empty observable	298.76 ns	1.87 ns	1.87 ns	1.00	1.56 ns
Subscribe empty callbacks to empty observable via pipe operator	303.61 ns	1.87 ns	1.87 ns	1.00	1.56 ns

Sources

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
from array of 1 - create + subscribe + immediate	713.90 ns	0.31 ns	0.31 ns	1.00	0.31 ns
from array of 1 - create + subscribe + current_thread	1052.64 ns	3.73 ns	3.74 ns	1.00	3.74 ns
concat_as_source of just(1 immediate) create + subscribe	2328.15 ns	114.73 ns	115.63 ns	0.99	122.64 ns
defer from array of 1 - defer + create + subscribe + immediate	739.96 ns	0.31 ns	0.31 ns	1.00	0.31 ns
interval - interval + take(3) + subscribe + immediate	2177.07 ns	59.72 ns	59.73 ns	1.00	59.70 ns
interval - interval + take(3) + subscribe + current_thread	3033.70 ns	32.98 ns	32.67 ns	1.01	34.22 ns
from array of 1 - create + as_blocking + subscribe + new_thread	38134.94 ns	34722.10 ns	34522.91 ns	1.01	33302.77 ns
from array of 1000 - create + as_blocking + subscribe + new_thread	45241.52 ns	57268.32 ns	57499.67 ns	1.00	54846.11 ns
concat_as_source of just(1 immediate) and just(1,2 immediate)create + subscribe	3504.90 ns	136.26 ns	154.84 ns	0.88	155.11 ns

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take(1)+subscribe	1107.50 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just+filter(true)+subscribe	855.82 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,2)+skip(1)+subscribe	1011.38 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,1,2)+distinct_until_changed()+subscribe	861.51 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,2)+first()+subscribe	1256.73 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,2)+last()+subscribe	935.09 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just+take_last(1)+subscribe	1119.54 ns	18.36 ns	18.99 ns	0.97	20.23 ns
immediate_just(1,2,3)+element_at(1)+subscribe	841.00 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Schedulers

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate scheduler create worker + schedule	275.53 ns	1.56 ns	0.47 ns	3.33	0.62 ns
current_thread scheduler create worker + schedule	377.17 ns	4.36 ns	4.36 ns	1.00	4.67 ns
current_thread scheduler create worker + schedule + recursive schedule	822.53 ns	62.21 ns	61.49 ns	1.01	61.12 ns

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+map(v*2)+subscribe	859.35 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just+scan(10, std::plus)+subscribe	907.88 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just+flat_map(immediate_just(v*2))+subscribe	2343.12 ns	121.30 ns	137.71 ns	0.88	173.77 ns
immediate_just+buffer(2)+subscribe	1542.99 ns	13.70 ns	14.01 ns	0.98	18.07 ns
immediate_just+window(2)+subscribe + subscsribe inner	2562.50 ns	1268.96 ns	1296.71 ns	0.98	1345.27 ns

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take_while(false)+subscribe	839.33 ns	-	-	0.00	-
immediate_just+take_while(true)+subscribe	831.08 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Utility Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(1)+subscribe_on(immediate)+subscribe	1967.04 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Combining Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	3483.48 ns	156.28 ns	165.36 ns	0.95	172.77 ns
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	3614.27 ns	163.99 ns	172.20 ns	0.95	167.24 ns
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	141.61 ns	135.07 ns	1.05	154.59 ns
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	3602.29 ns	372.38 ns	383.72 ns	0.97	393.71 ns
immediate_just(1) + zip(immediate_just(2)) + subscribe	2141.92 ns	215.16 ns	217.36 ns	0.99	209.03 ns
immediate_just(immediate_just(1), immediate_just(1)) + concat() + subscribe	3178.38 ns	234.34 ns	254.77 ns	0.92	260.30 ns

Subjects

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
publish_subject with 1 observer - on_next	34.85 ns	14.82 ns	14.82 ns	1.00	15.22 ns
subscribe 100 observers to publish_subject	205615.00 ns	17559.53 ns	18241.89 ns	0.96	17735.43 ns
100 on_next to 100 observers to publish_subject	28526.55 ns	16834.75 ns	16932.07 ns	0.99	16954.26 ns

Scenarios

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
basic sample	1439.38 ns	14.51 ns	13.07 ns	1.11	22.41 ns
basic sample with immediate scheduler	1438.00 ns	5.29 ns	5.29 ns	1.00	17.43 ns
mix operators with disposables and without disposables	6514.95 ns	1417.61 ns	1465.78 ns	0.97	1885.93 ns
single disposable and looooooong indentity chain	25084.59 ns	1033.41 ns	1045.04 ns	0.99	4826.62 ns

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+reduce(10, std::plus)+subscribe	907.81 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	2084.16 ns	1009.67 ns	999.51 ns	1.01	1058.71 ns
create(on_error())+retry(1)+subscribe	598.67 ns	117.92 ns	112.19 ns	1.05	127.03 ns

ci-macos

General

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
Subscribe empty callbacks to empty observable	562.50 ns	0.61 ns	0.57 ns	1.08	0.48 ns
Subscribe empty callbacks to empty observable via pipe operator	399.53 ns	0.59 ns	0.69 ns	0.85	0.49 ns

Sources

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
from array of 1 - create + subscribe + immediate	1010.47 ns	0.39 ns	0.41 ns	0.94	0.31 ns
from array of 1 - create + subscribe + current_thread	1094.48 ns	4.67 ns	4.60 ns	1.02	3.75 ns
concat_as_source of just(1 immediate) create + subscribe	3244.01 ns	224.55 ns	203.42 ns	1.10	168.89 ns
defer from array of 1 - defer + create + subscribe + immediate	1025.21 ns	0.51 ns	0.39 ns	1.32	0.31 ns
interval - interval + take(3) + subscribe + immediate	2547.58 ns	71.53 ns	63.96 ns	1.12	50.62 ns
interval - interval + take(3) + subscribe + current_thread	3000.00 ns	36.27 ns	67.18 ns	0.54	30.19 ns
from array of 1 - create + as_blocking + subscribe + new_thread	27748.89 ns	46076.92 ns	39013.91 ns	1.18	25483.00 ns
from array of 1000 - create + as_blocking + subscribe + new_thread	49632.26 ns	31792.94 ns	36184.52 ns	0.88	29445.12 ns
concat_as_source of just(1 immediate) and just(1,2 immediate)create + subscribe	8378.86 ns	261.99 ns	261.20 ns	1.00	183.45 ns

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take(1)+subscribe	1320.48 ns	0.41 ns	0.40 ns	1.01	0.35 ns
immediate_just+filter(true)+subscribe	873.39 ns	0.40 ns	0.41 ns	0.98	0.34 ns
immediate_just(1,2)+skip(1)+subscribe	2559.63 ns	0.38 ns	0.51 ns	0.74	0.34 ns
immediate_just(1,1,2)+distinct_until_changed()+subscribe	1098.36 ns	0.38 ns	0.49 ns	0.77	0.34 ns
immediate_just(1,2)+first()+subscribe	1708.86 ns	0.39 ns	0.42 ns	0.93	0.35 ns
immediate_just(1,2)+last()+subscribe	1225.61 ns	0.86 ns	1.09 ns	0.79	1.00 ns
immediate_just+take_last(1)+subscribe	1410.37 ns	0.39 ns	0.39 ns	0.99	0.34 ns
immediate_just(1,2,3)+element_at(1)+subscribe	1256.96 ns	0.49 ns	0.38 ns	1.29	0.34 ns

Schedulers

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate scheduler create worker + schedule	353.65 ns	0.58 ns	0.58 ns	1.00	0.49 ns
current_thread scheduler create worker + schedule	498.61 ns	4.68 ns	6.88 ns	0.68	3.82 ns
current_thread scheduler create worker + schedule + recursive schedule	868.60 ns	105.98 ns	78.82 ns	1.34	60.57 ns

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+map(v*2)+subscribe	1111.16 ns	3.41 ns	2.91 ns	1.17	2.42 ns
immediate_just+scan(10, std::plus)+subscribe	1142.04 ns	0.39 ns	0.40 ns	0.98	0.31 ns
immediate_just+flat_map(immediate_just(v*2))+subscribe	2429.79 ns	239.42 ns	380.43 ns	0.63	222.60 ns
immediate_just+buffer(2)+subscribe	1188.82 ns	18.90 ns	24.79 ns	0.76	20.51 ns
immediate_just+window(2)+subscribe + subscsribe inner	2371.21 ns	1250.00 ns	1301.79 ns	0.96	1057.75 ns

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take_while(false)+subscribe	912.75 ns	-	-	0.00	-
immediate_just+take_while(true)+subscribe	1111.36 ns	0.50 ns	0.38 ns	1.32	0.31 ns

Utility Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(1)+subscribe_on(immediate)+subscribe	4054.37 ns	2.64 ns	2.45 ns	1.08	1.91 ns

Combining Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	3565.51 ns	259.92 ns	251.20 ns	1.03	203.92 ns
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	4027.73 ns	264.80 ns	241.41 ns	1.10	201.55 ns
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	265.95 ns	401.30 ns	0.66	209.50 ns
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	3798.78 ns	623.78 ns	627.22 ns	0.99	477.78 ns
immediate_just(1) + zip(immediate_just(2)) + subscribe	2284.34 ns	309.43 ns	364.34 ns	0.85	267.94 ns
immediate_just(immediate_just(1), immediate_just(1)) + concat() + subscribe	3448.83 ns	392.60 ns	458.97 ns	0.86	317.50 ns

Subjects

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
publish_subject with 1 observer - on_next	51.85 ns	26.12 ns	25.06 ns	1.04	22.97 ns
subscribe 100 observers to publish_subject	169979.17 ns	20695.57 ns	36227.00 ns	0.57	17739.74 ns
100 on_next to 100 observers to publish_subject	42815.46 ns	15552.29 ns	14242.61 ns	1.09	11971.25 ns

Scenarios

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
basic sample	1335.71 ns	23.53 ns	11.62 ns	2.03	28.86 ns
basic sample with immediate scheduler	2638.16 ns	6.54 ns	6.21 ns	1.05	9.22 ns
mix operators with disposables and without disposables	8705.24 ns	2761.04 ns	2301.92 ns	1.20	1954.42 ns
single disposable and looooooong indentity chain	22225.45 ns	1988.54 ns	1958.42 ns	1.02	3901.94 ns

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+reduce(10, std::plus)+subscribe	1161.96 ns	0.38 ns	0.42 ns	0.89	0.35 ns

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	5160.81 ns	3678.79 ns	3716.55 ns	0.99	3341.64 ns
create(on_error())+retry(1)+subscribe	854.32 ns	230.90 ns	207.98 ns	1.11	189.10 ns

ci-ubuntu-clang

General

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
Subscribe empty callbacks to empty observable	268.27 ns	1.56 ns	0.64 ns	2.43	0.64 ns
Subscribe empty callbacks to empty observable via pipe operator	270.57 ns	1.55 ns	0.64 ns	2.43	0.64 ns

Sources

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
from array of 1 - create + subscribe + immediate	570.36 ns	0.31 ns	0.31 ns	1.00	0.31 ns
from array of 1 - create + subscribe + current_thread	788.99 ns	4.04 ns	4.04 ns	1.00	4.04 ns
concat_as_source of just(1 immediate) create + subscribe	2400.08 ns	130.85 ns	131.02 ns	1.00	131.55 ns
defer from array of 1 - defer + create + subscribe + immediate	814.29 ns	0.31 ns	0.31 ns	1.00	0.31 ns
interval - interval + take(3) + subscribe + immediate	2247.31 ns	58.74 ns	58.77 ns	1.00	58.76 ns
interval - interval + take(3) + subscribe + current_thread	3191.10 ns	31.10 ns	31.15 ns	1.00	31.74 ns
from array of 1 - create + as_blocking + subscribe + new_thread	38017.41 ns	34911.70 ns	34636.79 ns	1.01	33228.21 ns
from array of 1000 - create + as_blocking + subscribe + new_thread	45075.92 ns	39528.72 ns	41944.75 ns	0.94	40850.92 ns
concat_as_source of just(1 immediate) and just(1,2 immediate)create + subscribe	3748.22 ns	151.30 ns	148.54 ns	1.02	151.00 ns

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take(1)+subscribe	1209.27 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just+filter(true)+subscribe	876.53 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,2)+skip(1)+subscribe	1122.40 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,1,2)+distinct_until_changed()+subscribe	907.24 ns	0.33 ns	0.33 ns	0.99	0.32 ns
immediate_just(1,2)+first()+subscribe	1433.62 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,2)+last()+subscribe	1052.64 ns	0.31 ns	0.31 ns	1.01	0.31 ns
immediate_just+take_last(1)+subscribe	1265.75 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just(1,2,3)+element_at(1)+subscribe	900.59 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Schedulers

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate scheduler create worker + schedule	281.32 ns	0.64 ns	1.56 ns	0.41	1.55 ns
current_thread scheduler create worker + schedule	420.60 ns	4.36 ns	4.05 ns	1.08	4.04 ns
current_thread scheduler create worker + schedule + recursive schedule	883.62 ns	57.74 ns	58.06 ns	0.99	57.98 ns

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+map(v*2)+subscribe	881.61 ns	0.31 ns	0.31 ns	1.00	0.31 ns
immediate_just+scan(10, std::plus)+subscribe	1007.76 ns	0.62 ns	0.62 ns	1.00	0.32 ns
immediate_just+flat_map(immediate_just(v*2))+subscribe	2296.37 ns	137.36 ns	138.89 ns	0.99	135.02 ns
immediate_just+buffer(2)+subscribe	1589.66 ns	14.30 ns	14.62 ns	0.98	14.61 ns
immediate_just+window(2)+subscribe + subscsribe inner	2550.24 ns	905.18 ns	906.68 ns	1.00	935.55 ns

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take_while(false)+subscribe	886.20 ns	-	-	0.00	-
immediate_just+take_while(true)+subscribe	881.62 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Utility Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(1)+subscribe_on(immediate)+subscribe	2095.81 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Combining Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	3437.69 ns	179.21 ns	160.23 ns	1.12	155.71 ns
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	3815.19 ns	138.79 ns	140.61 ns	0.99	139.44 ns
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	142.01 ns	143.13 ns	0.99	139.33 ns
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	3430.12 ns	383.35 ns	378.64 ns	1.01	379.64 ns
immediate_just(1) + zip(immediate_just(2)) + subscribe	2290.61 ns	200.39 ns	204.79 ns	0.98	197.31 ns
immediate_just(immediate_just(1), immediate_just(1)) + concat() + subscribe	3305.83 ns	224.81 ns	228.10 ns	0.99	224.63 ns

Subjects

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
publish_subject with 1 observer - on_next	54.35 ns	19.89 ns	19.80 ns	1.00	20.31 ns
subscribe 100 observers to publish_subject	212535.60 ns	17454.33 ns	17511.56 ns	1.00	17597.67 ns
100 on_next to 100 observers to publish_subject	37702.32 ns	20260.15 ns	20258.60 ns	1.00	20281.73 ns

Scenarios

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
basic sample	1350.97 ns	11.51 ns	11.52 ns	1.00	21.47 ns
basic sample with immediate scheduler	1346.07 ns	5.91 ns	5.91 ns	1.00	6.53 ns
mix operators with disposables and without disposables	6673.64 ns	1230.76 ns	1211.18 ns	1.02	1492.24 ns
single disposable and looooooong indentity chain	27923.98 ns	1254.46 ns	1487.40 ns	0.84	4707.14 ns

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+reduce(10, std::plus)+subscribe	1029.30 ns	0.31 ns	0.31 ns	1.00	0.31 ns

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	2248.87 ns	1181.45 ns	1178.99 ns	1.00	1195.48 ns
create(on_error())+retry(1)+subscribe	684.37 ns	138.72 ns	139.29 ns	1.00	140.91 ns

ci-windows

General

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
Subscribe empty callbacks to empty observable	556.07 ns	2.16 ns	2.16 ns	1.00	1.85 ns
Subscribe empty callbacks to empty observable via pipe operator	572.10 ns	2.16 ns	2.16 ns	1.00	1.85 ns

Sources

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
from array of 1 - create + subscribe + immediate	1155.57 ns	4.93 ns	4.94 ns	1.00	5.55 ns
from array of 1 - create + subscribe + current_thread	1419.82 ns	15.75 ns	15.74 ns	1.00	15.45 ns
concat_as_source of just(1 immediate) create + subscribe	3696.46 ns	174.60 ns	174.62 ns	1.00	178.05 ns
defer from array of 1 - defer + create + subscribe + immediate	1180.65 ns	5.24 ns	5.24 ns	1.00	5.24 ns
interval - interval + take(3) + subscribe + immediate	3736.36 ns	139.81 ns	139.74 ns	1.00	142.29 ns
interval - interval + take(3) + subscribe + current_thread	3460.30 ns	60.15 ns	59.86 ns	1.00	62.80 ns
from array of 1 - create + as_blocking + subscribe + new_thread	123175.00 ns	118700.00 ns	120400.00 ns	0.99	119788.89 ns
from array of 1000 - create + as_blocking + subscribe + new_thread	132655.56 ns	135675.00 ns	136975.00 ns	0.99	137275.00 ns
concat_as_source of just(1 immediate) and just(1,2 immediate)create + subscribe	5329.67 ns	206.57 ns	201.39 ns	1.03	215.13 ns

Filtering Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take(1)+subscribe	1818.86 ns	19.42 ns	19.42 ns	1.00	21.36 ns
immediate_just+filter(true)+subscribe	1610.58 ns	18.50 ns	18.51 ns	1.00	21.59 ns
immediate_just(1,2)+skip(1)+subscribe	2008.80 ns	17.89 ns	17.91 ns	1.00	21.60 ns
immediate_just(1,1,2)+distinct_until_changed()+subscribe	1328.95 ns	20.67 ns	20.68 ns	1.00	26.86 ns
immediate_just(1,2)+first()+subscribe	2372.16 ns	18.20 ns	18.21 ns	1.00	19.43 ns
immediate_just(1,2)+last()+subscribe	1464.98 ns	19.13 ns	19.15 ns	1.00	22.84 ns
immediate_just+take_last(1)+subscribe	2023.02 ns	64.85 ns	64.93 ns	1.00	70.17 ns
immediate_just(1,2,3)+element_at(1)+subscribe	1622.12 ns	20.97 ns	20.98 ns	1.00	21.63 ns

Schedulers

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate scheduler create worker + schedule	477.39 ns	4.32 ns	4.32 ns	1.00	4.32 ns
current_thread scheduler create worker + schedule	648.68 ns	11.11 ns	11.11 ns	1.00	11.11 ns
current_thread scheduler create worker + schedule + recursive schedule	1079.50 ns	103.96 ns	101.36 ns	1.03	103.12 ns

Transforming Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+map(v*2)+subscribe	1328.23 ns	18.80 ns	18.80 ns	1.00	21.63 ns
immediate_just+scan(10, std::plus)+subscribe	1422.12 ns	20.96 ns	20.96 ns	1.00	23.80 ns
immediate_just+flat_map(immediate_just(v*2))+subscribe	3837.33 ns	186.30 ns	185.11 ns	1.01	223.46 ns
immediate_just+buffer(2)+subscribe	2291.50 ns	64.26 ns	63.51 ns	1.01	72.33 ns
immediate_just+window(2)+subscribe + subscsribe inner	3974.07 ns	1216.92 ns	1192.15 ns	1.02	1226.57 ns

Conditional Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+take_while(false)+subscribe	1305.28 ns	17.57 ns	17.57 ns	1.00	19.13 ns
immediate_just+take_while(true)+subscribe	1329.92 ns	18.50 ns	18.50 ns	1.00	21.60 ns

Utility Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(1)+subscribe_on(immediate)+subscribe	3205.02 ns	11.10 ns	11.11 ns	1.00	11.11 ns

Combining Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	5054.46 ns	197.43 ns	195.27 ns	1.01	222.13 ns
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	5710.11 ns	186.78 ns	178.89 ns	1.04	204.81 ns
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	195.15 ns	194.65 ns	1.00	199.56 ns
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	6119.16 ns	444.60 ns	443.65 ns	1.00	489.05 ns
immediate_just(1) + zip(immediate_just(2)) + subscribe	3865.37 ns	538.25 ns	519.56 ns	1.04	516.04 ns
immediate_just(immediate_just(1), immediate_just(1)) + concat() + subscribe	4853.97 ns	314.42 ns	314.38 ns	1.00	325.14 ns

Subjects

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
publish_subject with 1 observer - on_next	36.72 ns	29.42 ns	29.27 ns	1.01	30.01 ns
subscribe 100 observers to publish_subject	264250.00 ns	25013.33 ns	25450.00 ns	0.98	24830.23 ns
100 on_next to 100 observers to publish_subject	51881.82 ns	35939.39 ns	35922.58 ns	1.00	32997.22 ns

Scenarios

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
basic sample	1853.25 ns	96.57 ns	96.80 ns	1.00	112.11 ns
basic sample with immediate scheduler	1857.89 ns	66.73 ns	68.51 ns	0.97	82.35 ns
mix operators with disposables and without disposables	9381.75 ns	1787.22 ns	1797.99 ns	0.99	2439.19 ns
single disposable and looooooong indentity chain	25641.46 ns	1655.18 ns	1630.37 ns	1.02	6267.21 ns

Aggregating Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
immediate_just+reduce(10, std::plus)+subscribe	1458.16 ns	19.11 ns	19.11 ns	1.00	22.83 ns

Error Handling Operators

name	rxcpp	rpp	prev rpp	ratio	rpp no optimization
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe	1896.36 ns	357.24 ns	353.06 ns	1.01	380.86 ns
create(on_error())+retry(1)+subscribe	1721.11 ns	138.15 ns	137.92 ns	1.00	139.85 ns

AlexInLog approved these changes Aug 17, 2026

View reviewed changes

Merge branch 'v2' into share-operator

e7d117e

AlexInLog merged commit 3163483 into AlexInLog:v2 Aug 20, 2026
39 checks passed

Conversation

btilds commented Aug 16, 2026 • edited by coderabbitai Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary by CodeRabbit

Uh oh!

coderabbitai Bot commented Aug 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Walkthrough

Changes

Sequence Diagram(s)

Uh oh!

AlexInLog left a comment

Choose a reason for hiding this comment

Uh oh!

btilds commented Aug 19, 2026

Uh oh!

AlexInLog commented Aug 19, 2026

Uh oh!

github-actions Bot commented Aug 20, 2026

BENCHMARK RESULTS (AUTOGENERATED)

ci-ubuntu-gcc

General

Sources

Filtering Operators

Schedulers

Transforming Operators

Conditional Operators

Utility Operators

Combining Operators

Subjects

Scenarios

Aggregating Operators

Error Handling Operators

ci-macos

General

Sources

Filtering Operators

Schedulers

Transforming Operators

Conditional Operators

Utility Operators

Combining Operators

Subjects

Scenarios

Aggregating Operators

Error Handling Operators

ci-ubuntu-clang

General

Sources

Filtering Operators

Schedulers

Transforming Operators

Conditional Operators

Utility Operators

Combining Operators

Subjects

Scenarios

Aggregating Operators

Error Handling Operators

ci-windows

General

Sources

Filtering Operators

Schedulers

Transforming Operators

Conditional Operators

Utility Operators

Combining Operators

Subjects

Scenarios

Aggregating Operators

Error Handling Operators

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

btilds commented Aug 16, 2026 •

edited by coderabbitai Bot

Loading

coderabbitai Bot commented Aug 16, 2026 •

edited

Loading