Git-Mirrors/RetroShare
1
0
Fork 0
mirror of https://github.com/RetroShare/RetroShare.git synced 2024-10-01 02:35:48 -04:00
Code Issues Actions Packages Projects Releases Wiki Activity

use notify for gxs events. Removed the previous polling based system. Now multiple clients can receive gxs changes. This also fixes the always growing changes queue in rs-nogui.

Browse Source 
git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@8057 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
electron128 2015-03-20 10:39:17 +00:00
parent b912ac656c
commit 6dff335515
11 changed files with 117 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
libretroshare/src/gxs/rsgenexchange.cc
View File

@ -35,6 +35,7 @@
#include "retroshare/rsgrouter.h"
#include "rsgixs.h"
#include "rsgxsutil.h"
#include "rsserver/p3face.h"
#include <algorithm>
@ -992,105 +993,67 @@ bool RsGenExchange::checkAuthenFlag(const PrivacyBitPos& pos, const uint8_t& fla
}
}
void RsGenExchange::receiveChanges(std::vector<RsGxsNotify*>& changes)
{
RS_STACK_MUTEX(mGenMtx) ;
#ifdef GEN_EXCH_DEBUG
std::cerr << "RsGenExchange::receiveChanges()" << std::endl;
#endif
std::vector<RsGxsNotify*>::iterator vit = changes.begin();
for(; vit != changes.end(); ++vit)
{
RsGxsNotify* n = *vit;
RsGxsGroupChange* gc;
RsGxsMsgChange* mc;
if((mc = dynamic_cast<RsGxsMsgChange*>(n)) != NULL)
{
mMsgChange.push_back(mc);
}
else if((gc = dynamic_cast<RsGxsGroupChange*>(n)) != NULL)
{
mGroupChange.push_back(gc);
}
else
{
#warning cyril: very weird code. Why delete an element without removing it from the array
delete n;
}
}
}
static void addMessageChanged(std::map<RsGxsGroupId, std::vector<RsGxsMessageId> > &msgs, const std::map<RsGxsGroupId, std::vector<RsGxsMessageId> > &msgChanged)
{
if (msgs.empty()) {
msgs = msgChanged;
} else {
std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >::const_iterator mapIt;
for (mapIt = msgChanged.begin(); mapIt != msgChanged.end(); ++mapIt) {
const RsGxsGroupId &grpId = mapIt->first;
const std::vector<RsGxsMessageId> &srcMsgIds = mapIt->second;
std::vector<RsGxsMessageId> &destMsgIds = msgs[grpId];
if (msgs.empty()) {
msgs = msgChanged;
} else {
std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >::const_iterator mapIt;
for (mapIt = msgChanged.begin(); mapIt != msgChanged.end(); ++mapIt) {
const RsGxsGroupId &grpId = mapIt->first;
const std::vector<RsGxsMessageId> &srcMsgIds = mapIt->second;
std::vector<RsGxsMessageId> &destMsgIds = msgs[grpId];
std::vector<RsGxsMessageId>::const_iterator msgIt;
for (msgIt = srcMsgIds.begin(); msgIt != srcMsgIds.end(); ++msgIt) {
if (std::find(destMsgIds.begin(), destMsgIds.end(), *msgIt) == destMsgIds.end()) {
destMsgIds.push_back(*msgIt);
}
}
}
}
std::vector<RsGxsMessageId>::const_iterator msgIt;
for (msgIt = srcMsgIds.begin(); msgIt != srcMsgIds.end(); ++msgIt) {
if (std::find(destMsgIds.begin(), destMsgIds.end(), *msgIt) == destMsgIds.end()) {
destMsgIds.push_back(*msgIt);
}
}
}
}
}
void RsGenExchange::msgsChanged(std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >& msgs, std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >& msgsMeta)
void RsGenExchange::receiveChanges(std::vector<RsGxsNotify*>& changes)
{
if(mGenMtx.trylock())
{
while(!mMsgChange.empty())
{
RsGxsMsgChange* mc = mMsgChange.back();
if (mc->metaChange())
{
addMessageChanged(msgsMeta, mc->msgChangeMap);
}
else
{
addMessageChanged(msgs, mc->msgChangeMap);
}
mMsgChange.pop_back();
delete mc;
}
mGenMtx.unlock();
}
}
#ifdef GEN_EXCH_DEBUG
std::cerr << "RsGenExchange::receiveChanges()" << std::endl;
#endif
RsGxsChanges out;
out.mService = getTokenService();
void RsGenExchange::groupsChanged(std::list<RsGxsGroupId>& grpIds, std::list<RsGxsGroupId>& grpIdsMeta)
{
if(mGenMtx.trylock())
{
while(!mGroupChange.empty())
{
RsGxsGroupChange* gc = mGroupChange.back();
std::list<RsGxsGroupId>& gList = gc->mGrpIdList;
std::list<RsGxsGroupId>::iterator lit = gList.begin();
for(; lit != gList.end(); ++lit)
if (gc->metaChange())
{
grpIdsMeta.push_back(*lit);
}
else
{
grpIds.push_back(*lit);
}
mGroupChange.pop_back();
delete gc;
}
mGenMtx.unlock();
}
// collect all changes in one GxsChanges object
std::vector<RsGxsNotify*>::iterator vit = changes.begin();
for(; vit != changes.end(); ++vit)
{
RsGxsNotify* n = *vit;
RsGxsGroupChange* gc;
RsGxsMsgChange* mc;
if((mc = dynamic_cast<RsGxsMsgChange*>(n)) != NULL)
{
if (mc->metaChange())
{
addMessageChanged(out.mMsgsMeta, mc->msgChangeMap);
}
else
{
addMessageChanged(out.mMsgs, mc->msgChangeMap);
}
}
else if((gc = dynamic_cast<RsGxsGroupChange*>(n)) != NULL)
{
if(gc->metaChange())
{
out.mGrpsMeta.splice(out.mGrpsMeta.end(), gc->mGrpIdList);
}
else
{
out.mGrps.splice(out.mGrps.end(), gc->mGrpIdList);
}
}
delete n;
}
RsServer::notify()->notifyGxsChange(out);
}
bool RsGenExchange::subscribeToGroup(uint32_t& token, const RsGxsGroupId& grpId, bool subscribe)
@ -1118,40 +1081,6 @@ bool RsGenExchange::getServiceStatistic(const uint32_t& token, GxsServiceStatist
return mDataAccess->getServiceStatistic(token, stats);
}
bool RsGenExchange::updated(bool willCallGrpChanged, bool willCallMsgChanged)
{
bool changed = false;
if(mGenMtx.trylock())
{
changed = (!mGroupChange.empty() || !mMsgChange.empty());
if(!willCallGrpChanged)
{
while(!mGroupChange.empty())
{
RsGxsGroupChange* gc = mGroupChange.back();
mGroupChange.pop_back();
delete gc;
}
}
if(!willCallMsgChanged)
{
while(!mMsgChange.empty())
{
RsGxsMsgChange* mc = mMsgChange.back();
mMsgChange.pop_back();
delete mc;
}
}
mGenMtx.unlock();
}
return changed;
}
bool RsGenExchange::getGroupList(const uint32_t &token, std::list<RsGxsGroupId> &groupIds)
{
return mDataAccess->getGroupList(token, groupIds);

41
libretroshare/src/gxs/rsgenexchange.h
View File

@ -255,44 +255,6 @@ public:
*/
virtual void receiveChanges(std::vector<RsGxsNotify*>& changes);
/*!
* Checks to see if a change has been received for
* for a message or group
* @param willCallGrpChanged if this is set to true, group changed function will return list
* groups that have changed, if false, the group changed list is cleared
* @param willCallMsgChanged if this is set to true, msgChanged function will return map
* messages that have changed, if false, the message changed map is cleared
* @return true if a change has occured for msg or group
* @see groupsChanged
* @see msgsChanged
*/
bool updated(bool willCallGrpChanged = false, bool willCallMsgChanged = false);
/*!
* The groups changed. \n
* class can reimplement to use to tailor
* the group actually set for ui notification.
* If receivedChanges is not passed RsGxsNotify changes
* this function does nothing
* @param grpIds returns list of grpIds that have changed
* @param grpIdsMeta returns list of grpIds with meta data changes
* @see updated
*/
void groupsChanged(std::list<RsGxsGroupId>& grpIds, std::list<RsGxsGroupId>& grpIdsMeta);
/*!
* The msg changed. \n
* class can reimplement to use to tailor
* the msg actually set for ui notification.
* If receivedChanges is not passed RsGxsNotify changes
* this function does nothing
* @param msgs returns map of message ids that have changed
* @param msgsMeta returns map of message ids with meta data changes
* @see updated
*/
void msgsChanged(std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >& msgs, std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >& msgsMeta);
bool subscribeToGroup(uint32_t& token, const RsGxsGroupId& grpId, bool subscribe);
/*!
@ -878,9 +840,6 @@ private:
private:
std::vector<RsGxsGroupChange*> mGroupChange;
std::vector<RsGxsMsgChange*> mMsgChange;
const uint8_t CREATE_FAIL, CREATE_SUCCESS, CREATE_FAIL_TRY_LATER, SIGN_MAX_ATTEMPTS;
const uint8_t SIGN_FAIL, SIGN_SUCCESS, SIGN_FAIL_TRY_LATER;
const uint8_t VALIDATE_FAIL, VALIDATE_SUCCESS, VALIDATE_FAIL_TRY_LATER, VALIDATE_MAX_ATTEMPTS;

1
libretroshare/src/pqi/p3notify.cc
View File

@ -235,6 +235,7 @@ void p3Notify::notifyOwnAvatarChanged ()
void p3Notify::notifyOwnStatusMessageChanged() { FOR_ALL_NOTIFY_CLIENTS (*it)->notifyOwnStatusMessageChanged() ; }
void p3Notify::notifyDiskFull (uint32_t location , uint32_t size_limit_in_MB ) { FOR_ALL_NOTIFY_CLIENTS (*it)->notifyDiskFull (location,size_limit_in_MB) ; }
void p3Notify::notifyPeerStatusChanged (const std::string& peer_id , uint32_t status ) { FOR_ALL_NOTIFY_CLIENTS (*it)->notifyPeerStatusChanged (peer_id,status) ; }
void p3Notify::notifyGxsChange (const RsGxsChanges& changes) {FOR_ALL_NOTIFY_CLIENTS (*it)->notifyGxsChange(changes) ;}
void p3Notify::notifyPeerStatusChangedSummary () { FOR_ALL_NOTIFY_CLIENTS (*it)->notifyPeerStatusChangedSummary() ; }
void p3Notify::notifyDiscInfoChanged () { FOR_ALL_NOTIFY_CLIENTS (*it)->notifyDiscInfoChanged () ; }

1
libretroshare/src/pqi/p3notify.h
View File

@ -110,6 +110,7 @@ class p3Notify: public RsNotify
void notifyOwnStatusMessageChanged () ;
void notifyDiskFull (uint32_t /* location */, uint32_t /* size limit in MB */) ;
void notifyPeerStatusChanged (const std::string& /* peer_id */, uint32_t /* status */) ;
void notifyGxsChange (const RsGxsChanges& /* changes */);
void notifyPeerStatusChangedSummary () ;
void notifyDiscInfoChanged () ;

