2 Copyright (C) 2012 Intel Corporation
4 This library is free software; you can redistribute it and/or
5 modify it under the terms of the GNU Lesser General Public
6 License as published by the Free Software Foundation; either
7 version 2.1 of the License, or (at your option) any later version.
9 This library is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 Lesser General Public License for more details.
14 You should have received a copy of the GNU Lesser General Public
15 License along with this library; if not, write to the Free Software
16 Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
23 #include "listplusplus.h"
26 using namespace std::placeholders;
31 static int PPSUpdate(void* data)
33 Core::Performance *performance = (Core::Performance*)data;
35 if(performance->propertiesPerSecond > 0 && performance->propertiesPerSecond != lastpps)
37 lastpps = performance->propertiesPerSecond;
38 DebugOut(1)<<"Property updates per second: "<<performance->propertiesPerSecond<<endl;
41 performance->propertiesPerSecond = 0;
43 if(performance->firedPropertiesPerSecond > 0 && performance->firedPropertiesPerSecond != lastfpps)
45 lastfpps = performance->firedPropertiesPerSecond;
46 DebugOut(1)<<"Fired property updates per second: "<<performance->firedPropertiesPerSecond<<endl;
49 performance->firedPropertiesPerSecond = 0;
54 Core::Core(std::map<string, string> config): AbstractRoutingEngine(config), handleCount(0)
56 g_timeout_add(1000,PPSUpdate,&performance);
58 auto simpleCb = [this](amb::Queue<AbstractPropertyType*, amb::PropertyCompare>* q)
62 AbstractPropertyType* value = q->pop();
63 updateProperty(value);
71 if(config.find("highPriorityQueueSize") != config.end())
73 hpqs = boost::lexical_cast<int, std::string>(config["highPriorityQueueSize"]);
76 if(config.find("normalPriorityQueueSize") != config.end())
78 npqs = boost::lexical_cast<int, std::string>(config["normalPriorityQueueSize"]);
81 if(config.find("lowPriorityQueueSize") != config.end())
83 lpqs = boost::lexical_cast<int, std::string>(config["lowPriorityQueueSize"]);
86 watcherPtr = new amb::AsyncQueueWatcher<AbstractPropertyType*, amb::PropertyCompare>(&updatePropertyQueue, simpleCb, npqs);
87 watcherPtrLow = new amb::AsyncQueueWatcher<AbstractPropertyType*, amb::PropertyCompare>(&updatePropertyQueueLow,
89 AbstractPropertyType::Low);
90 watcherPtrHigh = new amb::AsyncQueueWatcher<AbstractPropertyType*, amb::PropertyCompare>(&updatePropertyQueueHigh,
92 AbstractPropertyType::High);
100 for(auto itr = mSinks.begin(); itr != mSinks.end(); ++itr)
103 itr = mSinks.begin();
108 void Core::registerSource(AbstractSource *source)
110 mSources.insert(source);
113 void Core::updateSupported(PropertyList added, PropertyList removed, AbstractSource* source)
115 if(!source || mSources.find(source) == mSources.end())
118 /// add the newly supported to master list
121 handleAddSupported(added, source);
123 /// removed no longer supported properties from master list.
125 handleRemoveSupported(removed, source);
127 /// tell all sinks about the newly supported properties.
129 PropertyList s = supported();
131 if(!s.size()) return;
133 for(auto sink : mSinks)
135 sink->supportedChanged(s);
139 PropertyList Core::supported()
141 PropertyList supportedProperties;
143 std::transform(mMasterPropertyList.begin(), mMasterPropertyList.end(), std::back_inserter(supportedProperties),
144 [](const std::multimap<AbstractSource*, VehicleProperty::Property>::value_type& itr)
149 // remove duplicates:
150 std::sort(supportedProperties.begin(), supportedProperties.end());
151 auto itr = std::unique(supportedProperties.begin(), supportedProperties.end());
153 supportedProperties.erase(itr,supportedProperties.end());
155 return supportedProperties;
158 void Core::updateProperty(AbstractPropertyType *value, const string & uuid)
160 if(value->sourceUuid != uuid)
162 value->sourceUuid = uuid;
165 if(value->priority == AbstractPropertyType::Instant)
166 updateProperty(value);
167 else if(value->priority == AbstractPropertyType::High)
169 value->destroyed.push_back([this](AbstractPropertyType* v)
171 updatePropertyQueueHigh.remove(v);
173 updatePropertyQueueHigh.append(value);
175 else if(value->priority == AbstractPropertyType::Normal)
177 value->destroyed.push_back([this](AbstractPropertyType* v)
179 updatePropertyQueue.remove(v);
181 updatePropertyQueue.append(value);
183 else if(value->priority == AbstractPropertyType::Low)
185 value->destroyed.push_back([this](AbstractPropertyType* v)
187 updatePropertyQueueLow.remove(v);
189 updatePropertyQueueLow.append(value);
193 void Core::updateProperty(AbstractPropertyType * value)
195 VehicleProperty::Property & property = value->name;
196 const string & uuid = value->sourceUuid;
198 performance.propertiesPerSecond++;
200 //auto sinks = propertySinkMap[property]; !!! this will insert empty std::set<AbstractSink*> into propertySinkMap !!!
201 auto filteredSourceSinkMapIt = propertySinkMap.find(property);
202 auto cbMapItr = propertyCbMap.find(property);
204 if(filteredSourceSinkMapIt != propertySinkMap.end())
206 const FilteredSourceSinkMap & filteredSourceSinks = filteredSourceSinkMapIt->second;
208 DebugOut()<<__FUNCTION__<<"() there are "<<filteredSourceSinks.size()<<" sinks connected to property: "<<property<<endl;
210 performance.firedPropertiesPerSecond++;
212 for(auto itr = filteredSourceSinks.begin(); itr != filteredSourceSinks.end(); ++itr)
214 AbstractSink* sink = itr->first;
215 const std::string & sourceUuid = itr->second;
217 bool isFiltered = !sourceUuid.empty();
221 DebugOut()<<"Property ("<<property<<") for sink is filtered for source: "<<sourceUuid<<endl;
224 if( !isFiltered || sourceUuid == uuid)
226 sink->propertyChanged(value);
232 DebugOut()<<__FUNCTION__<<"() there are no sinks connected to property: "<<property<<endl;
235 if(cbMapItr != propertyCbMap.end())
237 FilteredSourceCbMap cbs = (*cbMapItr).second;
241 uint handle = itr.first;
242 const std::string& sourceUuid = itr.second;
244 bool isFiltered = !sourceUuid.empty();
248 DebugOut()<<"Property ("<<property<<") for cb is filtered for source: "<<sourceUuid<<endl;
251 if( !isFiltered || sourceUuid == uuid)
253 if(handleCbMap.count(handle))
255 auto cb = handleCbMap[handle];
262 DebugOut(DebugOut::Warning)<<"Failed to call callback subscribed to property: "<<property<<endl;
270 DebugOut()<<__FUNCTION__<<"() there are no cb connected to property: "<<property<<endl;
275 void Core::registerSink(AbstractSink *self)
280 void Core::unregisterSink(AbstractSink *self)
285 AsyncPropertyReply *Core::getPropertyAsync(AsyncPropertyRequest request)
287 AbstractSource* source = sourceForProperty(request.property, request.sourceUuidFilter);
289 AsyncPropertyReply* reply = new AsyncPropertyReply(request);
291 if(!source || ((source->supportedOperations() & AbstractSource::Get) != AbstractSource::Get)) { // not found or doesn't support AbstractSource::Get
292 // Don't wait until timer expire, complete with error here.
293 reply->error = AsyncPropertyReply::InvalidOperation;
294 if(request.completed)
295 request.completed(reply);
298 source->getPropertyAsync(reply);
302 /** right now the owner of the reply becomes the requestor that called this method.
303 * reply will become invalid after the first reply. */
307 void Core::getRangePropertyAsync(AsyncRangePropertyRequest request)
309 AsyncRangePropertyReply * reply = new AsyncRangePropertyReply(request);
311 bool anySupport = false;
312 for(auto src : mSources)
314 if(((src->supportedOperations() & AbstractSource::GetRanged) == AbstractSource::GetRanged))
317 src->getRangePropertyAsync(reply);
323 reply->success = false;
324 reply->error = AsyncPropertyReply::InvalidOperation;
325 reply->completed(reply);
329 AsyncPropertyReply * Core::setProperty(AsyncSetPropertyRequest request)
331 AbstractSource* src = sourceForProperty(request.property, request.sourceUuidFilter);
333 if(src && ((src->supportedOperations() & AbstractSource::Set) == AbstractSource::Set))
334 return src->setProperty(request);
336 DebugOut(DebugOut::Warning)<<"Error: setProperty opration failed. Property may not be supported: "<<request.property<<endl;
340 bool Core::subscribeToProperty(const VehicleProperty::Property &property, AbstractSink* sink)
342 auto sinksIt = propertySinkMap.find(property);
343 if(sinksIt != propertySinkMap.end() && sinksIt->second.find(sink) != sinksIt->second.end())
345 DebugOut(1)<<__FUNCTION__<<" property " << property << " has already been subscribed." << endl;
349 DebugOut(1)<<"Subscribing to: "<<property<<endl;
351 bool subscribed(false);
352 auto itr = mMasterPropertyList.begin();
353 while(itr != mMasterPropertyList.end())
355 VehicleProperty::Property prop = itr->second;
356 if(prop == property) {
357 AbstractSource* src = itr->first;
358 src->subscribeToPropertyChanges(property);
359 // Move to next source. It will skip all the remaining properties in this source.
360 itr = mMasterPropertyList.upper_bound(src);
369 propertySinkMap[property].emplace(sink, std::string(""));
374 bool Core::subscribeToProperty(const VehicleProperty::Property &property, const string &sourceUuidFilter, AbstractSink *sink)
376 auto sinksIt = propertySinkMap.find(property);
377 if(sinksIt != propertySinkMap.end() && sinksIt->second.find(sink) != sinksIt->second.end())
379 DebugOut(1)<<__FUNCTION__<<" property " << property << " has already been subscribed." << endl;
383 DebugOut(1)<<"Subscribing to: "<<property<<endl;
385 AbstractSource* src = sourceForProperty(property, sourceUuidFilter);
389 propertySinkMap[property].emplace(sink, sourceUuidFilter);
391 src->subscribeToPropertyChanges(property);
396 bool Core::subscribeToProperty(const VehicleProperty::Property &, const string & sourceUuidFilter, Zone::Type zoneFilter, AbstractSink *sink)
399 throw std::runtime_error("Not implemented");
402 uint Core::subscribeToProperty(const VehicleProperty::Property &property, AbstractRoutingEngine::PropertyChangedType cb, std::string pid)
404 DebugOut(1)<<"Subscribing to: "<<property<<endl;
406 auto itr = mMasterPropertyList.begin();
407 while(itr != mMasterPropertyList.end())
409 VehicleProperty::Property prop = itr->second;
410 if(prop == property) {
411 AbstractSource* src = itr->first;
412 src->subscribeToPropertyChanges(property);
413 // Move to next source. It will skip all the remaining properties in this source.
414 itr = mMasterPropertyList.upper_bound(src);
421 handleCbMap[++handleCount] = cb;
422 propertyCbMap[property].emplace(handleCount, std::string(""));
426 bool Core::unsubscribeToProperty(const VehicleProperty::Property & property, AbstractSink* sink)
428 auto sinksIt = propertySinkMap.find(property);
429 if(sinksIt == propertySinkMap.end())
431 DebugOut(1)<<__FUNCTION__<<" property not subscribed to: "<<property<<endl;
435 sinksIt->second.erase(sink);
437 /// Now we check to see if this is the last subscriber
438 if(sinksIt->second.empty())
440 propertySinkMap.erase(sinksIt);
441 auto itr = mMasterPropertyList.begin();
442 while(itr != mMasterPropertyList.end())
444 if(itr->second == property) {
445 AbstractSource* src = itr->first;
446 src->unsubscribeToPropertyChanges(property);
447 // Move to next source. It will skip all the remaining properties in this source.
448 itr = mMasterPropertyList.upper_bound(src);
459 void Core::unsubscribeToProperty(uint handle)
461 handleCbMap.erase(handle);
462 /// TODO: unsubscribe from source
465 PropertyInfo Core::getPropertyInfo(const VehicleProperty::Property &property, const string &sourceUuid)
468 return PropertyInfo::invalid();
470 auto srcs = sourcesForProperty(property);
472 if(!contains(srcs, sourceUuid))
473 return PropertyInfo::invalid();
475 auto theSource = find_if(mSources.begin(), mSources.end(),[&sourceUuid](const std::set<AbstractSource*>::value_type & itr)
477 return (itr)->uuid() == sourceUuid;
480 return (*theSource)->getPropertyInfo(property);
483 std::vector<string> Core::sourcesForProperty(const VehicleProperty::Property & property)
485 std::vector<std::string> list;
487 for(auto src : mSources)
489 if(contains(src->supported(), property))
490 list.push_back(src->uuid());
496 void Core::inspectSupported()
498 for(AbstractSource* src : mSources)
500 updateSupported(src->supported(), PropertyList(), src);
504 void Core::handleAddSupported(const PropertyList& added, AbstractSource* source)
507 throw std::runtime_error("Core::handleAddSupported passed a null source");
509 if(!contains(mSources, source))
511 mSources.insert(source);
514 for(auto property : added)
516 if(!sourceForProperty(property, source->uuid()))
517 mMasterPropertyList.emplace(source, property);
519 // Subscribe to property in a new source if such property was subscribed. This catches newly supported properties in the process.
520 if( propertySinkMap.find(property) != propertySinkMap.end()){
521 source->subscribeToPropertyChanges(property);
526 void Core::handleRemoveSupported(const PropertyList& removed, AbstractSource* source)
531 auto range = mMasterPropertyList.equal_range(source);
532 for(auto itr = removed.begin(); itr != removed.end(); ++itr)
535 // TODO: We do not have info about all subscribed sources in
536 // std::unordered_map<VehicleProperty::Property, std::map<AbstractSink*, std::string> > propertySinkMap
537 // so we do not know if we can/should remove property from propertySinkMap,
538 // but I suppose this should be handled by each AbstractSink implementation in a callback AbstractSink::supportedChanged().
540 const VehicleProperty::Property property(*itr);
543 range.first, // the first property in source
544 range.second, // one item right after the last property in source
545 [&property](const std::multimap<AbstractSource*, VehicleProperty::Property>::value_type& it) { return it.second == property; }
548 if (it != range.second)// property was found
550 mMasterPropertyList.erase(it);// References and iterators to the erased elements are invalidated. Other iterators and references are not invalidated.
552 // TODO: Do we need to unsubscribe here ???
557 AbstractSource* Core::sourceForProperty(const VehicleProperty::Property& property, const std::string& sourceUuidFilter) const
559 auto it = mMasterPropertyList.end();
560 if(sourceUuidFilter.empty()){
561 it = std::find_if(mMasterPropertyList.begin(), mMasterPropertyList.end(),
562 [&property](const std::multimap<AbstractSource*, VehicleProperty::Property>::value_type& it) { return it.second == property; }
566 auto itSource = find_if(mSources.begin(),mSources.end(),[&sourceUuidFilter](const std::set<AbstractSource*>::value_type & it)
568 return (it)->uuid() == sourceUuidFilter;
570 if(itSource != mSources.end()){
571 auto range = mMasterPropertyList.equal_range(*itSource);
573 range.first, // the first property in source
574 range.second, // one item right after the last property in source
575 [&property](const std::multimap<AbstractSource*, VehicleProperty::Property>::value_type& it)
577 return it.second == property; }
580 if (temp != range.second)// property was found
585 if(it == mMasterPropertyList.end())