#include <ObserverManager.h>

Inheritance diagram for knowrob::ObserverManager:

Collaboration diagram for knowrob::ObserverManager:

Public Member Functions
	ObserverManager (const QueryableBackendPtr &backend)

	~ObserverManager ()

void	stop ()

ObserverPtr	observe (const GraphQueryPtr &query, const BindingsHandler &callback)

void	synchronize ()

void	stopObservation (const Observer &observer)

void	insert (const TripleContainerPtr &triples)

void	remove (const TripleContainerPtr &triples)

void	query (const GraphQueryPtr &query, const BindingsHandler &callback)

	ObserverManager (const QueryableBackendPtr &backend)

	~ObserverManager ()

void	stop ()

ObserverPtr	observe (const GraphQueryPtr &query, const BindingsHandler &callback)

void	synchronize ()

void	stopObservation (const Observer &observer)

void	insert (const TripleContainerPtr &triples)

void	remove (const TripleContainerPtr &triples)

void	query (const GraphQueryPtr &query, const BindingsHandler &callback)

Protected Member Functions
void	run ()

void	run ()

Protected Attributes
std::unique_ptr< Impl >	impl_

QueryableBackendPtr	backend_

Detailed Description

An observer manager is responsible for managing observers.

Definition at line 18 of file ObserverManager.h.

Constructor & Destructor Documentation

◆ ObserverManager() [1/2]

ObserverManager::ObserverManager ( const QueryableBackendPtr & backend )

explicit

Create an observer manager.

Parameters

backend the backend to observe.

Definition at line 40 of file ObserverManager.cpp.

         : impl_(std::make_unique<Impl>()), backend_(backend) {
     impl_->thread_ = std::thread(&ObserverManager::run, this);
 }

◆ ~ObserverManager() [1/2]

ObserverManager::~ObserverManager ( )

Definition at line 45 of file ObserverManager.cpp.

                                   {
     stop();
     if (impl_) {
         if (impl_->thread_.joinable()) {
             impl_->thread_.join();
         }
         impl_ = nullptr;
     }
 }

◆ ObserverManager() [2/2]

knowrob::ObserverManager::ObserverManager ( const QueryableBackendPtr & backend )

explicit

Create an observer manager.

Parameters

backend the backend to observe.

◆ ~ObserverManager() [2/2]

knowrob::ObserverManager::~ObserverManager ( )

Member Function Documentation

◆ insert() [1/2]

void ObserverManager::insert ( const TripleContainerPtr & triples )

Insert triples.

Parameters

triples the triples to insert.

Definition at line 106 of file ObserverManager.cpp.

                                                               {
     {
         std::lock_guard<std::mutex> lock(impl_->queueMutex_);
         impl_->queue_.push({Impl::Mode::INSERT, triples});
     }
     impl_->queueCondition_.notify_one();
 }

◆ insert() [2/2]

void knowrob::ObserverManager::insert ( const TripleContainerPtr & triples )

Insert triples.

Parameters

triples the triples to insert.

◆ observe() [1/2]

ObserverPtr ObserverManager::observe	(	const GraphQueryPtr &	query,
		const BindingsHandler &	callback
	)

Observe a query.

Parameters

query	the query to observe.
callback	the callback to invoke when the query matches.

Returns: the observer.

Definition at line 75 of file ObserverManager.cpp.

                                                                                                 {
     auto job = std::make_shared<ObserverJob>(shared_from_this(), query, callback);
     {
         std::lock_guard<std::mutex> lock(impl_->jobMutex_);
         impl_->jobs_.push_back(job);
     }
     return std::make_shared<Observer>(job);
 }

◆ observe() [2/2]

ObserverPtr knowrob::ObserverManager::observe	(	const GraphQueryPtr &	query,
		const BindingsHandler &	callback
	)

Observe a query.

Parameters

query	the query to observe.
callback	the callback to invoke when the query matches.

Returns: the observer.

◆ query() [1/2]

void ObserverManager::query	(	const GraphQueryPtr &	query,
		const BindingsHandler &	callback
	)

Query the backend.

Parameters

query	the query to execute.
callback	the callback to invoke with the results.

Definition at line 71 of file ObserverManager.cpp.

                                                                                        {
     backend_->query(query, callback);
 }

◆ query() [2/2]

void knowrob::ObserverManager::query	(	const GraphQueryPtr &	query,
		const BindingsHandler &	callback
	)

Query the backend.

Parameters

query	the query to execute.
callback	the callback to invoke with the results.

◆ remove() [1/2]

void ObserverManager::remove ( const TripleContainerPtr & triples )

Remove triples.

Parameters

triples the triples to remove.

Definition at line 114 of file ObserverManager.cpp.

                                                               {
     {
         std::lock_guard<std::mutex> lock(impl_->queueMutex_);
         impl_->queue_.push({Impl::Mode::REMOVE, triples});
     }
     impl_->queueCondition_.notify_one();
 }

