Implement super_stream

[DanielePalaia]

This closes #204

This PR implements the superstream functionality both for producer and consumer (For the moment it exclude the single active consumer implementation that will be implemented in a different PR).

It import the murmur3 dependency in order to use the same hashing algorithm also the other clients are using.

Examples are provided in examples folder: send_super_stream.rs and receive_super_stream.rs

[DanielePalaia]

@wolf4ood I was wondering if we can limit the usage of clone() here for message and cb. I made some tests but apparently this is not so trivial

[codecov]

Codecov Report

Attention: Patch coverage is 94.35484% with 14 lines in your changes missing coverage. Please review.

Project coverage is 89.75%. Comparing base (f2aca24) to head (2238a20).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
src/error.rs	0.00%	6 Missing ⚠️
src/consumer.rs	57.14%	3 Missing ⚠️
src/superstream_producer.rs	96.59%	3 Missing ⚠️
src/superstream_consumer.rs	97.43%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #232      +/-   ##
==========================================
+ Coverage   88.73%   89.75%   +1.02%     
==========================================
  Files          75       78       +3     
  Lines        6410     6658     +248     
==========================================
+ Hits         5688     5976     +288     
+ Misses        722      682      -40     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[DanielePalaia]

Hey @Gsantomaggio @wolf4ood I know this is quite huge, but if you get time to start reviewing it when you have time it would be great. I created examples in the doc folders.

There are still a couple of small issues (written above) and probably the filtering part integration needs to be tested more carefully but I think we are in a good way now!

[Gsantomaggio]

@DanielePalaia, the super-stream producer and super-stream consumer, seems to work correctly :)!

There is some problem with the connections.
Given a cluster ( I created it using https://github.com/rabbitmq/rabbitmq-stream-go-client and make rabbitmq-ha-proxy )

Changed the examples in this way:

    let environment = Environment::builder().port(5553).load_balancer_mode(true).build().await?;

Then:

 use std::io::{stdin,stdout,Write};
    let mut s=String::new();
    print!("Please enter close the super stream consumer");
    let _=stdout().flush();
    stdin().read_line(&mut s).expect("Did not enter a correct string");
    if let Some('\n')=s.chars().next_back() {
        s.pop();
    }


    let _ = super_stream_consumer.handle().close().await;

    print!("Please enter close the the example");
    let _=stdout().flush();
    stdin().read_line(&mut s).expect("Did not enter a correct string");
    if let Some('\n')=s.chars().next_back() {
        s.pop();
    }


    println!("Super stream consumer stopped");

Same for the producer:

  use std::io::{stdin,stdout,Write};
    let mut s=String::new();
    print!("Please enter close the super stream producer");
    let _=stdout().flush();
    stdin().read_line(&mut s).expect("Did not enter a correct string");
    if let Some('\n')=s.chars().next_back() {
        s.pop();
    }

    let _ = super_stream_producer.close().await;

    print!("Please enter close the example");
    let _=stdout().flush();
    stdin().read_line(&mut s).expect("Did not enter a correct string");
    if let Some('\n')=s.chars().next_back() {
        s.pop();
    }
    Ok(())

When the close() is called there are pending connections:

But all these connections are not linked to any stream.

All the connections should be closed.

[Gsantomaggio]

Ok, with this PR #235, the pending connections problem is solved.

There is still some problem with the load-banacer part #236
but we can solve in another PR.

Great job @DanielePalaia and Thank you @wolf4ood for the review!