BinaryProtocol.cpp

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

#include "BinaryProtocol.h"

/*BinaryProtocol::BinaryProtocol()
{
}

BinaryProtocol::~BinaryProtocol()
{
}*/
static pthread_mutex_t pktid_mutex = PTHREAD_MUTEX_INITIALIZER;
static int pktID=0;

BinaryProtocol::BinaryProtocol(BaseConfigFile* cf){
    //Ваяем мьютекс чтоб дождаться завершения работы потока, при отвязанном потоке
    pthread_mutex_init(&exit_mutex,NULL);
    pthread_mutex_init(&cs_mutex3,NULL);
    pthread_mutex_init(&cs_mutex4,NULL);
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);
    this->cf=cf; 
    this->max_pktsize=cf->getModuleAttributeInt(L"lib_file_opertion",L"max_packet_size",40960);
    cout<<"m pkt size"<<max_pktsize;
}

ssize_t BinaryProtocol::readToBuffer(void* buffer, size_t length) {
    //int sizeEnd=length;
    //while(sizeEnd>0){
    return recv(skt_id, buffer, length, MSG_WAITALL);
    //if (e<0){
    //sizeEnd-=e;
    //}else{
    //TODO Тут обработка исключения перехват его выше как-то и выход
    //                  exit(0);
    //}
    //  }
}

void BinaryProtocol::sendBuffer(const void* buffer, int length) {
    int err = 0;
    int l = length;
    char* buffero = (char*) buffer;
    while (l > 0) {
        err = send(skt_id, buffero, l, 0);
        if (err > 0) {
            #ifdef debug_output
            cout << "A OK";
            #endif
            buffero += err;
            l -= err;
        } else {
            cout << "A ERR";
            break;
        }
    }
}

void BinaryProtocol::sendPacket(BibaryPacket packet) {
    pthread_mutex_lock(&cs_mutex4);
    packet.header.headSize = sizeof (packet.header);
    sendBuffer(&packet.header, sizeof (packet.header));
    sendBuffer(packet.body, packet.header.bodySize); 
    pthread_mutex_unlock(&cs_mutex4);
}

BibaryPacket BinaryProtocol::readPacket() {
    BibaryPacket bp;
    if (readToBuffer(&bp.header, sizeof (bp.header)) < sizeof (bp.header)) {
        bp.header.headSize = 0;
        return bp;
    }
    if (sizeof (bp.header) < bp.header.headSize) {
        //Зарезервировано, на будующее если протокол будет расширятся типа
        size_t recervSize = sizeof (bp.header) - bp.header.headSize;
        void* buff = malloc(recervSize);
        if (readToBuffer(buff, recervSize) < recervSize) {
            bp.header.headSize = 0;
            free(buff);
            return bp;
        }
        free(buff);

    }
    bp.body = malloc(bp.header.bodySize);
    if (readToBuffer(bp.body, bp.header.bodySize) < bp.header.bodySize) {
        bp.header.headSize = 0;
        free(bp.body);
    }
    return bp;
}

void BinaryProtocol::setSendBufferData(const void* buffer, size_t size, unsigned long int pktID) {
    //size_t pktsize = 40960;
    size_t szForSend = size;
    char* buffer_cur = (char*) buffer;
    while (szForSend > 0) {
        BibaryPacket packet;
        packet.body = buffer_cur;
        packet.header.pktID = pktID;
        size_t ssend = max_pktsize;
        if (szForSend < max_pktsize) {
            ssend = szForSend;
        }
        packet.header.bodySize = ssend;
        sendPacket(packet);

        szForSend -= ssend;
        buffer_cur += ssend;
    }
}

void BinaryProtocol::setFillBuffer(void* buffer, size_t size, unsigned long int pktID) {
    BufferFiller* bf=new BufferFiller();
    bf->bufferSize = size;
    bf->forRead = size;

    bf->bufferC = buffer;
    bf->bufferS = buffer;

    bf->pktID = pktID;
    bf->allReaded=false;
    pthread_mutex_init(&(bf->mutex), NULL);

    pthread_cond_init(&(bf->cond_var),NULL);


    //Вставка и удаление не могут быть одновременными

    pthread_mutex_lock(&cs_mutex3);
    //pair < map<unsigned long int, BufferFiller>::iterator, bool> itBF =
    bufferFillers.insert(make_pair(pktID, bf));
    pthread_mutex_unlock(&cs_mutex3);
    
}

