AbstractMqttsnMessageStateServices.scheduleWork() stops working.

[canatella]

Hello,

I'm testing the gateway with devices that are sending 4k packets every 350ms. It seems to happen randomly but at some point, the futures in the AbstractMqttsnMessageStateServices.scheduleWork() stop working because the process variable is false due to the isDone() method returning true. Looking deeper at the code:

        boolean process = !flushOperations.containsKey(context) ||
                !flushOperations.get(context).isDone();
        ...
        if(process){
               ...
               executorService.schedule(...

The code says if there's no current flushOperation or if there is one but it is not done, schedule a new flush operation. It seems to me the intention is that, if there no current flushOperations or if there is one but it is done, schedule a new one. Am I misunderstanding something?

Also, there seems to be some synchronized access to the flushOperations map and some are not, like line 113. is this on purpose? Why not use a concurrent map?

Thanks!

Edit: I think I might be misunderstanding what that method is supposed to do. I'm thinking it scheduleWork should schedule a flush if there is none already scheduled. When the flush job start, it should check for something but I'm not sure about what. Should it check if there has been a newer operation scheduled and not call process queue in that case?