51
libretroshare/src/retroshare/rsgxsiface.h
View File

@ -31,6 +31,20 @@
#include "gxs/rsgxsdata.h"
#include "retroshare/rsgxsifacetypes.h"
/*!
* Stores ids of changed gxs groups and messages. It is used to notify the GUI about changes.
*/
class RsGxsChanges
{
public:
RsGxsChanges(): mService(0){}
RsTokenService *mService;
std::map<RsGxsGroupId, std::vector<RsGxsMessageId> > mMsgs;
std::map<RsGxsGroupId, std::vector<RsGxsMessageId> > mMsgsMeta;
std::list<RsGxsGroupId> mGrps;
std::list<RsGxsGroupId> mGrpsMeta;
};
/*!
* All implementations must offer thread safety
*/
@ -49,43 +63,6 @@ public:
*/
virtual void receiveChanges(std::vector<RsGxsNotify*>& changes) = 0;
/*!
* Checks to see if a change has been received for
* for a message or group
* @param willCallGrpChanged if this is set to true, group changed function will return list
* groups that have changed, if false, the group changed list is cleared
* @param willCallMsgChanged if this is set to true, msgChanged function will return map
* messages that have changed, if false, the message changed map is cleared
* @return true if a change has occured for msg or group
* @see groupsChanged
* @see msgsChanged
*/
virtual bool updated(bool willCallGrpChanged = false, bool willCallMsgChanged = false) = 0;
/*!
* The groups changed. \n
* class can reimplement to use to tailor
* the group actually set for ui notification.
* If receivedChanges is not passed RsGxsNotify changes
* this function does nothing
* @param grpIds returns list of grpIds that have changed
* @param grpIdsMeta returns list of grpIds with meta data changes
* @see updated
*/
virtual void groupsChanged(std::list<RsGxsGroupId>& grpIds, std::list<RsGxsGroupId>& grpIdsMeta) = 0;
/*!
* The msg changed. \n
* class can reimplement to use to tailor
* the msg actually set for ui notification.
* If receivedChanges is not passed RsGxsNotify changes
* this function does nothing
* @param msgs returns map of message ids that have changed
* @param msgsMeta returns map of message ids with meta data changes
* @see updated
*/
virtual void msgsChanged(std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >& msgs, std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >& msgsMeta) = 0;
/*!
* @return handle to token service for this GXS service
*/