void BinaryProtocol::waiteFillBuffer(unsigned int pktID) {
    //TODO Ждем открытия семафора, и убиваем его

    pthread_mutex_lock(&cs_mutex3);
    map<unsigned long int, BufferFiller*>::iterator itBF = bufferFillers.find(pktID);
    pthread_mutex_unlock(&cs_mutex3);

    if (itBF != bufferFillers.end()) {
        //pthread_mutex_unlock(&cs_mutex3);
        //pthread_mutex_t& t=;
        //cout <<
        pthread_mutex_lock(&(itBF->second->mutex));
        while (!itBF->second->allReaded){
            pthread_cond_wait(&(itBF->second->cond_var),&(itBF->second->mutex));
        }
        pthread_mutex_unlock(&(itBF->second->mutex));

        pthread_mutex_lock(&cs_mutex3);

        pthread_mutex_destroy(&(itBF->second->mutex));
        pthread_cond_destroy(&(itBF->second->cond_var));
        delete (itBF->second);
        //free(itBF->second.mutex);
        bufferFillers.erase(itBF);
        
        pthread_mutex_unlock(&cs_mutex3);
        //pthread_mutex_destroy(&t);
    }else{
        cout<<"!!!!мютекса нету!!!!"<<endl;
    }
}

void BinaryProtocol::parse() {
    BibaryPacket bp;
    while (1) {
        #ifdef debug_output
        cout << "READING BIN PACKET 1" << endl;
        #endif
        bp = readPacket();
        if (bp.header.headSize == 0) {
            break;
            //return;
        }
        #ifdef debug_output
        cout << "READING BIN PACKET OK id:" << bp.header.pktID << " headsize:" << bp.header.headSize << " bodySize:" << bp.header.bodySize << endl;
        #endif
        pthread_mutex_lock(&cs_mutex3);
        map<unsigned long int, BufferFiller*>::iterator itBF = bufferFillers.find(bp.header.pktID);
        if (itBF != bufferFillers.end()) {
            int sz = bp.header.bodySize;
            if (itBF->second->forRead < bp.header.bodySize) {
                sz = itBF->second->forRead;
            }
            memcpy(itBF->second->bufferC, bp.body, sz);
            itBF->second->forRead -= sz;
            itBF->second->bufferC = ((char*) itBF->second->bufferC) + sz;
            #ifdef debug_output
            cout << "For read size:" << itBF->second.forRead << endl;
            #endif
            if (itBF->second->forRead == 0) {
                //TODO Открыть семафор
                #ifdef debug_output
                cout << "Opening Semaphore BIN P" << endl;
                #endif

                pthread_mutex_lock(&(itBF->second->mutex));
                itBF->second->allReaded=true;
                pthread_cond_broadcast(&(itBF->second->cond_var));
                pthread_mutex_unlock(&(itBF->second->mutex));
                

                //Удаление не может быть произведено паралельно со вставкой и прочем
                //bufferFillers.erase(itBF);
            }
        }else{
            cout<<"!!!!потеряли пакет!!!!"<<endl;
        }
        pthread_mutex_unlock(&cs_mutex3);
        //тут что то делаем с пакетом
        //bp=readPacket(); НЕ ПОНЯЛ!
        list<BPHandler*>::iterator bpb = handlers.begin();
        while (bpb != handlers.end()) {
            (*bpb)->handleBinPacket(bp);
            ++bpb;
        }
        free(bp.body);
    }
}

void* binary_threadFunc(void* clnt) {
    BinaryProtocol* bp = static_cast<BinaryProtocol*> (clnt);
    pthread_mutex_lock(&(bp->exit_mutex));
    bp->parse();
    #ifdef debug_output
    cout<<"Выход из потока"<<endl;
    #endif
    pthread_mutex_unlock(&(bp->exit_mutex));
    return NULL;
}

unsigned long int BinaryProtocol::getNextId(){
    pthread_mutex_lock(&pktid_mutex);
    unsigned long int res=++pktID;
    pthread_mutex_unlock(&pktid_mutex);
    return res;
}

void BinaryProtocol::startTread(int skt_id) {
    //cout<<"вход в поток"<<endl;
    this->skt_id = skt_id;
    pthread_create(&binThread, &attr, binary_threadFunc, this);
}
BinaryProtocol::~BinaryProtocol(){
    //Убиваем сокет чтобы recv вышел не знаю как его лучше завершить?
    //cout<<"Деструктор бин протокола"<<endl;
    shutdown(skt_id, SHUT_RDWR);
    close(skt_id);
    pthread_mutex_lock(&exit_mutex);
    pthread_mutex_unlock(&exit_mutex);
    pthread_mutex_destroy(&exit_mutex);
    //
    pthread_attr_destroy(&attr);
    pthread_mutex_destroy(&cs_mutex3);
    pthread_mutex_destroy(&cs_mutex4);
    
}