Internal leakage occurs when the broker is established and the client is used to listen

[LongZoz]

Internal leakage occurs when the broker is established and the client is used to listen

[swanandx]

When broker gets a message published by client, it stores the message in a commitlog, which i believe, is the reason you are seeing the increase in memory usage as the number message published increase!

Growth in memory usage as the number of messages increase is expected till a certain limit. You can control the maximum size of the segment in commitlog by max_segment_size & max_segment_count in rumqttd.toml. ( commitlog is basically collection of those segments )

[LongZoz]

I look at these parameters, set the maximum number, the new package will not be accepted, is rejected; So how do you clear the old data to make room for the new data

[swanandx]

if there is no space, older segments are dropped automatically by the broker to make space for new data! can you provide rumqttd config and the client code you are using to we can debug this further?

[swanandx]

BTW, it would be also helpful if you paste the failure logs, like wdym by package being rejected, which packet it was and what did rumqttd log the reason of rejecting the packet.

[LongZoz]

This is my toml file
`[router]
id = 0
max_connections = 10010
max_outgoing_packet_count = 200
max_segment_size = 104857600
max_segment_count = 10

[v5.1]
name = "v5-1"
listen = "0.0.0.0:18804"
next_connection_delay_ms = 1
[v5.1.connections]
connection_timeout_ms = 60000
max_payload_size = 20480
max_inflight_count = 100
Here is my broker and client code：
thread::spawn(||{
let mut broker = Broker::new(configs);
broker.start().unwrap();
println!("start");
});

thread::sleep(Duration::from_secs(1));

let mut mqttoptions = MqttOptions::new("test", "127.0.0.1", 18804);
   
    mqttoptions
        .set_keep_alive(Duration::from_secs(5));
   
    let (client, mut eventloop) = AsyncClient::new(mqttoptions,5 );
    task::block_on(client.subscribe("/upload/data", QoS::AtLeastOnce)).unwrap();
        // thread::spawn(move||
            loop{
            let event_result = task::block_on(eventloop.poll());
            match event_result {
                Ok(event)=>{
                    match event{
                        Event::Incoming(incoming)=> {
                            
                            if let Incoming::Publish(publish) = incoming.clone() {
                                println!("message {:?}", publish.topic);
                                let res=task::block_on(handel_topic(publish, "test1".to_string(), client.clone()));
                            }
                  
                        }
                        Event::Outgoing(outgoing)=>{
                            println!("outgoing {:?}",outgoing)
                        }
                    }
                }
                Err(e)=>{
                    println!("err {:?}",e);
                }
            }
        }`

[LongZoz]

When the program reaches this step, "let event_result = task::block_on(eventloop.poll()); If the client continues to send messages, the memory will continue to expand

[LongZoz]

It seems to have been successfully solved, the method is that I set the toml file parameter max_segment_size = 10000, and then the program memory will not expand, I go to apply this configuration to my program. I'll see if it affects the content and quality of the message

[swanandx]

That's good to know! Closing this as it isn't an bug / issue ( as described in the comments ). Feel free to comment if you wish!

Btw, just curious why you are using Async client instead of sync Client if you want to block on everything instead of calling .await?

Thanks for reporting 😁

[LongZoz]

Thank you for your reminder to modify toml!
Asynchronous is chosen to avoid unnecessary waiting and blocking on synchronous clients, aren't asynchronous clients more friendly

[swanandx]

Asynchronous is chosen to avoid unnecessary waiting and blocking on synchronous client

Yup! But in your code, you are using task::block_on which blocks right? Async will be beneficial if u use .await instead of block_on right?

e.g. client.subscribe("topic").await? note: that ? can be replaced with .unwrap() as per ur provided example

[LongZoz]

Yes, we do! Thank u. Those are two things I can improve on