1 files changed, 259 insertions, 0 deletions
diff --git a/comphelper/source/misc/asyncnotification.cxx b/comphelper/source/misc/asyncnotification.cxx
new file mode 100644
index 000000000..cb8a2f251
--- /dev/null
+++ b/comphelper/source/misc/asyncnotification.cxx
@@ -0,0 +1,259 @@
+/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+/*
+ * This file is part of the LibreOffice project.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ *
+ * This file incorporates work covered by the following license notice:
+ *
+ *   Licensed to the Apache Software Foundation (ASF) under one or more
+ *   contributor license agreements. See the NOTICE file distributed
+ *   with this work for additional information regarding copyright
+ *   ownership. The ASF licenses this file to you under the Apache
+ *   License, Version 2.0 (the "License"); you may not use this file
+ *   except in compliance with the License. You may obtain a copy of
+ *   the License at http://www.apache.org/licenses/LICENSE-2.0 .
+ */
+
+#include <comphelper/asyncnotification.hxx>
+#include <comphelper/scopeguard.hxx>
+#include <mutex>
+#include <condition_variable>
+
+#include <cassert>
+#include <stdexcept>
+#include <vector>
+#include <algorithm>
+
+namespace comphelper
+{
+    AnyEvent::AnyEvent()
+    {
+    }
+
+    AnyEvent::~AnyEvent()
+    {
+    }
+
+    namespace {
+
+    struct ProcessableEvent
+    {
+        AnyEventRef                         aEvent;
+        ::rtl::Reference< IEventProcessor > xProcessor;
+    };
+
+    struct EqualProcessor
+    {
+        const ::rtl::Reference< IEventProcessor >&  rProcessor;
+        explicit EqualProcessor( const ::rtl::Reference< IEventProcessor >& _rProcessor ) :rProcessor( _rProcessor ) { }
+
+        bool operator()( const ProcessableEvent& _rEvent )
+        {
+            return _rEvent.xProcessor.get() == rProcessor.get();
+        }
+    };
+
+    }
+
+    struct EventNotifierImpl
+    {
+        std::mutex        aMutex;
+        std::condition_variable aPendingActions;
+        std::vector< ProcessableEvent > aEvents;
+        bool                bTerminate;
+        // only used for AsyncEventNotifierAutoJoin
+        char const*         name;
+        std::shared_ptr<AsyncEventNotifierAutoJoin> pKeepThisAlive;
+
+        EventNotifierImpl()
+            : bTerminate(false)
+            , name(nullptr)
+        {
+        }
+    };
+
+    AsyncEventNotifierBase::AsyncEventNotifierBase()
+        : m_xImpl(new EventNotifierImpl)
+    {
+    }
+
+
+    AsyncEventNotifierBase::~AsyncEventNotifierBase()
+    {
+    }
+
+
+    void AsyncEventNotifierBase::removeEventsForProcessor( const ::rtl::Reference< IEventProcessor >& _xProcessor )
+    {
+        std::scoped_lock aGuard( m_xImpl->aMutex );
+
+        // remove all events for this processor
+        m_xImpl->aEvents.erase(std::remove_if( m_xImpl->aEvents.begin(), m_xImpl->aEvents.end(), EqualProcessor( _xProcessor ) ), m_xImpl->aEvents.end());
+    }
+
+
+    void SAL_CALL AsyncEventNotifierBase::terminate()
+    {
+        std::scoped_lock aGuard( m_xImpl->aMutex );
+
+        // remember the termination request
+        m_xImpl->bTerminate = true;
+
+        // awake the thread
+        m_xImpl->aPendingActions.notify_all();
+    }
+
+
+    void AsyncEventNotifierBase::addEvent( const AnyEventRef& _rEvent, const ::rtl::Reference< IEventProcessor >& _xProcessor )
+    {
+        std::scoped_lock aGuard( m_xImpl->aMutex );
+
+        // remember this event
+        m_xImpl->aEvents.emplace_back( ProcessableEvent {_rEvent, _xProcessor} );
+
+        // awake the thread
+        m_xImpl->aPendingActions.notify_all();
+    }
+
+
+    void AsyncEventNotifierBase::execute()
+    {
+        for (;;)
+        {
+            std::vector< ProcessableEvent > aEvents;
+            {
+                std::unique_lock aGuard(m_xImpl->aMutex);
+                m_xImpl->aPendingActions.wait(aGuard,
+                    [this] { return m_xImpl->bTerminate || !m_xImpl->aEvents.empty(); } );
+                if (m_xImpl->bTerminate)
+                    return;
+                else
+                    std::swap(aEvents, m_xImpl->aEvents);
+            }
+            for (ProcessableEvent& rEvent : aEvents)
+            {
+                assert(rEvent.xProcessor.is());
+                rEvent.xProcessor->processEvent(*rEvent.aEvent);
+            }
+            aEvents.clear();
+        }
+    }
+
+    AsyncEventNotifier::AsyncEventNotifier(char const* name)
+        : salhelper::Thread(name)
+    {
+    }
+
+    AsyncEventNotifier::~AsyncEventNotifier()
+    {
+    }
+
+    void AsyncEventNotifier::execute()
+    {
+        return AsyncEventNotifierBase::execute();
+    }
+
+    void AsyncEventNotifier::terminate()
+    {
+        return AsyncEventNotifierBase::terminate();
+    }
+
+    namespace {
+
+    std::mutex& GetTheNotifiersMutex()
+    {
+        static std::mutex MUTEX;
+        return MUTEX;
+    }
+
+    }
+
+    static std::vector<std::weak_ptr<AsyncEventNotifierAutoJoin>> g_Notifiers;
+
+    void JoinAsyncEventNotifiers()
+    {
+        std::vector<std::weak_ptr<AsyncEventNotifierAutoJoin>> notifiers;
+        {
+            std::scoped_lock g(GetTheNotifiersMutex());
+            notifiers = g_Notifiers;
+        }
+        for (std::weak_ptr<AsyncEventNotifierAutoJoin> const& wNotifier : notifiers)
+        {
+            std::shared_ptr<AsyncEventNotifierAutoJoin> const pNotifier(
+                    wNotifier.lock());
+            if (pNotifier)
+            {
+                pNotifier->terminate();
+                pNotifier->join();
+            }
+        }
+        // note it's possible that g_Notifiers isn't empty now in case of leaks,
+        // particularly since the UNO service manager isn't disposed yet
+    }
+
+    AsyncEventNotifierAutoJoin::AsyncEventNotifierAutoJoin(char const* name)
+    {
+        m_xImpl->name = name;
+    }
+
+    AsyncEventNotifierAutoJoin::~AsyncEventNotifierAutoJoin()
+    {
+        std::scoped_lock g(GetTheNotifiersMutex());
+        // note: this doesn't happen atomically with the refcount
+        // hence it's possible this deletes > 1 or 0 elements
+        g_Notifiers.erase(
+            std::remove_if(g_Notifiers.begin(), g_Notifiers.end(),
+                [](std::weak_ptr<AsyncEventNotifierAutoJoin> const& w) {
+                    return w.expired();
+                } ),
+            g_Notifiers.end());
+    }
+
+    std::shared_ptr<AsyncEventNotifierAutoJoin>
+    AsyncEventNotifierAutoJoin::newAsyncEventNotifierAutoJoin(char const* name)
+    {
+        std::shared_ptr<AsyncEventNotifierAutoJoin> const ret(
+                new AsyncEventNotifierAutoJoin(name));
+        std::scoped_lock g(GetTheNotifiersMutex());
+        g_Notifiers.push_back(ret);
+        return ret;
+    }
+
+    void AsyncEventNotifierAutoJoin::terminate()
+    {
+        return AsyncEventNotifierBase::terminate();
+    }
+
+    void AsyncEventNotifierAutoJoin::launch(std::shared_ptr<AsyncEventNotifierAutoJoin> const& xThis)
+    {
+        // see salhelper::Thread::launch
+        xThis->m_xImpl->pKeepThisAlive = xThis;
+        comphelper::ScopeGuard g([&xThis] { xThis->m_xImpl->pKeepThisAlive.reset(); });
+        if (!xThis->create()) {
+            throw std::runtime_error("osl::Thread::create failed");
+        }
+        g.dismiss();
+    }
+
+    void AsyncEventNotifierAutoJoin::run()
+    {
+        // see salhelper::Thread::run
+        comphelper::ScopeGuard g([this] { onTerminated(); });
+        setName(m_xImpl->name);
+        execute();
+        g.dismiss();
+    }
+
+    void AsyncEventNotifierAutoJoin::onTerminated()
+    {
+        // try to delete "this"
+        m_xImpl->pKeepThisAlive.reset();
+    }
+
+} // namespace comphelper
+
+
+/* vim:set shiftwidth=4 softtabstop=4 expandtab: */