Protocol.cpp

См. документацию.

/*
 * Protocol.cpp
 *
 *  Created on: 09.04.2011
 *      Author: nick
 */

#include "Protocol.h"
#include <errno.h>

/*template <class Listener_T>
Protocol<Listener_T>::Protocol() {
        // TODO Auto-generated constructor stub
}*/
void* node_threadFunc(void* clnt) {
    pair<Protocol*, Node*>* pv = static_cast<pair<Protocol*, Node*>*> (clnt);
    Protocol* pt = pv->first;
    Node* cn = pv->second;
    delete pv;
    pt->handlerNode(cn);
    //=====123

    pthread_mutex_lock(&(pt->mutDelEndNodRu));
    --(pt->num_run_threads);
    if((pt->num_run_threads)==0){pthread_cond_signal(&(pt->cond_del_element));}
    pthread_mutex_unlock(&(pt->mutDelEndNodRu));

    //=====123
/*    pthread_mutex_lock(&(pt->cs_mutexUNIC));
    if((--(pt->num_run_threads))==0){
        #ifdef debug_output
        debug_out cout<<"Разблокируем мьютекс потоков"<<endl;
        #endif
        pthread_mutex_unlock(&pt->cs_mutexEndWork);
    }
    pthread_mutex_unlock(&(pt->cs_mutexUNIC));*/
    return NULL;
}

void
characterData(void *userData, const XML_Char *s, int len) {
    Protocol* protocol = (Protocol*) userData;
    protocol->curNode->addXMLBody(string(s, len));
    #ifdef debug_output
    static pthread_mutex_t deta_m=PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_lock(&deta_m);
    cout<<"deta!:"<<string(s,len)<<endl;
    pthread_mutex_unlock(&deta_m);
    #endif
}

void
startTag(void *userData, const XML_Char *name, const XML_Char **atts) {
    #ifdef debug_output
    cout << "start_tag" << endl;
    #endif
    Protocol* protocol = (Protocol*) userData;
    protocol->curDeph++;
    map<string, string> attrsMap;
    const XML_Char **curAttr = atts;
    while (*curAttr != NULL) {
        #ifdef debug_output
        cout << "name:" << *curAttr << " value: " << *(curAttr + 1) << endl;
        #endif
        attrsMap[*curAttr] = *(curAttr + 1);
        curAttr = curAttr + 2;
    }
    pthread_mutex_lock(&(protocol->cs_mutexNewNode));
    Node* curNode = new Node(name, attrsMap, protocol->curNode);
    pthread_mutex_unlock(&(protocol->cs_mutexNewNode));
    if(protocol->curNode==NULL){
        protocol->topNode=curNode;
    }
    protocol->curNode = curNode;
    if (protocol->curDeph == 1) {
        protocol->startXMLTagNodeHandler(curNode);
    }
}

void
endTag(void *userData, const XML_Char *name) {
    #ifdef debug_output
    cout << "end_tag" << endl;
    #endif
    Protocol* protocol = (Protocol*) userData;
    protocol->curDeph--;
    //protocol->curNode=protocol->curNode->getParent();
    Node* curnP = protocol->curNode->getParent();
    if (protocol->curDeph == 1) {//Пока нет начального тега!
        //if (protocol->curDeph==0){
        //      Node* curn=curNode;
        //      curNode=curn->getParent();
        pair<Protocol*, Node*>* pv = new pair<Protocol*, Node*>(protocol, protocol->curNode);
        //====
        pthread_mutex_lock(&(protocol->mutDelEndNodRu));
        ++(protocol->num_run_threads);
        pthread_mutex_unlock(&(protocol->mutDelEndNodRu));
        //pthread_mutex_lock(&(protocol->cs_mutexUNIC));
        //pthread_mutex_trylock(&protocol->cs_mutexEndWork);
        //++protocol->num_run_threads;
        //pthread_mutex_unlock(&(protocol->cs_mutexUNIC));
        pthread_create(&(protocol->nodeThread), &(protocol->attr), node_threadFunc, pv);
        //node_threadFunc(pv);
        //====
        //protocol->handlerNode(protocol->curNode);
    } else {
        if (protocol->curDeph == 0) {
            protocol->endXMLTagNodeHandler(protocol->curNode);
        }
        //protocol->curNode=protocol->curNode->getParent();
    }
    protocol->curNode = curnP;
    //===!!!
    /*  if (protocol->curDeph==1){//Пока нет начального тега!
            //if (protocol->curDeph==0){
                    protocol->handlerNode();
            }else if (protocol->curDeph==0){
                    protocol->endXMLTagNodeHandler(protocol->curNode);
                    protocol->curNode=protocol->curNode->getParent();
            }else{
                    protocol->curNode=protocol->curNode->getParent();
            }*/
    #ifdef debug_output
    cout << "== end end tag!== " << endl;
    #endif
}