45
libretroshare/src/retroshare/rsgxsifacehelper.h
View File

@ -61,51 +61,6 @@ public:
mGxs->receiveChanges(changes);
}
/*!
* Checks to see if a change has been received for
* for a message or group
* @param willCallGrpChanged if this is set to true, group changed function will return list
* groups that have changed, if false, the group changed list is cleared
* @param willCallMsgChanged if this is set to true, msgChanged function will return map
* messages that have changed, if false, the message changed map is cleared
* @return true if a change has occured for msg or group
* @see groupsChanged
* @see msgsChanged
*/
bool updated(bool willCallGrpChanged = false, bool willCallMsgChanged = false)
{
return mGxs->updated(willCallGrpChanged, willCallMsgChanged);
}
/*!
* The groups changed. \n
* class can reimplement to use to tailor
* the group actually set for ui notification.
* If receivedChanges is not passed RsGxsNotify changes
* this function does nothing
* @param grpIds returns list of grpIds that have changed
* @param grpIdsMeta returns list of grpIds with meta data changes
* @see updated
*/
void groupsChanged(std::list<RsGxsGroupId> &grpIds, std::list<RsGxsGroupId>& grpIdsMeta)
{
mGxs->groupsChanged(grpIds, grpIdsMeta);
}
/*!
* The msg changed. \n
* class can be reimplemented to use to tailor
* the msg actually set for ui notification.
* If receivedChanges is not passed RsGxsNotify changes
* this function does nothing
* @param msgs returns map of message ids that have changed
* @param msgsMeta returns map of message ids with meta data changes
* @see updated
*/
void msgsChanged(std::map<RsGxsGroupId, std::vector<RsGxsMessageId> > &msgs, std::map<RsGxsGroupId, std::vector<RsGxsMessageId> >& msgsMeta)
{
mGxs->msgsChanged(msgs, msgsMeta);
}
/*!
* @return handle to token service for this GXS service
*/

