knowrob::mongo::QueryWatch Class Reference

#include <QueryWatch.h>

Public Member Functions
	QueryWatch ()
	QueryWatch (const QueryWatch &)=delete
	~QueryWatch ()
void	setWatchRate (uint32_t rate)
long	watch (const std::shared_ptr< Collection > &collection, const bson_t *query, const ChangeStreamCallback &callback)
void	unwatch (long watcher_id)
	QueryWatch ()
	QueryWatch (const QueryWatch &)=delete
	~QueryWatch ()
void	setWatchRate (uint32_t rate)
long	watch (const std::shared_ptr< Collection > &collection, const bson_t *query, const ChangeStreamCallback &callback)
void	unwatch (long watcher_id)

Protected Member Functions
void	startWatchThread ()
void	stopWatchThread ()
void	loop ()
void	startWatchThread ()
void	stopWatchThread ()
void	loop ()

Protected Attributes
std::map< long, std::unique_ptr< ChangeStream > >	watcher_map_
std::thread *	thread_
bool	isRunning_
std::mutex	lock_
uint32_t	watchRate_

Static Protected Attributes
static std::atomic< long >	id_counter_ = 0

Detailed Description

Keeps track over time of query results and notifies a callback for each new result.

Definition at line 26 of file QueryWatch.h.

Constructor & Destructor Documentation

QueryWatch() [1/4]

QueryWatch::QueryWatch

(

)

Definition at line 16 of file QueryWatch.cpp.

        : thread_(nullptr),
          isRunning_(false),
          watchRate_(200) {
}

QueryWatch() [2/4]

knowrob::mongo::QueryWatch::QueryWatch ( const QueryWatch &  )

delete

~QueryWatch() [1/2]

QueryWatch::~QueryWatch

(

)

Definition at line 22 of file QueryWatch.cpp.

                        {
    // first let the thread terminate
    stopWatchThread();
    // then stop all watcher
    watcher_map_.clear();
}

QueryWatch() [3/4]

knowrob::mongo::QueryWatch::QueryWatch

(

)

QueryWatch() [4/4]

knowrob::mongo::QueryWatch::QueryWatch ( const QueryWatch &  )

delete

~QueryWatch() [2/2]

knowrob::mongo::QueryWatch::~QueryWatch

(

)

Member Function Documentation

loop() [1/2]

void QueryWatch::loop ( )

protected

Definition at line 73 of file QueryWatch.cpp.

                      {
    // loop as long isRunning_=true
    auto next = std::chrono::system_clock::now();
    while (isRunning_) {
        {
            std::lock_guard<std::mutex> guard(lock_);
            for (auto &it: watcher_map_) {
                try {
                    it.second->next();
                }
                catch (MongoException &exc) {
                    KB_WARN("exception in mongo watch: {}", exc.what());
                }
            }
        }
        // try to run at constant rate
        next += std::chrono::milliseconds(watchRate_);
        std::this_thread::sleep_until(next);
    }
}

loop() [2/2]

void knowrob::mongo::QueryWatch::loop ( )

protected

setWatchRate() [1/2]

void knowrob::mongo::QueryWatch::setWatchRate ( uint32_t  rate )

inline

The query watch actively polls change streams in this interval.

Parameters

rate	the rate in milliseconds.

Definition at line 38 of file QueryWatch.h.

{ watchRate_ = rate; }

setWatchRate() [2/2]

void knowrob::mongo::QueryWatch::setWatchRate ( uint32_t  rate )

inline

The query watch actively polls change streams in this interval.

Parameters

rate	the rate in milliseconds.

Definition at line 38 of file QueryWatch.h.

{ watchRate_ = rate; }

startWatchThread() [1/2]

void QueryWatch::startWatchThread ( )

protected

Definition at line 29 of file QueryWatch.cpp.

                                  {
    std::lock_guard<std::mutex> guard(lock_);
    if (!isRunning_ && !thread_) {
        isRunning_ = true;
        thread_ = new std::thread(&QueryWatch::loop, this);
    }
}

startWatchThread() [2/2]

void knowrob::mongo::QueryWatch::startWatchThread ( )

protected

stopWatchThread() [1/2]

void QueryWatch::stopWatchThread ( )

protected

Definition at line 37 of file QueryWatch.cpp.

                                 {
    std::lock_guard<std::mutex> guard(lock_);
    if (isRunning_ && thread_) {
        isRunning_ = false;
        thread_->join();
        delete thread_;
        thread_ = nullptr;
    }
}

stopWatchThread() [2/2]

void knowrob::mongo::QueryWatch::stopWatchThread ( )

protected

unwatch() [1/2]

void QueryWatch::unwatch

(

long

watcher_id

)

Stop watching a collection.

Parameters

watcher_id

the watcher id returned by watch.

Definition at line 62 of file QueryWatch.cpp.

                                        {
    auto needle = watcher_map_.find(watcher_id);
    if (needle != watcher_map_.end()) {
        std::lock_guard<std::mutex> guard(lock_);
        watcher_map_.erase(needle);
    }
    if (watcher_map_.empty()) {
        stopWatchThread();
    }
}

unwatch() [2/2]

void knowrob::mongo::QueryWatch::unwatch

(

long

watcher_id

)

Stop watching a collection.

Parameters

watcher_id

the watcher id returned by watch.

watch() [1/2]

long QueryWatch::watch	(	const std::shared_ptr< Collection > &	collection,
		const bson_t *	query,
		const ChangeStreamCallback &	callback
	)

Watch for changes in a collection.

Parameters

collection	the collection to watch.
query	the query to watch.
callback	the callback to invoke for each change.

Returns

a watcher id that can be used to unwatch.

Definition at line 47 of file QueryWatch.cpp.

                                              {
    auto next_id = (id_counter_++);
    // add to map
    {
        std::lock_guard<std::mutex> guard(lock_);
        watcher_map_.emplace(next_id, std::make_unique<ChangeStream>(collection, query, callback));
    }
    // start the thread when the first watch is added
    startWatchThread();
    return next_id;
}

watch() [2/2]

long knowrob::mongo::QueryWatch::watch	(	const std::shared_ptr< Collection > &	collection,
		const bson_t *	query,
		const ChangeStreamCallback &	callback
	)

Watch for changes in a collection.

Parameters

collection	the collection to watch.
query	the query to watch.
callback	the callback to invoke for each change.

Returns

a watcher id that can be used to unwatch.

Member Data Documentation

id_counter_

std::atomic< long > QueryWatch::id_counter_ = 0

staticprotected

Definition at line 64 of file QueryWatch.h.

isRunning_

bool knowrob::mongo::QueryWatch::isRunning_

protected

Definition at line 61 of file QueryWatch.h.

lock_

std::mutex knowrob::mongo::QueryWatch::lock_

protected

Definition at line 62 of file QueryWatch.h.

thread_

std::thread * knowrob::mongo::QueryWatch::thread_

protected

Definition at line 60 of file QueryWatch.h.

watcher_map_

std::map< long, std::unique_ptr< ChangeStream > > knowrob::mongo::QueryWatch::watcher_map_

protected

Definition at line 58 of file QueryWatch.h.

watchRate_

uint32_t knowrob::mongo::QueryWatch::watchRate_

protected

Definition at line 63 of file QueryWatch.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/mongo/QueryWatch.h
• src/storage/mongo/QueryWatch.cpp