void Protocol::startXMLTagNodeHandler(Node* node) {

}

void Protocol::endXMLTagNodeHandler(Node* node) {

}

void Protocol::handlNodes(Node* curNodeL, TagHandelrsMapT::iterator begin, TagHandelrsMapT::iterator end) {
    #ifdef debug_output
    cout << "handle_nodes" << endl;
    #endif
    TagHandelrsMapT::iterator it = begin;
    while (it != end) {
        (*it).second->handleTag(*curNodeL);
        it++;
    }
}

/*void Protocol::handlIQs(IQHandelrsMapT::iterator begin,IQHandelrsMapT::iterator end,TagIQ &iq){
        cout<<"handle_IQs"<<endl;
        IQHandelrsMapT::iterator it=begin;
        //cout<< "IQTname: "<<iq.getSubNode()->getName()<<" IQTNS: "<<iq.getSubNode()->getNamesapace()<<endl;
        while (it!=end){
                cout<< "Handle node IQ!"<<endl;
                //Вероятно сама нода должна обрабатываться в отдельном потоке
                (*it).handler->handleTag(iq);
                it++;
        }
}*/

void Protocol::handlerIQ(Node* curNodeL) {
    IQHandelrsMapT::iterator it = iqHandlers.begin();
    //IQHandelrsMapT::iterator it2;
    TagIQ iq(*curNodeL); //=new TagIQ(*curNode);
    while (it != iqHandlers.end()) {
        Node* subnode = NULL;
        if (it->xmlns != "") {
            subnode = iq.getKind(it->xmlns);
            if ((subnode != NULL) &&
                    ((subnode->getName() == it->subtagName) || (it->subtagName == ""))) {
                it->handler->handleTag(iq);
            }
        } else
            if (it->subtagName != "") {
            _Kinds_iterator it1 = iq.kindsNameBegin(it->subtagName);
            _Kinds_iterator it2 = iq.kindsNameEnd(it->subtagName);
            if (it1 != it2) {
                it->handler->handleTag(iq);
            }
        } else {
            it->handler->handleTag(iq);
        }
        //if (((it->xmlns=="")||(it->xmlns==iq.))&&((it->subtagName)))
        ++it;
    }
    //Тут надо обработать эксепшон
    /*  Node* subn=iq.getSubNode();
            cout<<"DEBUG IQ HANDLERS"<<endl;
            for (IQHandelrsMapT::iterator iit=iqHandlers.begin();
                            iit!=iqHandlers.end();iit++){
                    cout<< "name: "<<(*iit).first.first<<" namespace: "<<(*iit).first.second<<endl;
            }
            cout<<"=== === === === =="<<endl;
            //Непонятно как работает сравнение у pair, операторы там перегружены в STL, но как то странно...
            //if (subn->getNamesapace().length()>0){
                    it1=iqHandlers.lower_bound(make_pair(subn->getName(),subn->getNamesapace()));
                    it2=iqHandlers.upper_bound(make_pair(subn->getName(),subn->getNamesapace()));
                    handlIQs(it1,it2,iq);
            //}

            cout<<"handlerIQ 1 name: "<<subn->getName()<< " namespace: "<<subn->getNamesapace()<<endl;
            if(subn->getNamesapace().length()>0){
                    it1=iqHandlers.lower_bound(make_pair(subn->getName(),""));
                    it2=iqHandlers.upper_bound(make_pair(subn->getName(),""));
                    handlIQs(it1,it2,iq);
            }
        cout<<"handlerIQ 2"<<endl;

        //if(subn->getName().length()>0){
            it1=iqHandlers.lower_bound(make_pair("",subn->getNamesapace()));
            it2=iqHandlers.upper_bound(make_pair("",subn->getNamesapace()));
            handlIQs(it1,it2,iq);
        //}
        cout<<"handlerIQ 3"<<endl;*/
    manadgeIQResp(iq);
    //Node* curn=curNode;
    //curNode=curn->getParent();
    //и удалять всегда
    //delete iq;
    //delete curn;
}