3
libretroshare/src/retroshare/rsnotify.h
View File

@ -34,9 +34,11 @@
#include <stdint.h>
#include "rsturtle.h"
#include "rsgxsifacetypes.h"
class ChatId;
class ChatMessage;
class RsGxsChanges;
class RsNotify;
extern RsNotify *rsNotify;
@ -215,6 +217,7 @@ class NotifyClient
virtual void notifyOwnStatusMessageChanged () {}
virtual void notifyDiskFull (uint32_t /* location */, uint32_t /* size limit in MB */) {}
virtual void notifyPeerStatusChanged (const std::string& /* peer_id */, uint32_t /* status */) {}
virtual void notifyGxsChange (const RsGxsChanges& /* changes */) {}
/* one or more peers has changed the states */
virtual void notifyPeerStatusChangedSummary () {}

18
retroshare-gui/src/gui/notifyqt.cpp
View File

@ -18,6 +18,7 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
*************************************************************************/
#include <retroshare/rsgxsifacehelper.h>
#include <QInputDialog>
#include <QMessageBox>
@ -140,6 +141,7 @@ NotifyQt::NotifyQt() : cDialog(NULL)
// register to allow sending over Qt::QueuedConnection
qRegisterMetaType<ChatId>("ChatId");
qRegisterMetaType<ChatMessage>("ChatMessage");
qRegisterMetaType<RsGxsChanges>("RsGxsChanges");
}
void NotifyQt::notifyErrorMsg(int list, int type, std::string msg)
@ -417,6 +419,22 @@ void NotifyQt::notifyPeerStatusChangedSummary()
emit peerStatusChangedSummary();
}
void NotifyQt::notifyGxsChange(const RsGxsChanges& changes)
{
{
QMutexLocker m(&_mutex) ;
if(!_enabled)
return ;
}
#ifdef NOTIFY_DEBUG
std::cerr << "Notifyqt:: notified that gxs has changes" << std::endl;
#endif
emit gxsChange(changes);
}
#ifdef REMOVE
void NotifyQt::notifyForumMsgReadSatusChanged(const std::string& forumId, const std::string& msgId, uint32_t status)
{

4
retroshare-gui/src/gui/notifyqt.h
View File

@ -57,6 +57,9 @@ class NotifyQt: public QObject, public NotifyClient
virtual void notifyPeerStatusChanged(const std::string& peer_id, uint32_t state);
/* one or more peers has changed the states */
virtual void notifyPeerStatusChangedSummary();
virtual void notifyGxsChange(const RsGxsChanges& change);
#ifdef REMOVE
virtual void notifyForumMsgReadSatusChanged(const std::string& forumId, const std::string& msgId, uint32_t status);
virtual void notifyChannelMsgReadSatusChanged(const std::string& channelId, const std::string& msgId, uint32_t status);
@ -121,6 +124,7 @@ class NotifyQt: public QObject, public NotifyClient
void diskFull(int,int) const ;
void peerStatusChanged(const QString& /* peer_id */, int /* status */);
void peerStatusChangedSummary() const;
void gxsChange(const RsGxsChanges& /* changes */);
#ifdef REMOVE
void publicChatChanged(int type) const ;
void privateChatChanged(int list, int type) const ;

47
retroshare-gui/src/util/RsGxsUpdateBroadcast.cpp
View File

@ -1,20 +1,21 @@
//#include <QTimer>
#include <QMap>
#include "RsGxsUpdateBroadcast.h"
#include "RsProtectedTimer.h"
#include "gui/notifyqt.h"
#include <retroshare/rsgxsifacehelper.h>
// previously gxs allowed only one event consumer to poll for changes
// this required a single broadcast instance per service
// now the update notify works through rsnotify and notifyqt
// so the single instance per service is not really needed anymore
QMap<RsGxsIfaceHelper*, RsGxsUpdateBroadcast*> updateBroadcastMap;
RsGxsUpdateBroadcast::RsGxsUpdateBroadcast(RsGxsIfaceHelper *ifaceImpl) :
QObject(NULL), mIfaceImpl(ifaceImpl)
{
mTimer = new RsProtectedTimer(this);
mTimer->setInterval(1000);
mTimer->setSingleShot(true);
connect(mTimer, SIGNAL(timeout()), this, SLOT(poll()));
connect(NotifyQt::getInstance(), SIGNAL(gxsChange(RsGxsChanges)), this, SLOT(onChangesReceived(RsGxsChanges)));
}
void RsGxsUpdateBroadcast::cleanup()
@ -36,34 +37,24 @@ RsGxsUpdateBroadcast *RsGxsUpdateBroadcast::get(RsGxsIfaceHelper *ifaceImpl)
RsGxsUpdateBroadcast *updateBroadcast = new RsGxsUpdateBroadcast(ifaceImpl);
updateBroadcastMap.insert(ifaceImpl, updateBroadcast);
updateBroadcast->poll();
return updateBroadcast;
}
void RsGxsUpdateBroadcast::poll()
void RsGxsUpdateBroadcast::onChangesReceived(const RsGxsChanges& changes)
{
std::map<RsGxsGroupId, std::vector<RsGxsMessageId> > msgs;
std::map<RsGxsGroupId, std::vector<RsGxsMessageId> > msgsMeta;
std::list<RsGxsGroupId> grps;
std::list<RsGxsGroupId> grpsMeta;
if(changes.mService != mIfaceImpl->getTokenService())
return;
if (mIfaceImpl->updated(true, true))
{
mIfaceImpl->msgsChanged(msgs, msgsMeta);
if (!msgs.empty() || !msgsMeta.empty())
{
emit msgsChanged(msgs, msgsMeta);
}
if (!changes.mMsgs.empty() || !changes.mMsgsMeta.empty())
{
emit msgsChanged(changes.mMsgs, changes.mMsgsMeta);
}
mIfaceImpl->groupsChanged(grps, grpsMeta);
if (!grps.empty() || !grpsMeta.empty())
{
emit grpsChanged(grps, grpsMeta);
}
if (!changes.mGrps.empty() || !changes.mGrpsMeta.empty())
{
emit grpsChanged(changes.mGrps, changes.mGrpsMeta);
}
emit changed();
}
mTimer->start();
emit changed();
}

5
retroshare-gui/src/util/RsGxsUpdateBroadcast.h
View File

@ -6,7 +6,7 @@
#include <retroshare/rsgxsifacetypes.h>
class RsGxsIfaceHelper;
class QTimer;
class RsGxsChanges;
class RsGxsUpdateBroadcast : public QObject
{
@ -23,14 +23,13 @@ signals:
void grpsChanged(const std::list<RsGxsGroupId>& grpIds, const std::list<RsGxsGroupId>& grpIdsMeta);
private slots:
void poll();
void onChangesReceived(const RsGxsChanges& changes);
private:
explicit RsGxsUpdateBroadcast(RsGxsIfaceHelper* ifaceImpl);
private:
RsGxsIfaceHelper* mIfaceImpl;
QTimer *mTimer;
};
#endif // RSGXSUPDATEBROADCAST_H