mirror of
https://github.com/RetroShare/RetroShare.git
synced 2025-05-08 17:15:16 -04:00
first version of distant messages. Still needs some debugging
git-svn-id: http://svn.code.sf.net/p/retroshare/code/branches/v0.5-GenericTunneling@6344 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
csoler
parent
6a88d7382a
commit
ca279ec0c2
13 changed files with 442 additions and 24 deletions
|
|
@ -29,6 +29,7 @@
|
|||
#include "pqi/pqibin.h"
|
||||
#include "pqi/pqiarchive.h"
|
||||
#include "pqi/p3linkmgr.h"
|
||||
#include "pqi/authgpg.h"
|
||||
|
||||
#include "services/p3msgservice.h"
|
||||
#include "pqi/pqinotify.h"
|
||||
|
|
@ -36,12 +37,17 @@
|
|||
#include "util/rsdebug.h"
|
||||
#include "util/rsdir.h"
|
||||
#include "util/rsstring.h"
|
||||
#include "util/rsrandom.h"
|
||||
|
||||
#include <iomanip>
|
||||
#include <map>
|
||||
|
||||
const int msgservicezone = 54319;
|
||||
|
||||
static const uint32_t RS_DISTANT_MSG_STATUS_TUNNEL_OK = 0x0001 ;
|
||||
static const uint32_t RS_DISTANT_MSG_STATUS_TUNNEL_DN = 0x0000 ;
|
||||
static const uint32_t DISTANT_MSG_HASH_SIZE = 20 ;
|
||||
|
||||
/* Another little hack ..... unique message Ids
|
||||
* will be handled in this class.....
|
||||
* These are unique within this run of the server,
|
||||
|
|
@ -81,6 +87,15 @@ int p3MsgService::tick()
|
|||
*/
|
||||
|
||||
incomingMsgs();
|
||||
|
||||
static time_t last_management_time = 0 ;
|
||||
time_t now = time(NULL) ;
|
||||
|
||||
if(now > last_management_time + 5)
|
||||
{
|
||||
manageDistantPeers() ;
|
||||
last_management_time = now ;
|
||||
}
|
||||
//checkOutgoingMessages();
|
||||
|
||||
return 0;
|
||||
|
|
@ -187,22 +202,27 @@ int p3MsgService::incomingMsgs()
|
|||
{
|
||||
RsMsgItem *mi;
|
||||
int i = 0;
|
||||
bool changed = false ;
|
||||
|
||||
while((mi = (RsMsgItem *) recvItem()) != NULL)
|
||||
{
|
||||
changed = true ;
|
||||
++i;
|
||||
handleIncomingItem(mi) ;
|
||||
++i ;
|
||||
}
|
||||
|
||||
if(checkAndRebuildPartialMessage(mi)) // only returns true when a msg is complete.
|
||||
{
|
||||
processMsg(mi, true);
|
||||
}
|
||||
return i;
|
||||
}
|
||||
|
||||
void p3MsgService::handleIncomingItem(RsMsgItem *mi)
|
||||
{
|
||||
bool changed = false ;
|
||||
|
||||
if(checkAndRebuildPartialMessage(mi)) // only returns true when a msg is complete.
|
||||
{
|
||||
processMsg(mi, true);
|
||||
changed = true ;
|
||||
}
|
||||
if(changed)
|
||||
rsicontrol->getNotify().notifyListChange(NOTIFY_LIST_MESSAGELIST,NOTIFY_TYPE_MOD);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
void p3MsgService::statusChange(const std::list<pqipeer> &/*plist*/)
|
||||
|
|
@ -1562,3 +1582,268 @@ RsMsgItem *p3MsgService::initMIRsMsg(MessageInfo &info, const std::string &to)
|
|||
//msg->print(std::cerr);
|
||||
return msg;
|
||||
}
|
||||
|
||||
void p3MsgService::connectToTurtleRouter(p3turtle *pt)
|
||||
{
|
||||
mTurtle = pt ;
|
||||
pt->registerTunnelService(this) ;
|
||||
}
|
||||
bool p3MsgService::createDistantOfflineMessengingInvite(time_t time_of_validity,TurtleFileHash& hash)
|
||||
{
|
||||
unsigned char hash_bytes[DISTANT_MSG_HASH_SIZE] ;
|
||||
RSRandom::random_bytes( hash_bytes, DISTANT_MSG_HASH_SIZE) ;
|
||||
|
||||
hash = t_RsGenericIdType<DISTANT_MSG_HASH_SIZE>(hash_bytes).toStdString(false) ;
|
||||
|
||||
DistantMessengingInvite invite ;
|
||||
invite.time_of_validity = time_of_validity ;
|
||||
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
_messenging_invites[hash] = invite ;
|
||||
}
|
||||
return true ;
|
||||
}
|
||||
bool p3MsgService::getDistantOfflineMessengingInvites(std::vector<DistantOfflineMessengingInvite>& invites)
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
|
||||
for(std::map<std::string,DistantMessengingInvite>::const_iterator it(_messenging_invites.begin());it!=_messenging_invites.end();++it)
|
||||
{
|
||||
DistantOfflineMessengingInvite invite ;
|
||||
invite.hash = it->first ;
|
||||
invite.issuer_pgp_id = AuthGPG::getAuthGPG()->getGPGOwnId() ;
|
||||
invite.time_of_validity = it->second.time_of_validity ;
|
||||
}
|
||||
|
||||
return true ;
|
||||
}
|
||||
bool p3MsgService::handleTunnelRequest(const std::string& hash,const std::string& peer_id)
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
std::map<std::string,DistantMessengingInvite>::const_iterator it = _messenging_invites.find(hash) ;
|
||||
|
||||
return it != _messenging_invites.end() ;
|
||||
}
|
||||
|
||||
void p3MsgService::manageDistantPeers()
|
||||
{
|
||||
// now possibly flush pending messages
|
||||
|
||||
std::cerr << "p3MsgService::manageDistantPeers()" << std::endl;
|
||||
|
||||
std::vector<std::pair<std::string,RsMsgItem*> > to_send ;
|
||||
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
for(std::map<std::string,DistantMessengingContact>::iterator it(_messenging_contacts.begin());it!=_messenging_contacts.end();++it)
|
||||
if(it->second.status == RS_DISTANT_MSG_STATUS_TUNNEL_OK)
|
||||
for(uint32_t i=0;i<it->second.pending_messages.size();++i)
|
||||
to_send.push_back(std::pair<std::string,RsMsgItem*>(it->first,it->second.pending_messages[i])) ;
|
||||
}
|
||||
|
||||
for(uint32_t i=0;i<to_send.size();++i)
|
||||
{
|
||||
std::cerr << " Flushing msg " << to_send[i].second->msgId << std::endl;
|
||||
sendTurtleData(to_send[i].first,to_send[i].second) ;
|
||||
}
|
||||
|
||||
time_t now = time(NULL) ;
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
|
||||
// clean dead invites.
|
||||
//
|
||||
for(std::map<std::string,DistantMessengingInvite>::iterator it(_messenging_invites.begin());it!=_messenging_invites.end();)
|
||||
if(it->second.time_of_validity < now)
|
||||
{
|
||||
std::cerr << " Removing outdated invite " << it->second.time_of_validity << ", hash=" << it->first << std::endl;
|
||||
std::map<std::string,DistantMessengingInvite>::iterator tmp(it) ;
|
||||
++tmp ;
|
||||
_messenging_invites.erase(it) ;
|
||||
it = tmp ;
|
||||
}
|
||||
else
|
||||
++it ;
|
||||
|
||||
// clean dead contacts.
|
||||
//
|
||||
for(std::map<std::string,DistantMessengingContact>::iterator it(_messenging_contacts.begin());it!=_messenging_contacts.end();)
|
||||
if(it->second.pending_messages.empty() && it->second.status == RS_DISTANT_MSG_STATUS_TUNNEL_DN)
|
||||
{
|
||||
std::cerr << " Removing dead contact with no pending msgs and dead tunnel. hash=" << it->first << std::endl;
|
||||
std::map<std::string,DistantMessengingContact>::iterator tmp(it) ;
|
||||
++tmp ;
|
||||
_messenging_contacts.erase(it) ;
|
||||
it = tmp ;
|
||||
}
|
||||
else
|
||||
++it ;
|
||||
}
|
||||
}
|
||||
|
||||
void p3MsgService::addVirtualPeer(const TurtleFileHash& hash, const TurtleVirtualPeerId& vpid,RsTurtleGenericTunnelItem::Direction dir)
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
// A new tunnel has been created. We need to flush pending messages for the corresponding peer.
|
||||
|
||||
//std::map<std::string,DistantMessengingContact>::const_iterator it = _messenging_contacts.find(hash) ;
|
||||
|
||||
DistantMessengingContact& contact(_messenging_contacts[hash]) ; // possibly creates it.
|
||||
|
||||
contact.virtual_peer_id = vpid ;
|
||||
contact.last_hit_time = time(NULL) ;
|
||||
contact.status = RS_DISTANT_MSG_STATUS_TUNNEL_OK ;
|
||||
|
||||
std::cerr << "p3MsgService::addVirtualPeer(): adding virtual peer " << vpid << " for hash " << hash << std::endl;
|
||||
}
|
||||
void p3MsgService::removeVirtualPeer(const TurtleFileHash& hash, const TurtleVirtualPeerId& vpid)
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
// A new tunnel has been created. We need to flush pending messages for the corresponding peer.
|
||||
|
||||
//std::map<std::string,DistantMessengingContact>::const_iterator it = _messenging_contacts.find(hash) ;
|
||||
|
||||
DistantMessengingContact& contact(_messenging_contacts[hash]) ; // possibly creates it.
|
||||
|
||||
contact.status = RS_DISTANT_MSG_STATUS_TUNNEL_DN ;
|
||||
contact.virtual_peer_id.clear() ;
|
||||
}
|
||||
static void printBinaryData(void *data,uint32_t size)
|
||||
{
|
||||
static const char outl[16] = { '0','1','2','3','4','5','6','7','8','9','a','b','c','d','e','f' } ;
|
||||
|
||||
for(uint32_t j = 0; j < size; j++)
|
||||
{
|
||||
std::cerr << outl[ ( ((uint8_t*)data)[j]>>4) ] ;
|
||||
std::cerr << outl[ ((uint8_t*)data)[j] & 0xf ] ;
|
||||
}
|
||||
}
|
||||
|
||||
void p3MsgService::sendTurtleData(const std::string& hash,RsMsgItem *msgitem)
|
||||
{
|
||||
// The item is serialized and turned into a generic turtle item.
|
||||
|
||||
uint32_t rssize = _serialiser->size(msgitem) ;
|
||||
unsigned char *data = new unsigned char[rssize] ;
|
||||
|
||||
if(!_serialiser->serialise(msgitem,data,&rssize))
|
||||
{
|
||||
std::cerr << "(EE) p3MsgService::sendTurtleData(): Serialization error." << std::endl;
|
||||
delete[] data ;
|
||||
return ;
|
||||
}
|
||||
|
||||
RsTurtleGenericDataItem *item = new RsTurtleGenericDataItem ;
|
||||
item->data_bytes = malloc(rssize) ;
|
||||
item->data_size = rssize ;
|
||||
memcpy(item->data_bytes,data,rssize) ;
|
||||
|
||||
delete[] data ;
|
||||
|
||||
printBinaryData(item->data_bytes,item->data_size) ;
|
||||
std::cerr << std::endl;
|
||||
|
||||
// do we have a working tunnel for that hash ?
|
||||
// If not, put on the contact's waiting list.
|
||||
|
||||
std::string virtual_peer_id ;
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
std::map<std::string,DistantMessengingContact>::const_iterator it = _messenging_contacts.find(hash) ;
|
||||
|
||||
if(it == _messenging_contacts.end())
|
||||
{
|
||||
std::cerr << "(EE) p3MsgService::sendTurtleData(): Can't find hash " << hash << " in recorded contact list." << std::endl;
|
||||
delete[] data ;
|
||||
return ;
|
||||
}
|
||||
|
||||
if(!it->second.status == RS_DISTANT_MSG_STATUS_TUNNEL_OK)
|
||||
{
|
||||
std::cerr << "p3MsgService::sendTurtleData(): tunnel is not ok. Putting items on waiting list." << std::endl;
|
||||
return ;
|
||||
}
|
||||
virtual_peer_id = it->second.virtual_peer_id ;
|
||||
}
|
||||
std::cerr << "p3MsgService::sendTurtleData(): Sending through virtual peer: " << virtual_peer_id << std::endl;
|
||||
std::cerr << " item->data_size = " << item->data_size << std::endl;
|
||||
std::cerr << " data = " ;
|
||||
|
||||
mTurtle->sendTurtleData(virtual_peer_id,item) ;
|
||||
}
|
||||
|
||||
void p3MsgService::receiveTurtleData(RsTurtleGenericTunnelItem *gitem,const std::string& hash,
|
||||
const std::string& virtual_peer_id,RsTurtleGenericTunnelItem::Direction direction)
|
||||
{
|
||||
RsTurtleGenericDataItem *item = dynamic_cast<RsTurtleGenericDataItem*>(gitem) ;
|
||||
|
||||
if(item == NULL)
|
||||
{
|
||||
std::cerr << "(EE) p3MsgService::receiveTurtleData(): item is not a data item. That is an error." << std::endl;
|
||||
return ;
|
||||
}
|
||||
std::cerr << "p3MsgService::sendTurtleData(): Receiving through virtual peer: " << virtual_peer_id << std::endl;
|
||||
std::cerr << " gitem->data_size = " << item->data_size << std::endl;
|
||||
std::cerr << " data = " ;
|
||||
printBinaryData(item->data_bytes,item->data_size) ;
|
||||
std::cerr << std::endl;
|
||||
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
std::map<std::string,DistantMessengingContact>::iterator it = _messenging_contacts.find(hash) ;
|
||||
|
||||
if(it == _messenging_contacts.end())
|
||||
{
|
||||
std::cerr << "(EE) p3MsgService::sendTurtleData(): Can't find hash " << hash << " in recorded contact list." << std::endl;
|
||||
return ;
|
||||
}
|
||||
|
||||
it->second.status = RS_DISTANT_MSG_STATUS_TUNNEL_OK ;
|
||||
it->second.last_hit_time = time(NULL) ;
|
||||
}
|
||||
|
||||
RsItem *itm = _serialiser->deserialise(item->data_bytes,&item->data_size) ;
|
||||
|
||||
RsMsgItem *mitm = dynamic_cast<RsMsgItem*>(itm) ;
|
||||
|
||||
if(mitm != NULL)
|
||||
{
|
||||
mitm->PeerId(hash) ;
|
||||
handleIncomingItem(mitm) ;
|
||||
}
|
||||
else
|
||||
{
|
||||
std::cerr << "(EE) p3MsgService::receiveTurtleData(): received item is not a RsMsgItem!!" << std::endl;
|
||||
delete itm ;
|
||||
}
|
||||
}
|
||||
|
||||
void p3MsgService::sendPrivateMsgItem(RsMsgItem *msgitem)
|
||||
{
|
||||
std::cerr << "p3MsgService::sendDistanteMsgItem(): sending distant msg item to peer " << msgitem->PeerId() << std::endl;
|
||||
std::cerr << " asking for tunnels" << std::endl;
|
||||
std::cerr << " recording msg info" << std::endl;
|
||||
|
||||
const std::string& hash = msgitem->PeerId() ;
|
||||
rsTurtle->monitorTunnels(hash,this) ; // create a tunnel for it, and put the msg on the waiting list.
|
||||
|
||||
{
|
||||
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
|
||||
|
||||
// allocate a new contact. If it does not exist, set its tunnel state to DN
|
||||
//
|
||||
std::map<std::string,DistantMessengingContact>::iterator it = _messenging_contacts.find(hash) ;
|
||||
|
||||
DistantMessengingContact& contact( _messenging_contacts[hash] ) ;
|
||||
|
||||
if(it == _messenging_contacts.end())
|
||||
it->second.status = RS_DISTANT_MSG_STATUS_TUNNEL_DN ;
|
||||
|
||||
contact.pending_messages.push_back(msgitem) ; // record the msg to be sent.
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue