drifty/comaps
1
0
Fork 0
mirror of https://codeberg.org/comaps/comaps synced 2026-01-03 19:33:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

[traffxml] Make sources pluggable

Browse Source 
Signed-off-by: mvglasow <michael -at- vonglasow.com>
This commit is contained in:
mvglasow
2025-07-03 23:22:17 +03:00
parent f32493faaa
commit 6656c7e441
7 changed files with 658 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
map/framework.cpp
View File

@@ -23,6 +23,8 @@
#include "storage/routing_helpers.hpp"
#include "storage/storage_helpers.hpp"
#include "traffxml/traff_source.hpp"
#include "drape_frontend/color_constants.hpp"
#include "drape_frontend/gps_track_point.hpp"
#include "drape_frontend/visual_params.hpp"
@@ -392,6 +394,12 @@ Framework::Framework(FrameworkParams const & params, bool loadMaps)
LoadMapsSync();
m_trafficManager.SetEnabled(LoadTrafficEnabled());
/*
* MockTraffSource for debugging purposes.
* TODO Replace with a real source, parametrized and conditionally loaded, once we have one.
*/
traffxml::MockTraffSource::Create(m_trafficManager);
}
Framework::~Framework()

154
map/traffic_manager.cpp
View File

@@ -439,78 +439,31 @@ void TrafficManager::UpdateViewport(ScreenBase const & screen)
UpdateActiveMwms(screen.ClipRect(), m_lastDrapeMwmsByRect, m_activeDrapeMwms);
}
std::string TrafficManager::GetMwmFilters(std::set<MwmSet::MwmId> & mwms)
{
std::vector<m2::RectD> rects;
for (auto mwmId : mwms)
rects.push_back(mwmId.GetInfo()->m_bordersRect);
return traffxml::FiltersToXml(rects);
}
// TODO make this work with multiple sources (e.g. Android)
bool TrafficManager::Subscribe(std::set<MwmSet::MwmId> & mwms)
{
// TODO what if we’re subscribed already?
std::string filterList = GetMwmFilters(mwms);
// TODO
LOG(LINFO, ("Would subscribe to:\n", filterList));
m_subscriptionId = "placeholder_subscription_id";
m_isPollNeeded = true; // would be false if we got a feed here
return true;
}
// TODO make this work with multiple sources (e.g. Android)
bool TrafficManager::ChangeSubscription(std::set<MwmSet::MwmId> & mwms)
{
if (!IsSubscribed())
return false;
std::string filterList = GetMwmFilters(mwms);
// TODO
LOG(LINFO, ("Would change subscription", m_subscriptionId, "to:\n", filterList));
m_isPollNeeded = true; // would be false if we got a feed here
return true;
}
bool TrafficManager::SetSubscriptionArea()
void TrafficManager::SubscribeOrChangeSubscription()
{
std::set<MwmSet::MwmId> activeMwms;
if (!IsSubscribed())
if (m_activeMwmsChanged)
{
{
std::lock_guard<std::mutex> lock(m_mutex);
m_activeMwmsChanged = false;
UniteActiveMwms(activeMwms);
}
if (!Subscribe(activeMwms))
return false;
}
else if (m_activeMwmsChanged)
{
{
std::lock_guard<std::mutex> lock(m_mutex);
m_activeMwmsChanged = false;
UniteActiveMwms(activeMwms);
std::lock_guard<std::mutex> lock(m_trafficSourceMutex);
for (auto & source : m_trafficSources)
source->SubscribeOrChangeSubscription(activeMwms);
}
if (!ChangeSubscription(activeMwms))
return false;
}
return true;
}
// TODO make this work with multiple sources (e.g. Android)
void TrafficManager::Unsubscribe()
{
if (!IsSubscribed())
return;
// TODO
LOG(LINFO, ("Would unsubscribe from", m_subscriptionId));
m_subscriptionId.clear();
}
bool TrafficManager::IsSubscribed()
{
return !m_subscriptionId.empty();
std::lock_guard<std::mutex> lock(m_trafficSourceMutex);
for (auto & source : m_trafficSources)
source->Unsubscribe();
}
bool TrafficManager::RestoreCache()
@@ -563,51 +516,42 @@ bool TrafficManager::RestoreCache()
return false;
}
// TODO make this work with multiple sources (e.g. Android)
// TODO deal with subscriptions rejected by the server (delete, resubscribe)
bool TrafficManager::Poll()
void TrafficManager::Poll()
{
// TODO
//std::string fileName("test_data/traff/PL-A18-Krzyzowa-Lipiany.xml");
std::string fileName("test_data/traff/PL-A18-Krzyzowa-Lipiany-bidir.xml");
//std::string fileName("test_data/traff/LT-A1-Vezaiciai-Endriejavas.xml");
traffxml::LocalStorage storage(fileName);
pugi::xml_document document;
auto const load_result = storage.Load(document);
if (!load_result)
return false;
std::setlocale(LC_ALL, "en_US.UTF-8");
traffxml::TraffFeed feed;
if (traffxml::ParseTraff(document, std::nullopt /* dataSource */, feed))
{
{
std::lock_guard<std::mutex> lock(m_mutex);
m_feedQueue.push_back(feed);
}
m_lastResponseTime = steady_clock::now();
m_isPollNeeded = false;
return true;
}
else
{
LOG(LWARNING, ("An error occurred parsing the TraFF feed"));
// TODO should we really reset m_isPollNeeded here?
m_isPollNeeded = false;
return false;
}
std::lock_guard<std::mutex> lock(m_trafficSourceMutex);
for (auto & source : m_trafficSources)
if (source->IsPollNeeded())
source->Poll();
}
void TrafficManager::Push(traffxml::TraffFeed feed)
void TrafficManager::ReceiveFeed(traffxml::TraffFeed feed)
{
{
std::lock_guard<std::mutex> lock(m_mutex);
m_feedQueue.push_back(feed);
// TODO should we update m_lastResponseTime?
}
m_condition.notify_one();
}
void TrafficManager::RegisterSource(std::unique_ptr<traffxml::TraffSource> source)
{
std::set<MwmSet::MwmId> activeMwms;
{
std::lock_guard<std::mutex> lock(m_mutex);
UniteActiveMwms(activeMwms);
}
source->SubscribeOrChangeSubscription(activeMwms);
{
std::lock_guard<std::mutex> lock(m_trafficSourceMutex);
m_trafficSources.push_back(std::move(source));
}
m_isPollNeeded = true;
}
void TrafficManager::PurgeExpiredMessages()
{
PurgeExpiredMessagesImpl();
@@ -759,25 +703,17 @@ void TrafficManager::ThreadRoutine()
LOG(LINFO, ("active MWMs changed:", m_activeMwmsChanged, ", poll needed:", m_isPollNeeded));
// this is a no-op if active MWMs have not changed
if (!SetSubscriptionArea())
{
LOG(LWARNING, ("SetSubscriptionArea failed."));
if (!IsSubscribed())
// do not skip out of the loop, we may need to process pushed feeds
LOG(LWARNING, ("No subscription, no traffic data will be retrieved."));
}
SubscribeOrChangeSubscription();
/*
* Fetch traffic data if needed and we have a subscription.
* m_isPollNeeded may be set by WaitForRequest() and set/unset by SetSubscriptionArea().
*/
if (m_isPollNeeded && IsSubscribed())
* Poll sources if needed.
* m_isPollNeeded may be set by WaitForRequest() and set/unset by SubscribeOrChangeSubscription().
*/
if (m_isPollNeeded)
{
if (!Poll())
{
LOG(LWARNING, ("Poll failed."));
// TODO set failed status somewhere and retry
}
m_lastResponseTime = steady_clock::now();
m_isPollNeeded = false;
Poll();
}
}
LOG(LINFO, (m_feedQueue.size(), "feed(s) in queue"));
@@ -1152,6 +1088,12 @@ void TrafficManager::OnTrafficDataResponse(traffic::TrafficInfo && info)
}
#endif
void TrafficManager::GetActiveMwms(std::set<MwmSet::MwmId> & activeMwms)
{
std::lock_guard<std::mutex> lock(m_mutex);
UniteActiveMwms(activeMwms);
}
void TrafficManager::UniteActiveMwms(std::set<MwmSet::MwmId> & activeMwms) const
{
activeMwms.insert(m_activeDrapeMwms.cbegin(), m_activeDrapeMwms.cend());

121
map/traffic_manager.hpp
View File

@@ -15,6 +15,7 @@
#include "traffxml/traff_decoder.hpp"
#include "traffxml/traff_model.hpp"
#include "traffxml/traff_source.hpp"
#include "traffxml/traff_storage.hpp"
#include "geometry/point2d.hpp"
@@ -36,7 +37,7 @@
#include <utility>
#include <vector>
class TrafficManager final
class TrafficManager final : public traffxml::TraffSourceManager
{
public:
using CountryInfoGetterFn = std::function<storage::CountryInfoGetter const &()>;
@@ -234,15 +235,36 @@ public:
void SetTestMode();
/**
* @brief Processes a traffic feed received through a push operation.
* @brief Processes a traffic feed.
*
* Push is safe to call from any thread.
* The feed may be a result of a pull operation, or received through a push operation.
* (Push operations are not supported by all sources.)
*
* Push operations are not supported on all platforms.
* This method is safe to call from any thread.
*
* @param feed The traffic feed.
*/
void Push(traffxml::TraffFeed feed);
virtual void ReceiveFeed(traffxml::TraffFeed feed) override;
/**
* @brief Registers a `TraffSource`.
* @param source The source.
*/
virtual void RegisterSource(std::unique_ptr<traffxml::TraffSource> source) override;
/**
* @brief Retrieves all currently active MWMs.
*
* This method retrieves all MWMs in the viewport, within a certain distance of the current
* position (if there is a valid position) or part of the route (if any), and stores them in
* `activeMwms`.
*
* This method locks `m_mutex` and is therefore safe to call from any thread. Callers which
* already hold `m_mutex` can use the private `UniteActiveMwms()` method instead.
*
* @param activeMwms Retrieves the list of active MWMs.
*/
virtual void GetActiveMwms(std::set<MwmSet::MwmId> & activeMwms) override;
/**
* @brief Purges expired messages from the cache.
@@ -332,50 +354,21 @@ private:
};
/**
* @brief Returns a TraFF filter list for a set of MWMs.
* @brief Ensures every TraFF source has a subscription covering all currently active MWMs.
*
* @param mwms The MWMs for which a filter list is to be created.
* @return A `filter_list` in XML format.
* This method cycles through all TraFF sources in `m_trafficSources` and calls
* `SubscribeOrChangeSubscription()` on each of them.
*/
std::string GetMwmFilters(std::set<MwmSet::MwmId> & mwms);
void SubscribeOrChangeSubscription();
/**
* @brief Subscribes to a traffic service.
* @brief Unsubscribes from all traffic services we are subscribed to.
*
* @param mwms The MWMs for which data is needed.
* @return true on success, false on failure.
*/
bool Subscribe(std::set<MwmSet::MwmId> & mwms);
/**
* @brief Changes an existing traffic subscription.
*
* @param mwms The new set of MWMs for which data is needed.
* @return true on success, false on failure.
*/
bool ChangeSubscription(std::set<MwmSet::MwmId> & mwms);
/**
* @brief Ensures we have a subscription covering all currently active MWMs.
*
* This method subscribes to a traffic service if not already subscribed, or changes the existing
* subscription otherwise.
*
* @return true on success, false on failure.
*/
bool SetSubscriptionArea();
/**
* @brief Unsubscribes from a traffic service we are subscribed to.
* This method cycles through all TraFF sources in `m_trafficSources` and calls `Unsubscribe()`
* on each of them.
*/
void Unsubscribe();
/**
* @brief Whether we are currently subscribed to a traffic service.
* @return
*/
bool IsSubscribed();
/**
* @brief Restores the message cache from file storage.
*
@@ -393,11 +386,12 @@ private:
bool RestoreCache();
/**
* @brief Polls the traffic service for updates.
* @brief Polls all traffic services for updates.
*
* @return true on success, false on failure.
* This method cycles through all TraFF sources in `m_trafficSources` and calls `IsPollNeeded()`
* on each of them. If this method returns true, it then calls `Poll()` on the source.
*/
bool Poll();
void Poll();
/**
* @brief Purges expired messages from the cache.
@@ -568,6 +562,18 @@ private:
void OnChangeRoutingSessionState(routing::SessionState previous, routing::SessionState current);
/**
* @brief Retrieves all currently active MWMs.
*
* This method retrieves all MWMs in the viewport, within a certain distance of the current
* position (if there is a valid position) or part of the route (if any), and stores them in
* `activeMwms`.
*
* The caller must hold `m_mutex` prior to calling this method. `GetActiveMwms()` is available
* as a convenience wrapper which locks `m_mutex`, calls this method and releases it.
*
* @param activeMwms Retrieves the list of active MWMs.
*/
void UniteActiveMwms(std::set<MwmSet::MwmId> & activeMwms) const;
void Pause();
@@ -634,6 +640,13 @@ private:
std::map<MwmSet::MwmId, CacheEntry> m_mwmCache;
/**
* @brief The TraFF sources from which we get traffic information.
*
* Threads must lock `m_trafficSourceMutex` prior to accessing this member.
*/
std::vector<std::unique_ptr<traffxml::TraffSource>> m_trafficSources;
bool m_isRunning;
std::condition_variable m_condition;
@@ -691,9 +704,18 @@ private:
* @brief Mutex for access to shared members.
*
* Threads which access shared members (see documentation) must lock this mutex while doing so.
*
* @note To access `m_trafficSource`, lock `m_trafficSourceMutex`, not this mutex.
*/
std::mutex m_mutex;
/**
* @brief Mutex for access to `m_trafficSources`.
*
* Threads which access `m_trafficSources` must lock this mutex while doing so.
*/
std::mutex m_trafficSourceMutex;
/**
* @brief Worker thread which fetches traffic updates.
*/
@@ -724,18 +746,11 @@ private:
*/
bool m_activeMwmsChanged = false;
/**
* @brief The subscription ID received from the traffic server.
*
* An empty subscription ID means no subscription.
*/
std::string m_subscriptionId;
/**
* @brief Whether a poll operation is needed.
*
* Used in the worker thread. A poll operation is needed unless a subscription (or subscription
* change) operation was performed before and a feed was received as part of it.
* Used in the worker thread to indicate we need to poll all sources. The poll operation may still
* be inhibited for individual sources.
*/
bool m_isPollNeeded;