void Protocol::handlerNode(Node* curNodeL) {
    TagHandelrsMapT::iterator it1;
    TagHandelrsMapT::iterator it2;
    #ifdef debug_output
    cout << "handle node name:" << curNodeL->getName() << " namespace: " << curNodeL->getNamesapace() << endl;
    #endif
    //Непонятно как работает сравнение у pair, операторы там перегружены в STL, но как то странно...
    //if (curNode->getNamesapace().length()>0){
    it1 = tagHandlers.lower_bound(make_pair(curNodeL->getName(), curNodeL->getNamesapace()));
    it2 = tagHandlers.upper_bound(make_pair(curNodeL->getName(), curNodeL->getNamesapace()));
    handlNodes(curNodeL, it1, it2);
    //}

    if (curNodeL->getNamesapace().length() > 0) {
        it1 = tagHandlers.lower_bound(make_pair(curNodeL->getName(), ""));
        it2 = tagHandlers.upper_bound(make_pair(curNodeL->getName(), ""));
        handlNodes(curNodeL, it1, it2);
    }

    //if(curNode->getName().length()>0){
    it1 = tagHandlers.lower_bound(make_pair("", curNodeL->getNamesapace()));
    it2 = tagHandlers.upper_bound(make_pair("", curNodeL->getNamesapace()));
    handlNodes(curNodeL, it1, it2);
    //}

    //Пребор для qi,если iq
    if (curNodeL->getName() == "iq") {
        handlerIQ(curNodeL);
    }
    //  Node* curn=curNode;
    //  curNode=curn->getParent();
    pthread_mutex_lock(&cs_mutexNewNode);
    delete curNodeL;
    pthread_mutex_unlock(&cs_mutexNewNode);
}

void startDocType(void *userData, const XML_Char *doctypeName, const XML_Char *sysid,
        const XML_Char *pubid, int has_internal_subset) {
    #ifdef debug_output
    cout << "start doc type" << endl;
    #endif
}

void endDocType(void *userData) {
    #ifdef debug_output
    cout << "end doc type" << endl;
    #endif
}

void Protocol::parserCreate(int sktID) {
    xml_parser = XML_ParserCreate("UTF-8");
    if (!xml_parser) {
        std::cerr << "Couldn't allocate memory for parser\n";
        exit(-1);
    }
    curDeph = 0;
    XML_SetUserData(xml_parser, this);
    XML_SetCharacterDataHandler(xml_parser, characterData);
    XML_SetElementHandler(xml_parser, startTag, endTag);
    XML_SetDoctypeDeclHandler(xml_parser, startDocType, endDocType);
    skt_id = sktID;
}

