fixed mutil event

Signed-off-by: Sven Wang <wanghancai@huawei.com>

services/distributeddataservice/adapter/communicator/src/device_manager_adapter.cpp

@@ -457,7 +457,11 @@ std::vector<std::string> DeviceManagerAdapter::ToUUID(const std::vector<std::str
 {
     std::vector<std::string> uuids;
     for (auto &device : devices) {
-        uuids.push_back(ToUUID(device));
+        auto uuid = ToUUID(device);
+        if (uuid.empty()) {
+            continue ;
+        }
+        uuids.push_back(std::move(uuid));
     }
     return uuids;
 }

services/distributeddataservice/app/src/kvstore_data_service.cpp

@@ -32,6 +32,8 @@
 #include "constant.h"
 #include "dds_trace.h"
 #include "device_manager_adapter.h"
+#include "device_matrix.h"
+#include "eventcenter/event_center.h"
 #include "executor_factory.h"
 #include "hap_token_info.h"
 #include "if_system_ability_manager.h"
@@ -56,7 +58,6 @@
 #include "utils/block_integer.h"
 #include "utils/converter.h"
 #include "utils/crypto.h"
-#include "device_matrix.h"
 
 namespace OHOS::DistributedKv {
 using namespace std::chrono;
@@ -217,6 +218,7 @@ int KvStoreDataService::Dump(int fd, const std::vector<std::u16string> &args)
 void KvStoreDataService::OnStart()
 {
     ZLOGI("distributeddata service onStart");
+    EventCenter::Defer defer;
     AccountDelegate::GetInstance()->RegisterHashFunc(Crypto::Sha256);
     static constexpr int32_t RETRY_TIMES = 50;
     static constexpr int32_t RETRY_INTERVAL = 500 * 1000; // unit is ms

services/distributeddataservice/framework/eventcenter/event.cpp

@@ -26,5 +26,10 @@ int32_t Event::GetEventId() const
 {
     return evtId_;
 }
+
+bool Event::Equals(const Event &) const
+{
+    return false;
+}
 } // namespace DistributedData
 } // namespace OHOS

services/distributeddataservice/framework/eventcenter/event_center.cpp

@@ -112,7 +112,7 @@ EventCenter::AsyncQueue &EventCenter::AsyncQueue::operator--()
         if (handler != handlers_.end()) {
             handler->second(*evt);
         }
-        events_.pop();
+        events_.pop_front();
     }
     depth_ = 0;
     return *this;
@@ -125,7 +125,16 @@ bool EventCenter::AsyncQueue::operator<=(int32_t depth) const
 
 void EventCenter::AsyncQueue::Post(std::unique_ptr<Event> evt)
 {
-    events_.push(std::move(evt));
+    for (auto &event : events_) {
+        if (event->GetEventId() != evt->GetEventId()) {
+            continue;
+        }
+
+        if (event->Equals(*evt)) {
+            return;
+        }
+    }
+    events_.push_back(std::move(evt));
 }
 
 void EventCenter::AsyncQueue::AddHandler(int32_t evtId, std::function<void(const Event &)> handler)

services/distributeddataservice/framework/include/eventcenter/event.h

@@ -35,7 +35,9 @@ public:
     API_EXPORT Event &operator=(Event &&) noexcept = delete;
     API_EXPORT Event &operator=(const Event &) = delete;
     API_EXPORT virtual ~Event();
+    API_EXPORT virtual bool Equals(const Event &) const;
     API_EXPORT int32_t GetEventId() const;
+
 private:
     int32_t evtId_;
 };

services/distributeddataservice/framework/include/eventcenter/event_center.h

@@ -56,7 +56,7 @@ private:
         void AddHandler(int32_t evtId, std::function<void(const Event &)> handler);
     private:
         std::map<int32_t, std::function<void(const Event &)>> handlers_;
-        std::queue<std::unique_ptr<Event>> events_;
+        std::deque<std::unique_ptr<Event>> events_;
         int32_t depth_ = 0;
     };
     ConcurrentMap<int32_t, std::list<std::function<void(const Event &)>>> observers_;

services/distributeddataservice/service/matrix/include/matrix_event.h

@@ -26,6 +26,7 @@ public:
     ~MatrixEvent() = default;
     uint16_t GetMask() const;
     std::string GetDeviceId() const;
+    bool Equals(const Event &event) const override;
 
 private:
     uint16_t mask_;

services/distributeddataservice/service/matrix/src/matrix_event.cpp

@@ -28,4 +28,10 @@ std::string MatrixEvent::GetDeviceId() const
 {
     return deviceId_;
 }
+
+bool MatrixEvent::Equals(const Event &event) const
+{
+    auto &evt = static_cast<const MatrixEvent &>(event);
+    return (deviceId_ == evt.deviceId_) && (mask_ == evt.mask_);
+}
 } // namespace OHOS::DistributedData