diff options
| -rw-r--r-- | pica-client/skynet.cpp | 318 | ||||
| -rw-r--r-- | pica-client/skynet.h | 29 |
2 files changed, 232 insertions, 115 deletions
diff --git a/pica-client/skynet.cpp b/pica-client/skynet.cpp index a3bf44d..399b39e 100644 --- a/pica-client/skynet.cpp +++ b/pica-client/skynet.cpp @@ -40,78 +40,134 @@ PICA_client_callbacks cbs = { PICA_client_init(&cbs); -this->nodelink = NULL; +this->active_nodelink = NULL; this->acc = NULL; connect(this, SIGNAL(PeerCertificateReceived(QByteArray,QString,bool*)), this, SLOT(verify_peer_cert(QByteArray,QString,bool*)),Qt::DirectConnection); +event_loop_timer_id = startTimer(0); + } -void SkyNet::nodethread_finished() +void SkyNet::nodelink_activated(PICA_c2n *c2n) { -//удалить здесь завершившийся поток - for (int i=0;i<threads.count();i++) - { - if (threads[i]->isFinished()) - { - delete threads.takeAt(i); - break; - } - } + for (int i = 0; i < connecting_nodes.size(); i++) + { + if (connecting_nodes[i].first == c2n) + { + node_status_changed(connecting_nodes[i].second, true); + connecting_nodes.removeAt(i); + break; + } + } - if (threads.count() == 0) - { - self_aware = false; - emit LostSelfAwareness(); - } -} + if (self_aware) + { + PICA_close_c2n(c2n); + } + else + { + self_aware = true; + active_nodelink = c2n; -void SkyNet::nodethread_connected(QString addr, quint16 port, NodeThread *thread) -{ - if (self_aware) - { - emit BecameSelfAware(); + emit BecameSelfAware(); + // + //restore old peer connections, if any + QList<QByteArray> c2c_peer_ids; - //restore old peer connections, if any - QList<QByteArray> c2c_peer_ids; + //load undelivered messages from history + if (msgqueues.isEmpty()) + { + History h(config_dbname, Accounts::GetCurrentAccount().id); - //load undelivered messages from history - if (msgqueues.isEmpty()) - { - History h(config_dbname, Accounts::GetCurrentAccount().id); + msgqueues = h.GetUndeliveredMessages(); + } - msgqueues = h.GetUndeliveredMessages(); - } + //load file transfers to be completed + // sndfilequeues = ... - //load file transfers to be completed - // sndfilequeues = ... + QMap<QByteArray, QList<QString> > queues = msgqueues; - QMap<QByteArray, QList<QString> > queues = msgqueues; + queues.unite(sndfilequeues); - queues.unite(sndfilequeues); + c2c_peer_ids = queues.uniqueKeys(); - c2c_peer_ids = queues.uniqueKeys(); + for (int i = 0; i < c2c_peer_ids.size(); i++) + { + int ret; + struct PICA_c2c *chan = NULL; - for (int i = 0; i < c2c_peer_ids.size(); i++) - { - int ret; - write_mutex.lock(); - struct PICA_c2c *chan = NULL; + ret = PICA_new_c2c(active_nodelink, (const unsigned char*)c2c_peer_ids[i].constData(), NULL, &chan); - ret = PICA_new_c2c(nodelink, (const unsigned char*)c2c_peer_ids[i].constData(), NULL, &chan); + qDebug()<<"restoring c2c to "<<c2c_peer_ids[i]<<" ret ="<<ret<<"\n"; + } + + //start timer + retry_timer_id = startTimer(10000); + // + } - qDebug()<<"restoring c2c to "<<c2c_peer_ids[i]<<" ret ="<<ret<<"\n"; - write_mutex.unlock(); - } - //start timer - timer_id = startTimer(10000); - } } -void SkyNet::node_status_changed(QString addr, quint16 port, bool alive) +void SkyNet::nodelink_closed(PICA_c2n *c2n, int error) +{ + if (self_aware && c2n == active_nodelink) + { + active_nodelink = NULL; + emit LostSelfAwareness(); + } + else + { + for (int i = 0; i < connecting_nodes.size(); i++) + { + if (connecting_nodes[i].first == c2n) + { + connecting_nodes.removeAt(i); + break; + } + } + } +} + +void SkyNet::nodelink_failed(PICA_c2n *c2n, int error) +{ + for (int i = 0; i < connecting_nodes.size(); i++) + { + if (connecting_nodes[i].first == c2n) + { + switch(error) + { + case PICA_ERRPROTONEW: + + emit StatusMsg(QString("Node %1:%2 has newer protocol version. Please check if update for pica-client is available") + .arg(connecting_nodes[i].second.address) + .arg(connecting_nodes[i].second.port), + true); + break; + + case PICA_ERRPROTOOLD: + + emit StatusMsg(QString("Node %1:%2 has older protocol. Disconnected.") + .arg(connecting_nodes[i].second.address) + .arg(connecting_nodes[i].second.port), + false); + break; + } + + node_status_changed(connecting_nodes[i].second, false); + connecting_nodes.removeAt(i); + break; + } + } + if (connecting_nodes.empty()) + emit LostSelfAwareness(); +} + + + +void SkyNet::node_status_changed(Nodes::NodeRecord nr, bool alive) { - Nodes::NodeRecord nr = {addr, port}; nodes.UpdateStatus(nr, alive); } @@ -173,30 +229,54 @@ void SkyNet::verify_peer_cert(QByteArray peer_id, QString cert_pem, bool *verifi void SkyNet::timerEvent(QTimerEvent *e) { + if (e->timerId() == event_loop_timer_id) + { + if (connecting_nodes.size() > 0 || self_aware) + { + int ret; + QVector<struct PICA_c2n *> nodelinks; + + nodelinks.reserve(connecting_nodes.size() + 1); + + for (int i = 0; i < connecting_nodes.size(); i++) + nodelinks.append(connecting_nodes[i].first); + + if (self_aware) + nodelinks.append(active_nodelink); + + nodelinks.append(NULL); + + ret = PICA_event_loop(nodelinks.data(), NULL, 1); + + if (ret != PICA_OK) + emit StatusMsg("event loop error!", true);//show some error message + } + + return; + } + + if (self_aware) { - if (write_mutex.tryLock(10)) - { - QMap<QByteArray, QList<QString> > queues = msgqueues; - queues.unite(sndfilequeues); + QMap<QByteArray, QList<QString> > queues = msgqueues; + queues.unite(sndfilequeues); - QList<QByteArray> c2c_peer_ids = queues.uniqueKeys(); + QList<QByteArray> c2c_peer_ids = queues.uniqueKeys(); - c2c_peer_ids = filter_existing_chans(c2c_peer_ids); + c2c_peer_ids = filter_existing_chans(c2c_peer_ids); - for (int i = 0; i < c2c_peer_ids.size(); i++) - { - int ret; + for (int i = 0; i < c2c_peer_ids.size(); i++) + { + int ret; - struct PICA_c2c *chan = NULL; + struct PICA_c2c *chan = NULL; - ret = PICA_new_c2c(nodelink, (const unsigned char*)c2c_peer_ids[i].constData(), NULL, &chan); + ret = PICA_new_c2c(active_nodelink, (const unsigned char*)c2c_peer_ids[i].constData(), NULL, &chan); - qDebug()<<"restoring c2c to "<<c2c_peer_ids[i]<<" ret ="<<ret<<" in timer event\n"; + qDebug()<<"restoring c2c to "<<c2c_peer_ids[i]<<" ret ="<<ret<<" in timer event\n"; + + } - } - write_mutex.unlock(); - } } else { @@ -233,9 +313,9 @@ bool SkyNet::open_account() void SkyNet::Join(Accounts::AccountRecord &accrec) { - QList<Nodes::NodeRecord> nodelist; skynet_account = accrec; - nodelist = nodes.GetNodes(); + + QList<Nodes::NodeRecord> noderecords = nodes.GetNodes(); if (!acc) { @@ -243,12 +323,37 @@ void SkyNet::Join(Accounts::AccountRecord &accrec) return; } - if (nodelist.count()==0) + if (noderecords.count()==0) { - status = QObject::tr("No known Pica Pica nodes"); + emit StatusMsg(QString(tr("No known Pica Pica nodes")), true); return; } + if (!connecting_nodes.empty()) + { + for (int i = 0; i < connecting_nodes.size(); i++) + PICA_close_c2n(connecting_nodes[i].first); + + connecting_nodes.clear(); + } + + for (int i = 0; i < noderecords.size();i++) + { + int ret; + struct PICA_c2n *c2n = NULL; + + ret = PICA_new_c2n(acc, noderecords[i].address.toUtf8().constData(), noderecords[i].port, &c2n); + + if (ret == PICA_OK) + { + connecting_nodes.append(QPair<struct PICA_c2n *, Nodes::NodeRecord>(c2n, noderecords[i])); + } + else + { + node_status_changed(noderecords[i], false); + } + } +/* for (int i=0;i<nodelist.count() && !self_aware;i++) {//TODO FIXME сделать запуск потоков порциями, а не все сразу threads.append(new NodeThread(nodelist[i],&self_aware, acc,&nodelink, &write_mutex)); @@ -256,18 +361,30 @@ void SkyNet::Join(Accounts::AccountRecord &accrec) connect(threads.last(), SIGNAL(NodeStatusChanged(QString,quint16,bool)), this, SLOT(node_status_changed(QString,quint16,bool))); connect(threads.last(), SIGNAL(ConnectedToNode(QString,quint16,NodeThread*)), this, SLOT(nodethread_connected(QString,quint16,NodeThread*))); connect(threads.last(), SIGNAL(ErrorMsg(QString)), this, SIGNAL(ErrMsgFromNode(QString))); - } + }*/ } void SkyNet::Exit() { - for (int i = 0; i< threads.count(); i++) - threads[i]->CloseThread(); + if (self_aware && active_nodelink) + { + PICA_close_c2n(active_nodelink); + active_nodelink = NULL; + } self_aware = false; - killTimer(timer_id); + killTimer(retry_timer_id); + + msgqueues.clear(); + + for (int i = 0; i < connecting_nodes.size(); i++) + PICA_close_c2n(connecting_nodes[i].first); + + connecting_nodes.clear(); + + emit LostSelfAwareness(); } bool SkyNet::isSelfAware() @@ -280,7 +397,7 @@ void SkyNet::SendFile(QByteArray to, QString filepath) int ret = PICA_OK; struct PICA_c2c *iptr; -write_mutex.lock();//<< + if ( (iptr = find_active_chan(to)) ) { @@ -291,7 +408,7 @@ if ( (iptr = find_active_chan(to)) ) //show error somewhere somehow } - write_mutex.unlock();//>> + return; } @@ -303,14 +420,14 @@ if ( (iptr = find_active_chan(to)) ) { struct PICA_c2c *chan = NULL; - ret = PICA_new_c2c(nodelink, (const unsigned char*)to.constData(), NULL, &chan); + ret = PICA_new_c2c(active_nodelink, (const unsigned char*)to.constData(), NULL, &chan); QList<QString> l; l.append(filepath); sndfilequeues[to] = l; } -write_mutex.unlock();//>> + if (ret!= PICA_OK) { @@ -323,7 +440,7 @@ void SkyNet::AcceptFile(QByteArray from, QString filepath) int ret = PICA_OK; struct PICA_c2c *iptr; -write_mutex.lock();//<< + if ( (iptr = find_active_chan(from)) ) { ret = PICA_accept_file(iptr, filepath.toUtf8().data(), filepath.toUtf8().size()); @@ -334,7 +451,7 @@ if ( (iptr = find_active_chan(from)) ) //show error somewhere somehow } } -write_mutex.unlock();//>> + } void SkyNet::DenyFile(QByteArray from) @@ -342,7 +459,7 @@ void SkyNet::DenyFile(QByteArray from) int ret = PICA_OK; struct PICA_c2c *iptr; -write_mutex.lock();//<< + if ( (iptr = find_active_chan(from)) ) { ret = PICA_deny_file(iptr); @@ -353,7 +470,7 @@ if ( (iptr = find_active_chan(from)) ) //show error somewhere somehow } } -write_mutex.unlock();//>> + } void SkyNet::PauseFile(QByteArray peer_id, bool pause_sending) @@ -361,7 +478,7 @@ void SkyNet::PauseFile(QByteArray peer_id, bool pause_sending) int ret = PICA_OK; struct PICA_c2c *iptr; -write_mutex.lock();//<< + if ( (iptr = find_active_chan(peer_id)) ) { qDebug() << "calling PICA_pause_file(" << iptr << "," << pause_sending << ")\n"; @@ -372,7 +489,7 @@ if ( (iptr = find_active_chan(peer_id)) ) qDebug() << "PICA_pause_file() returned " << ret << "\n"; } } -write_mutex.unlock();//>> + } void SkyNet::ResumeFile(QByteArray peer_id, bool resume_sending) @@ -380,7 +497,7 @@ void SkyNet::ResumeFile(QByteArray peer_id, bool resume_sending) int ret = PICA_OK; struct PICA_c2c *iptr; -write_mutex.lock();//<< + if ( (iptr = find_active_chan(peer_id)) ) { ret = PICA_resume_file(iptr, (int)resume_sending ); @@ -390,7 +507,7 @@ if ( (iptr = find_active_chan(peer_id)) ) //show error somewhere somehow } } -write_mutex.unlock();//>> + } void SkyNet::CancelFile(QByteArray peer_id, bool cancel_sending) @@ -398,7 +515,7 @@ void SkyNet::CancelFile(QByteArray peer_id, bool cancel_sending) int ret = PICA_OK; struct PICA_c2c *iptr; -write_mutex.lock();//<< + if ( (iptr = find_active_chan(peer_id)) ) { ret = PICA_cancel_file(iptr, (int)cancel_sending ); @@ -408,7 +525,7 @@ if ( (iptr = find_active_chan(peer_id)) ) //show error somewhere somehow } } -write_mutex.unlock();//>> + } void SkyNet::SendMessage(QByteArray to, QString msg) @@ -416,7 +533,7 @@ void SkyNet::SendMessage(QByteArray to, QString msg) int ret = PICA_OK; struct PICA_c2c *iptr; -write_mutex.lock();//<< + if ( (iptr = find_active_chan(to)) ) { @@ -427,7 +544,7 @@ if ( (iptr = find_active_chan(to)) ) if (ret != PICA_OK) emit UnableToDeliver(to, msg); - write_mutex.unlock();//>> + return; } @@ -439,7 +556,7 @@ if ( (iptr = find_active_chan(to)) ) { struct PICA_c2c *chan = NULL; - ret = PICA_new_c2c(nodelink, (const unsigned char*)to.constData(), NULL, &chan); + ret = PICA_new_c2c(active_nodelink, (const unsigned char*)to.constData(), NULL, &chan); QList<QString> l; l.append(msg); @@ -447,7 +564,7 @@ if ( (iptr = find_active_chan(to)) ) } -write_mutex.unlock();//>> + if (ret != PICA_OK) emit UnableToDeliver(to, msg); @@ -506,10 +623,10 @@ void SkyNet::flush_queues(QByteArray to) struct PICA_c2c * SkyNet::find_active_chan(QByteArray peer_id) { -if (!nodelink) +if (!active_nodelink) return NULL; -struct PICA_c2c *iptr = nodelink->chan_list_head; +struct PICA_c2c *iptr = active_nodelink->chan_list_head; while(iptr) { @@ -526,10 +643,10 @@ QList<QByteArray> SkyNet::filter_existing_chans(QList<QByteArray> peer_ids) { QList<QByteArray> ret = peer_ids; - if (!nodelink) + if (!active_nodelink) return ret; - struct PICA_c2c *iptr = nodelink->chan_list_head; + struct PICA_c2c *iptr = active_nodelink->chan_list_head; while(iptr) { @@ -640,9 +757,6 @@ void SkyNet::emit_c2cClosed(QByteArray peer_id) //callbacks -//all callbacks are executed in separate thread, created in Nodethread instance, REMEMBER THAT !!! -// write_mutex is locked - void SkyNet::newmsg_cb(const unsigned char *peer_id, const char *msgbuf, unsigned int nb, int type) { QString msg = QString::fromUtf8(msgbuf, nb); @@ -800,17 +914,17 @@ void SkyNet::file_finished(const unsigned char *peer_id, int sending) void SkyNet::c2n_established_cb(struct PICA_c2n *c2n) { - + skynet->nodelink_activated(c2n); } -void SkyNet::c2n_failed_cb(struct PICA_c2n *c2n) +void SkyNet::c2n_failed_cb(struct PICA_c2n *c2n, int error) { - + skynet->nodelink_failed(c2n, error); } -void SkyNet::c2n_closed_cb(struct PICA_c2n *c2n) +void SkyNet::c2n_closed_cb(struct PICA_c2n *c2n, int error) { - + skynet->nodelink_closed(c2n, error); } void SkyNet::listener_error_cb(struct PICA_listener *lst, int errorcode) diff --git a/pica-client/skynet.h b/pica-client/skynet.h index b9e1298..3e03b87 100644 --- a/pica-client/skynet.h +++ b/pica-client/skynet.h @@ -2,12 +2,13 @@ #define SKYNET_H #include "nodes.h" -#include "nodethread.h" #include "../PICA_client.h" #include "../PICA_proto.h" #include "accounts.h" #include <QObject> #include <QMap> +#include <QVector> +#include <QPair> class SkyNet : public QObject @@ -17,7 +18,6 @@ public: SkyNet(); void Join(Accounts::AccountRecord &accrec); bool isSelfAware(); - QString& Status(); void Exit(); void SendMessage(QByteArray to, QString msg); @@ -38,7 +38,7 @@ signals: void LostSelfAwareness(); void PeerCertificateReceived(QByteArray peer_id, QString cert_pem, bool *verified); void CertificateForged(QByteArray peer_id, QString received_cert, QString stored_cert); - void ErrMsgFromNode(QString msg); + void StatusMsg(QString msg, bool is_critical); void ContactsUpdated(); void IncomingFileRequestReceived(QByteArray peer_id, quint64 file_size, QString filename); void OutgoingFileRequestAccepted(QByteArray peer_id); @@ -58,17 +58,18 @@ signals: void c2cClosed(QByteArray peer_id); private: - Nodes nodes; - QList<NodeThread*> threads; + Nodes nodes; + QList<QPair<struct PICA_c2n *, Nodes::NodeRecord> > connecting_nodes; bool self_aware; QString status; - struct PICA_c2n *nodelink; + struct PICA_c2n *active_nodelink; struct PICA_acc *acc; Accounts::AccountRecord skynet_account; - QMutex write_mutex; + QMap<QByteArray, QList<QString> > msgqueues; QMap<QByteArray, QList<QString> > sndfilequeues; - int timer_id; + int retry_timer_id; + int event_loop_timer_id; void timerEvent(QTimerEvent * e); @@ -133,16 +134,18 @@ private: static void c2n_established_cb(struct PICA_c2n *c2n); - static void c2n_failed_cb(struct PICA_c2n *c2n); + static void c2n_failed_cb(struct PICA_c2n *c2n, int error); - static void c2n_closed_cb(struct PICA_c2n *c2n); + static void c2n_closed_cb(struct PICA_c2n *c2n, int error); static void listener_error_cb(struct PICA_listener *lst, int errorcode); private slots: - void nodethread_finished(); - void nodethread_connected(QString addr, quint16 port, NodeThread *thread); - void node_status_changed(QString addr,quint16 port,bool alive); + void nodelink_activated(PICA_c2n *c2n); + void nodelink_closed(PICA_c2n *c2n, int error); + void nodelink_failed(PICA_c2n *c2n, int error); + + void node_status_changed(Nodes::NodeRecord nr,bool alive); void verify_peer_cert(QByteArray peer_id, QString cert_pem, bool *verified); }; |