refactor(core): move AckStream, DropStream, ResponseHandlers and ChanStream into core behind remote-adapter feature#758
Conversation
…Stream into core behind remote-adapter feature Moves the duplicated stream constructs from all three remote adapters (redis, postgres, mongodb) into socketioxide-core behind the existing remote-adapter feature flag. Removes 737 lines of duplicated code. - DropStream<S, T>: generic cleanup-on-drop stream wrapper - AckStream<S, R, T, E>: generic ack stream merging local + remote - ResponseHandlers<T>: generic type alias for HashMap<Sid, mpsc::Sender<T>> - ChanStream<T>: generic mpsc::Receiver stream wrapper Each adapter now provides a decode function passed as a fn pointer to AckStream, avoiding orphan rule issues with trait impls on foreign types like Vec<u8>. Closes Totodore#756
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
Totodore
added
A-core
|
@Totodore this is ready for review. The 3 clippy code-scanning comments are now fixed (module docs + AckDecoder type alias). The cargo test --all-features --workspace CI failure appears pre-existing on main, looks related to fred/redis-cluster optional feature doctests. |
|
CI is green on main, https://github.com/Totodore/socketioxide/actions/runs/29045217076. So there is no reason for it to be red here. Formatting is not even ok. |
|
engine_io (V3) seems to be a flaky e2e test, this changes dont touch anything in engineioxide. could you re-run the job? |
Totodore
left a comment
Totodore
left a comment
There was a problem hiding this comment.
Neat!
Just some comments, not exhaustive though as I'm on my phone.
I will do a more thorough review later.
| serde.workspace = true | ||
| thiserror.workspace = true | ||
| futures-core.workspace = true | ||
| futures-util = { workspace = true, optional = true } |
There was a problem hiding this comment.
Please, add default-features = false + only pick the features needed. Same goes for the other added deps if they have features.
Co-authored-by: Théodore Prévot <prevottheodore@gmail.com>
… 11 AckStream tests
|
Okay, looks like all green man, except socket_io (v4), which i expect to be a flaky e2e test again, try to re run as soon as you can, if not i will do a further analysis to see if maybe it is an actual regression. |
Totodore
left a comment
Totodore
left a comment
There was a problem hiding this comment.
Please also check dependencies feature flags!
| pub struct AckStream<S, R: Stream, T, E> { | ||
| #[pin] | ||
| local: S, | ||
| #[pin] | ||
| remote: DropStream<TakeUntil<R, time::Sleep>, T>, | ||
| decode: AckDecoder<E, R::Item>, | ||
| ack_cnt: u32, | ||
| total_ack_cnt: usize, | ||
| serv_cnt: u16, | ||
| } | ||
| } |
There was a problem hiding this comment.
Because E: SocketEmitter lives in the crate I think we could reduce AckStream to depend on E: SocketEmitter and use its GAT (AckStream, AckError).
Also, little nit for AckDecoder, it is better take the error as a second param.
| pub struct AckStream<S, R: Stream, T, E> { | |
| #[pin] | |
| local: S, | |
| #[pin] | |
| remote: DropStream<TakeUntil<R, time::Sleep>, T>, | |
| decode: AckDecoder<E, R::Item>, | |
| ack_cnt: u32, | |
| total_ack_cnt: usize, | |
| serv_cnt: u16, | |
| } | |
| } | |
| pub struct AckStream<E: SocketEmitter, R: Stream, T> { | |
| #[pin] | |
| local: E::AckStream, | |
| #[pin] | |
| remote: DropStream<TakeUntil<R, time::Sleep>, T>, | |
| decode: AckDecoder<R::Item, E::AckError>, | |
| ack_cnt: u32, | |
| total_ack_cnt: usize, | |
| serv_cnt: u16, | |
| } | |
| } |
There was a problem hiding this comment.
I think you can drop the test that already exists in the core lib and move those few lines to lib.rs.
There was a problem hiding this comment.
I think you can drop the test that already exists in the core lib and move those few lines to lib.rs.
There was a problem hiding this comment.
I think you can drop the test that already exists in the core lib and move those few lines to lib.rs.
| } | ||
|
|
||
| /// Create a new `AckStream` for local-only use, with an empty remote stream. | ||
| pub fn new_empty_remote(local: S, empty_remote: R, decode: AckDecoder<E, R::Item>) -> Self { |
There was a problem hiding this comment.
I'm not really fond of this. Two solutions, either AckStream becomes an enum with two variants:
- Remote which is equivalent to the current impl
- Local which is only simply a wrapper arround the local emitter stream.
Another solution is to remove this "empty remote" think entirely and make each adapter define type AckStream = Either<E::AckStream, AckStream<...>>. But that's a bit verbose.
3 participants
Motivation
All three remote adapters (redis, postgres, mongodb) independently define AckStream, DropStream, and ResponseHandlers with near-identical code. This duplication makes maintenance harder and is contrary to the goal of #727 (share more code between adapters).
Solution
Move AckStream<S, R, T, E>, DropStream<S, T>, ResponseHandlers, and ChanStream into socketioxide-core behind the existing remote-adapter feature flag. Each adapter now provides a stateless decode function passed as a fn pointer to AckStream::new(), avoiding orphan rule issues that a trait-based approach would cause when implementing on foreign types like Vec.