summaryrefslogtreecommitdiffstats
path: root/comphelper/source/misc/asyncnotification.cxx
diff options
context:
space:
mode:
Diffstat (limited to 'comphelper/source/misc/asyncnotification.cxx')
-rw-r--r--comphelper/source/misc/asyncnotification.cxx259
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 0000000000..cb8a2f2511
--- /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: */