◆ remove() [2/2]

void knowrob::ObserverManager::remove ( const TripleContainerPtr & triples )

Remove triples.

Parameters

triples the triples to remove.

◆ run() [1/2]

void ObserverManager::run ( )

protected

Definition at line 122 of file ObserverManager.cpp.

                           {
     while (impl_->running_) {
         std::pair<Impl::Mode,TripleContainerPtr> next;
         {
             std::unique_lock<std::mutex> lock(impl_->queueMutex_);
             if(impl_->running_) {
                 impl_->queueCondition_.wait(lock, [this] { return !impl_->running_ || !impl_->queue_.empty(); });
             }
             if (!impl_->running_) {
                 break;
             }
             next = impl_->queue_.front();
         }
         {
             std::lock_guard<std::mutex> lock(impl_->jobMutex_);
             if (next.first == Impl::Mode::INSERT) {
                 for (auto &job: impl_->jobs_) {
                     job->processInsertion(next.second);
                 }
             } else {
                 for (auto &job: impl_->jobs_) {
                     job->processRemoval(next.second);
                 }
             }
         }
         {
             std::unique_lock<std::mutex> lock(impl_->queueMutex_);
             impl_->queue_.pop();
             if (impl_->queue_.empty()) {
                 impl_->syncCondition_.notify_all();
             }
         }
     }
 }

◆ run() [2/2]

void knowrob::ObserverManager::run ( )

protected

◆ stop() [1/2]

void ObserverManager::stop ( )

Stop the observer manager.

Definition at line 55 of file ObserverManager.cpp.

                            {
     {
         std::lock_guard<std::mutex> lock(impl_->queueMutex_);
         while (!impl_->queue_.empty()) {
             impl_->queue_.pop();
         }
     }
     {
         std::lock_guard<std::mutex> lock(impl_->jobMutex_);
         impl_->jobs_.clear();
     }
     impl_->running_ = false;
     backend_ = nullptr;
     impl_->queueCondition_.notify_one();
 }

◆ stop() [2/2]

void knowrob::ObserverManager::stop ( )

Stop the observer manager.

◆ stopObservation() [1/2]

void ObserverManager::stopObservation ( const Observer & observer )

Stop observing a query.

Parameters

observer the observer to stop.

Definition at line 95 of file ObserverManager.cpp.

                                                               {
     std::lock_guard<std::mutex> lock(impl_->jobMutex_);
     for (auto it = impl_->jobs_.begin(); it != impl_->jobs_.end(); ++it) {
         auto &job = *it;
         if (job.get() == observer.job().get()) {
             impl_->jobs_.erase(it);
             break;
         }
     }
 }

◆ stopObservation() [2/2]

void knowrob::ObserverManager::stopObservation ( const Observer & observer )

Stop observing a query.

Parameters

observer the observer to stop.

◆ synchronize() [1/2]

void ObserverManager::synchronize ( )

Block until all observers have been synchronized.

Definition at line 84 of file ObserverManager.cpp.

                                   {
     {
         std::lock_guard<std::mutex> lock(impl_->queueMutex_);
         if (impl_->queue_.empty()) return;
     }
     {
         std::unique_lock<std::mutex> lock(impl_->syncMutex_);
         impl_->syncCondition_.wait(lock, [this] { return impl_->queue_.empty(); });
     }
 }

◆ synchronize() [2/2]

void knowrob::ObserverManager::synchronize ( )

Block until all observers have been synchronized.

Member Data Documentation

◆ backend_

QueryableBackendPtr knowrob::ObserverManager::backend_

protected

Definition at line 74 of file ObserverManager.h.

◆ impl_

std::unique_ptr< Impl > knowrob::ObserverManager::impl_

protected

Definition at line 73 of file ObserverManager.h.

The documentation for this class was generated from the following files:

build/_CPack_Packages/Linux/DEB/knowrob-2.1.0-Linux/usr/include/knowrob/storage/ObserverManager.h
src/storage/ObserverManager.cpp

Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ ObserverManager() [1/2]

◆ ~ObserverManager() [1/2]

◆ ObserverManager() [2/2]

◆ ~ObserverManager() [2/2]

Member Function Documentation

◆ insert() [1/2]

◆ insert() [2/2]

◆ observe() [1/2]

◆ observe() [2/2]

◆ query() [1/2]

◆ query() [2/2]

◆ remove() [1/2]

◆ remove() [2/2]

◆ run() [1/2]

◆ run() [2/2]

◆ stop() [1/2]

◆ stop() [2/2]

◆ stopObservation() [1/2]

◆ stopObservation() [2/2]

◆ synchronize() [1/2]

◆ synchronize() [2/2]

Member Data Documentation

◆ backend_

◆ impl_