void Protocol::startParsing(/*int sktID*/) {
    int len;
    //skt_id=sktID;
    do {
        #ifdef debug_output
        cout << "parse next buffer" << endl;
        #endif
        len = recv(skt_id, read_Buffer, BUFFSIZE, 0);
        if (len <= 0) {
            #ifdef debug_output
            cout << "1" << endl;
            #endif
            XML_Parse(xml_parser, read_Buffer, 0, 1);
            XML_ParserFree(xml_parser);
            break;
        } else {
            #ifdef debug_output
            cout << "2" << endl;
            #endif
            if (XML_Parse(xml_parser, read_Buffer, len, 0) == XML_STATUS_ERROR) {
                cout << " XML ERROR! :" <<
                        XML_ErrorString(XML_GetErrorCode(xml_parser)) << "at line:" << XML_GetCurrentLineNumber(xml_parser) << endl;
                break;
            }
        }
    } while (len > 0);
    //Ждем завершения ВСЕХ потоков
    #ifdef debug_output
    cout<<"Ждем все потоки, потоков:"<<num_run_threads<<endl;
    #endif
    //pthread_mutex_lock(&cs_mutexEndWork);

    pthread_mutex_lock(&mutDelEndNodRu);
    //Тут ждем переменную состояния
    while(num_run_threads!=0){pthread_cond_wait(&cond_del_element,&mutDelEndNodRu);}
    pthread_mutex_unlock(&mutDelEndNodRu);

    #ifdef debug_output
    cout<<"Дождались все потоки"<<endl;
    #endif
    //Непонятно зачем но valring что то хочет
    pthread_mutex_lock(&cs_mutexNewNode);
    if (topNode!=NULL){
        #ifdef debug_output
        cout<<"OK"<<endl;
        #endif
        delete topNode;

    }
    pthread_mutex_unlock(&cs_mutexNewNode);
    #ifdef debug_output
    cout<<"удалили корневую ноду"<<endl;
    #endif

    //123 pthread_mutex_unlock(&cs_mutexEndWork);
    #ifdef debug_output
    cout << "end working client" << endl;
    #endif
}

Protocol::Protocol() {
    #ifdef debug_output
    cout<<"конструктор протокола"<<endl;
    #endif
    
    curDeph = 0;
    curNode = NULL;
    topNode=NULL;
    curNodeSend = NULL;
    //addTagHandler(this,"iq","");
    //=======================================
    //pthread_mutexattr_init(&mutDelEndNodRuAttr);
    //pthread_mutexattr_settype(&mutDelEndNodRuAttr,PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(&mutDelEndNodRu,NULL/*&mutDelEndNodRuAttr*/);
    pthread_cond_init(&cond_del_element,NULL);
    num_run_threads=0;
    //=======================================
    pthread_mutex_init(&cs_mutex,NULL);
    pthread_mutex_init(&cs_mutex1,NULL);
//123    pthread_mutex_init(&cs_mutexUNIC,NULL);
    //123 pthread_mutex_init(&cs_mutexEndWork,NULL);
    pthread_mutex_init(&cs_mutexNewNode,NULL);
    pthread_mutex_init(&cs_mutexDEL,NULL);
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);
    //Нахрена это надо но valgrind говорит о возможной ошибке
    //pthread_mutex_lock(&cs_mutexUNIC);
    
    //pthread_mutex_unlock(&cs_mutexUNIC);

}

//template <class Listener_T>

void Protocol::addTagHandler(TagHandler<Node>::PTagHandler handler, string name, string xmlns) {
    tagHandlers.insert(make_pair(make_pair(name, xmlns), handler));
}

//bool delIQPredicate(IQHandleD hd){
    //cout<<"addr:"<<reinterpret_cast<long int> (hd.handler)<<"addr2:"<<reinterpret_cast<long int> (valPredIQDEL)<<" "<<(valPredIQDEL==hd.handler)<<endl;
    //return valPredIQDEL==hd.handler;
//}

void Protocol::removeIQHandler(TagHandler<TagIQ>::PTagHandler handler){
    pthread_mutex_lock(&cs_mutexDEL);
    IQHandelrsMapT::iterator it=iqHandlers.begin();
    while(it!=iqHandlers.end()){
        if(it->handler==handler){
            iqHandlers.erase(it);
            #ifdef debug_output
            cout<<"удалено"<<endl;
            #endif
            break;
        }
        ++it;
    }
  /*      valPredIQDEL=handler;
        int sz1=iqHandlers.size();
        remove_if(iqHandlers.begin(),iqHandlers.end(),delIQPredicate);
        cout<<"удалено элементов"<<sz1-iqHandlers.size()<<endl;*/
    pthread_mutex_unlock(&cs_mutexDEL);
}

void Protocol::addIQHandler(TagHandler<TagIQ>::PTagHandler handler, string xmlns, string name) {    
    #ifdef debug_output
    cout << "add iq handler" << name << " " << xmlns << endl;
    #endif
    IQHandleD hd;
    hd.xmlns = xmlns;
    hd.subtagName = name;
    hd.handler = handler;
    pthread_mutex_lock(&cs_mutexDEL);
    iqHandlers.push_back(hd);
    pthread_mutex_unlock(&cs_mutexDEL);
    //iqHandlers.insert(make_pair(make_pair(name,xmlns), handler));    
}

void Protocol::sendString(string data) {
    pthread_mutex_lock(&cs_mutex);
    #ifdef debug_output
    cout << "send string" << endl << "DATA:" << data << endl;
    #endif
    const char* buf = data.data();
    int pos = 0;
    int err = 0;
    int l = data.length();
    while (pos < l) {
        err = send(skt_id, buf + pos, l - pos, 0);
        if (err > 0) {
            #ifdef debug_output
            cout << "A OK";
            #endif
            pos += err;
        } else {
            cout << "A ERR";
            break;
        }
    }
    pthread_mutex_unlock(&cs_mutex);
}
//template <class Listener_T>

void Protocol /*<Listener_T>*/::sendNode(Node &stanza) {
    sendString(stanza.toXMLString());
}

void Protocol::sendNodeStart(Node &stanza) {
    sendString(stanza.toXMLStringStart());
}

void Protocol::sendNodeEnd(Node &stanza) {
    sendString(stanza.toXMLStringEnd());
}

void Protocol::manadgeIQResp(TagIQ& st) {
    //IQRespDKey key;
    //key.id=st.getID();
    //key.subtagName=st.getKind();//getSubNode()->getName();
    //key.xmlns=st.getSubNode()->getNamesapace();
    //multimap</*IQRespDKey*/string,IQRespD>::iterator b=req_id.lower_bound(st.getID());
    //multimap</*IQRespDKey*/string,IQRespD>::iterator be=req_id.upper_bound(st.getID());
    pthread_mutex_lock(&cs_mutex1);
    pair<multimap<string, IQRespD>::iterator, multimap<string, IQRespD>::iterator> it = req_id.equal_range(st.getID());
    #ifdef debug_output
    cout << "respCount" << req_id.size() << endl;
    #endif
    while (it.first != it.second) {
        #ifdef debug_output
        cout << "iq from get F:" << st.getFrom() << endl;
        #endif
        IQRespD* rd = &(it.first->second);
        Node *n = st.getKind(rd->xmlns);
        if ((n != NULL) && (n->getName() == rd->subtagName)) {
            rd->result = st;
            struct sembuf sops;
            memset(&sops, 0, sizeof sops);
            sops.sem_op = 1;
            #ifdef debug_output
            cout << "===opening semaphore!!!===" << endl;
            #endif
            //На всякий случай, вдруг что
            //int csem=it.first->second.semID;
            int sem_id = rd->semID;
            //it.first->second.semID=-1;
            rd->semID = -1;
            semop(/*csemrd.semID*/sem_id, &sops, 1);
            semctl(/*csemrd.semID*/sem_id, 0, IPC_RMID);
        }
        ++it.first;
    }
    pthread_mutex_unlock(&cs_mutex1);
}

TagIQ Protocol::sendIQResp(TagIQ& st, string resultXMLNS, string resultSubTagName) {

    struct sembuf sops;
    int cursem;
    struct timespec timeout;

    memset(&sops, 0, sizeof sops);
    memset(&timeout, 0, sizeof timeout);

    cursem = semget(IPC_PRIVATE, 1, 0777 | IPC_CREAT);
    #ifdef debug_output
    cout << "SEMAPHORE IDENTIFIER!: " << cursem << " errNo:" << errno << endl;
    #endif
    //ENOMEM
    sops.sem_op = -1;
    timeout.tv_sec = 600;
    //IQRespDKey key;
    //key.
    pthread_mutex_lock(&cs_mutex1);
    string key = st.getID();
    IQRespD rd;
    rd.semID = -1;
    multimap<string, IQRespD>::iterator it = req_id.insert(make_pair(key, rd));

    it->second.semID = cursem;
    it->second.subtagName = resultSubTagName;
    it->second.xmlns = resultXMLNS;
    pthread_mutex_unlock(&cs_mutex1);
    //req_id[key].semID=cursem;
    //req_id[key].resultXMLNS=resultXMLNS;

    sendNode(st);


    //Тут ждать открытия семафора
    semtimedop(cursem, &sops, 1, &timeout);

    //cout<<"iq from get:"<<req_id[key].result.getFrom()<<endl;
    pthread_mutex_lock(&cs_mutex1);
    TagIQ res(it->second.result); //req_id[key].result;
    req_id.erase(it/*key*/);
    pthread_mutex_unlock(&cs_mutex1);
    return res;
}

/*bool operator<(const IQRespDKey &obj1, const IQRespDKey &obj2){
        return (obj1.id<obj2.id) && (obj1.subtagName<obj2.subtagName) && (obj1.xmlns<obj2.xmlns);
}*/

//bool IQRespDKey::operator<(const IQRespDKey &obj2) const{
//      return (this->id<obj2.id) && (this->subtagName<obj2.subtagName) && (this->xmlns<obj2.xmlns);
//      return (this->id<obj2.id) || (this->subtagName<obj2.subtagName) || (this->xmlns<obj2.xmlns);
//}

/*bool IQRespDKey::operator==(const IQRespDKey &obj2) const{
        return (this->id==obj2.id) && (this->subtagName==obj2.subtagName) && (this->xmlns==obj2.xmlns);
}

bool IQRespDKey::operator!=(const IQRespDKey &obj2) const{
        return !((*this) == obj2);
}*/
/*void Protocol::handleTag(Node &tag){
        //tag1->getAttribute("xmlns");
        if (tag.getName()=="iq"){
                handlerIQ();
        }
}*/

/*template <class Listener_T>*/
Protocol::~Protocol() {
    // TODO Auto-generated destructor stub
    //req_id.erase
    //=========================================================
    //pthread_mutexattr_destroy(&mutDelEndNodRuAttr);
    pthread_mutex_destroy(&mutDelEndNodRu);
    pthread_cond_destroy(&cond_del_element);
    //=========================================================
    
    pthread_attr_destroy(&attr);    
    if(curNodeSend!=NULL){
        cout<<"delete curnodesend!"<<endl;
        delete curNodeSend;
    }
    cout<<"деструктор протокола"<<endl;
//123    pthread_mutex_destroy(&cs_mutexUNIC);
    pthread_mutex_destroy(&cs_mutex);
    pthread_mutex_destroy(&cs_mutex1);
    pthread_mutex_destroy(&cs_mutexNewNode);
    pthread_mutex_destroy(&cs_mutexDEL);
}

IQRespD::~IQRespD() {
    if (semID != -1) {
        semctl(semID, 0, IPC_RMID);
    }
}