zbsread.h

Go to the documentation of this file.

#ifndef ZBSREAD_H
#define ZBSREAD_H

#include <zmq.h>
#include <assert.h>
#include <signal.h>

#if HAVE_JSON
#include <json/json.h>
#endif

#include <zbsDataHolders.h>
#include <zbsdtHelper.h>

using namespace CAFE_BSHELPER;

//Preferred to BitsetV because of speed

IntegerBitset<unsigned char> bsetUChar; 
IntegerBitset<char> bsetChar; 
IntegerBitset<unsigned short> bsetUShort; 
IntegerBitset<short> bsetShort; 
IntegerBitset<unsigned int> bsetUInt; 
IntegerBitset<int> bsetInt; 
IntegerBitset<unsigned long long> bsetULongLong; 
IntegerBitset<long long> bsetLongLong; 

//IntegerBitsetV<unsigned char> bsetUCharV; 
//IntegerBitsetV<char> bsetCharV; 
//IntegerBitsetV<unsigned short> bsetUShortV; 
//IntegerBitsetV<short> bsetShortV; 
//IntegerBitsetV<unsigned int> bsetUIntV; 
//IntegerBitsetV<int> bsetIntV; 
//IntegerBitsetV<unsigned long long> bsetULongLongV; 
//IntegerBitsetV<long long> bsetLongLongV; 

FloatBitset<float>  bsetFloat;
FloatBitset<double> bsetDouble;
//FloatBitsetV<float>  bsetFloatV;
//FloatBitsetV<double> bsetDoubleV;


bool bsdtInsertFlag=false;
bool dataHeaderReceived=false;

unsigned short inDumpFlag=2;
std::string hashIs="";
std::string hashOriginal="";
unsigned short hashOriginalFlag=0;
bool fillBSPV=false;
std::vector<std::string> bsPV;

bool fill_bs_read_PV=false;
std::vector<std::string> bs_read_PV;



static void
z_bsread_dbpm (DBPMKeeper & dbpm)
{
                if (inDumpFlag==1) {
                        std::cout << "zeroMQ socket is busy " << std::endl;
                        std::cout << "waiting for zmq timeout " << std::endl;
                }
                
                //puts ("//START----------------------------------------//");
                
                void * socket = dbpm.subscriber;
                
                #if HAVE_JSON
                        Json::Value parsedFromString;
                        Json::Reader reader;
                        bool parsingSuccessful;
                        Json::FastWriter fastWriter;
                #endif
                
                int64_t more;           //  Multipart detection
    more = 0;
    size_t more_size = sizeof (more);
                
                int bsPVIdx=-1;
                dbpm.status=ICAFE_NORMAL;

                int nZeroSize=0;

    //std::cout << "df = " << inDumpFlag << std::endl;  

          while (inDumpFlag==1) {
                  std::cout << "df/ = " << inDumpFlag << std::endl; 
                        std::cout << " sleeping " << std::endl;                         
                }

                int subMessage=0;               
                int nSequentialHeader=0;

    while (1) {
                          inDumpFlag=1;
                                                        
                        //puts ("//WHILE LOOP ----------------------------------------//");
                        //std::cout << "subMessage " << subMessage << std::endl;
        //  Process all parts of the message
        zmq_msg_t message;
        zmq_msg_init (&message);
                                
        size_t size = zmq_msg_recv (&message, socket, 0);
                                //puts ("//MESSAGE RECEIVED     ----------------------------------------//");
                                
                                if (size == -1) {
                                        std::cout << " Error is " << zmq_errno() << " " << zmq_strerror(zmq_errno()) << std::endl;
                                        //Resource unavailable means that there is nothing to read now
                                        
                                        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                                        //std::cout << "message mulipart --> more == " << more << std::endl;
                                         zmq_msg_close (&message);
                                         //if (!more) {
                                         //                dbpm.status=ERRNO_EAGAIN; //ICAFE_ERRNO_BASE +zmq_errno() ;
                                         //                std::cout << "EARLY BREAK subMessage total: " << subMessage << std::endl;
                                         //                puts ("//------------------------------------END-------------------------------------//");
                                         //              break;      //  Last message
                                         // }



                                         dbpm.status=ICAFE_ERRNO_BASE+zmq_errno() ;
                                         break;
                                } 
                                
                                
                                else if (size == 0) {

                        
                                        ++nZeroSize;


                                         //zmq_msg_close (&message);
                                         //dbpm.status=ECAFE_NODATA;
                                         //break;
                                         //Comes in pairs; one for val one for timestamp
                                         if (nZeroSize%2==1) {
                                           ++bsPVIdx;
                                                if (bsPVIdx >0) {
                                                   std::cout << bsPV[bsPVIdx] << std::endl;
                                                }
                                         }

                                         ++subMessage;
                                         
                                         //continue;
                                } 
        else { 
                        
        //  Dump the message as text or binary
        char *data = (char*)zmq_msg_data (&message);
        int is_text = 1;
        int char_nbr;
                                
                                //char cmd[5000]="";
                                
        for (char_nbr = 0; char_nbr < size; char_nbr++)
            if ((unsigned char) data [char_nbr] < 32
            ||  (unsigned char) data [char_nbr] > 127)
                is_text = 0;


                //TExt is two header files
                //non-text size=8 is X1, Y1, Q1
                //non-text size=2 is -VALID
                //non-text size=16 is timestamp

                /*
                printf (" SIZE [%03d] ", size);


                if (is_text) {
                        std::cout << " TEXT +++++++++++ " << std::endl;
                }
                else {
                        std::cout << " NOT TEXT +++++++++++ " << std::endl;
                }


        for (char_nbr = 0; char_nbr < size; char_nbr++) {
            if (is_text) {
                printf ("%c", data [char_nbr]);
                                                                  //snprintf(cmd + strlen(cmd), (sizeof cmd) - strlen(cmd), "%c", data [char_nbr]);             
                                                                                                                                                                                                                                                                        
                                                }
            else {
                 printf ("%02X", (unsigned char) data [char_nbr]);
                                                                 //printf ("%d",   (unsigned char) data [char_nbr]);
                                                                 
                                                                //snprintf(cmd + strlen(cmd), (sizeof cmd) - strlen(cmd), "%d", (unsigned char)data [char_nbr]);
                                                                
                                                                //if (data[0] == '\x7') {
                                                                //      std::cout << " little endian " << std::endl;
                                                                //}
                                                                //else {
                                                                //      std::cout << "big endian " << std::endl;
                                                                //}
                                
                                                                
                                                }
        }//for
                printf ("\n");
                */
                                
                        
                                if (is_text) {
                                                parsingSuccessful=reader.parse(data, parsedFromString);
                                                if (parsingSuccessful) {
                                                        //Json::StyledWriter styledWriter;
                                                        //std::cout << "STYLED: --------------------------------" << std::endl;
                                                        //std::cout << styledWriter.write(parsedFromString) << std::endl;
                                                        //std::cout << "----------------------------------" << std::endl;
                                                        //std::cout << parsedFromString["htype"] << std::endl;
                                                        
                                                        
                                                        if (fastWriter.write(parsedFromString["htype"]).find("bsr_m-1.1") != std::string::npos) { 

                                                                ++nSequentialHeader;

                                                                hashIs=parsedFromString["hash"].asString();

                                                                if (hashOriginalFlag==0) {
                                                                        std::cout << hashIs << " is different to original/// " << hashOriginal << std::endl;    
                                                                        hashOriginal=hashIs;
                                                                        ++hashOriginalFlag;
                                                                        fillBSPV=true;
                                                                }

                                                                if (hashOriginal.compare(hashIs)!=0) {
                                                                        std::cout << hashIs << " is different to original " << hashOriginal << std::endl;       
                                                                        hashOriginal=hashIs;
                                                                        ++hashOriginalFlag;
                                                                        fillBSPV=true;
                                                                                                                                
                                                                }
                                                                 //std::cout << "p id " << parsedFromString["pulse_id"].asUInt64() << std::endl;

                                                                 dbpm.setPulse_id(parsedFromString["pulse_id"].asUInt64());

                                                                //Reset values as a change of hash signifies that data from two pulse ids
                                                                //is being sent in one zeromq messages

                                                                subMessage=0;
                                                                
                                                                bsPVIdx=-1;
                                                                dbpm.status=ICAFE_NORMAL;

                                                                nZeroSize=0;
                                                                
                                                                
                                                        /*
                                                        std::cout << "(1)++++++++++++++++++++++++++++++++++++++++MAIN++++++++++++++++++++++++++++++++++++++++++"        << std::endl;
                                                
                                                                std::cout << "hash " << parsedFromString["hash"] << std::endl;
                                                                std::cout << "p id " << parsedFromString["pulse_id"].asUInt64() << std::endl;
                                                                std::cout << "g ts " << parsedFromString["global_timestamp"] << std::endl;
                                                                std::cout << "comp " << parsedFromString["dh_compression"] << std::endl;
                                                        
                                                                std::cout << "sec " << parsedFromString["global_timestamp"]["sec"].asUInt() << std::endl;
                                                          std::cout << "nsec " << parsedFromString["global_timestamp"]["ns"].asUInt() << std::endl;
                                                        */
                                                                
                                                        }       
                                                        else if (fastWriter.write(parsedFromString["htype"]).find("bsr_d-1.1") != std::string::npos) { 
                                                        
                                                        ++nSequentialHeader;
                                                                
                                                         if  (fillBSPV) {
                                                        
                                                                        bsPV.clear();
                                                                        bsPV.reserve(dbpm.getNPV());
                                                                        /*
                                                                        if (dbpm.getNPV() != parsedFromString["channels"].size() ) {
                                                                                std::cout << "No of CONFIGURED BPMS: " << dbpm.getNPV() 
                                                                                     << " is diffent to that being channeled " <<  parsedFromString["channels"].size() << std::endl;
                                                                                bsPV.reserve( std::max( (size_t) parsedFromString["channels"].size(),dbpm.getNPV()) );   
                                                                                
                                                                        }
                                                                        
                                                                */
                
                                                                                                                                
                                                                //std::cout << "chan " << parsedFromString["channels"] << std::endl;
                                                                
                                                                //std::cout << "No of channels " << parsedFromString["channels"].size() << std::endl;

                                                                //std::cout << "(2)++++++++++++++++++++++++++++++++++++++++HEADER++++++++++++++++++++++++++++++++++++++++++"    << std::endl;
                                                                                                        
                                                                                for (Json::Value::ArrayIndex i=0; i < parsedFromString["channels"].size(); ++ i) {
                                                        
                                                                                        //std::cout << "name " << parsedFromString["channels"][i]["name"].asString() << std::endl;
                                                                                        //std::cout << "enco " << parsedFromString["channels"][i]["encoding"] << std::endl;
                                                                                //std::cout << "type " << parsedFromString["channels"][i]["type"] << std::endl;
                                                                         
                                                                                                bsPV.push_back( (parsedFromString["channels"][i]["name"]).asString());
                                                                                }
                                                                                /*
                                                                                std::cout << "LIST OF PVS " << std::endl;
                                                                                for (size_t i=0; i< bsPV.size(); ++i ) {                        
                                                                                        std::cout << i << " // " <<   bsPV[i].c_str() << " " << std::endl;
                                                                                }
                                                                                std::cout << std::endl;
                                                                                */
                                                                                
                                                                                //std::cout << "NEW FILL: size of bsPV " <<  bsPV.size() << std::endl;
                                                                                fillBSPV=false;
                                                                } //if fill

                                                        }       
                                                        
                                                        else if (fastWriter.write(parsedFromString["htype"]).find("bsr_reconnect-1.0") != std::string::npos) {
                                                        
                                                                                std::cout << "RECONNECT CONTROL MESSAGE " << std::endl;
                                                                                std::cout << "HAS NEW SOURCE ADDRESS: " <<  fastWriter.write(parsedFromString["address"])  << std::endl;
                                                                        
                                                        }
                                                        else if (fastWriter.write(parsedFromString["htype"]).find("bsr_stop-1.0") != std::string::npos) {
                                                        
                                                                                std::cout << "STOP CONTROL MESSAGE " << std::endl;
                                                                                std::cout << "RECEIVED: " <<  fastWriter.write(parsedFromString["htype"])  << std::endl;
                                                                                std::cout << "From BS Documentation: Message can be ignored as source will send from same address after startup" << std::endl;
                                                                                
                                                        }
                                                        
                                                        else {
                                                        
                                                          std::cout << "HEADER IS SOMETHING ELSE: " << std::endl;
                                                                                std::cout << parsedFromString["htype"] << std::endl;
                                                        
                                                                exit(1);
                                                        
                                                        }
                                                        
                                                        
                                                        //std::cout << "/----------------------------------/" << std::endl;
                                                                
                                        }
                                }
                                
                                
                                if (nSequentialHeader >3 && is_text==1) {
                                         std::cout << "WARNING: ZEROMQ SUB-MESSAGE DOES NOT CLOSE " << std::endl; 
                                         std::cout << "WARNING: FORCING ZMQ_MSG_CLOSE  " << std::endl;

                                         std::cout << "No of sequential headers " << nSequentialHeader << std::endl;
                                         std::cout << "is_text " << is_text << std::endl;
                                         //zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                                                                 
                                         //zmq_msg_close (&message);
                                         //std::cout << "message mulipart --> more == " << more << std::endl;
                                                

                                         //nSequentialHeader=0;
                                         //cannot break; results in seqmentation fault

                                         //if (!more) {
                                        //                 dbpm.status=ERRNO_EAGAIN; //ICAFE_ERRNO_BASE +zmq_errno() ;
                                        //                 std::cout << "EARLY BREAK subMessage total: " << subMessage << std::endl;
                                        //                 puts ("//------------------------------------END-------------------------------------//");

                                        // break;      //  Last message
                                        //}
                                
                                }

                                
                                
                                union foo
                                {
                                char c[sizeof(double)];
                                double d;
                        
                                } bar;
                                
                                //SIZE 16 is timestamp
                                //SIZE 8  is x,y,Q
                                //SIZE 2  is valid
                                
                                if (subMessage > 1430) {
                                  std::cout << " SIZE " << size << " subMessage " << subMessage << std::endl;
                                  std::cout << "size of bsPV " <<  bsPV.size() << std::endl;
                                }       
                                        
                                if (subMessage > 0 && subMessage%2 ==0) {
                                        if (size==8) {
                                                
                                                // big endian   
                                                for (char_nbr = 0; char_nbr < size; char_nbr++) {
                                                 bar.c[char_nbr]=data[size-1-char_nbr];  // THis works for big engian
                                                }
                                                // little endian        
                                                //for (char_nbr = 0; char_nbr < size; char_nbr++) {
                                                // bar.c[char_nbr]=data[char_nbr];  
                                                //}
                                                
                                                
                                                //std::cout << "UNION D " << bar.d << std::endl;
                                        
                                                
                                                double v; // = (double*) data;
                                                memcpy(&v, bar.c, sizeof(double));
                                                        //std::cout << " double val " << v << std::endl;
                                
                                        //This is BPM Data - value
                                        ++bsPVIdx;
                                         
                                        if (dbpm.getPVIdx(bsPV[bsPVIdx]) <0)  {
                                          std::cout << " WARNING--> THIS CHANNEL WAS NOT REQUESTED IN CONFIGURATION FILE " << std::endl;
                                                std::cout << " bsPV index = " <<  bsPVIdx << std::endl;
                                          std::cout << " pv from bs = " << bsPV[bsPVIdx] << std::endl;
                                          std::cout << " Illegal index Value =" << dbpm.getPVIdx(bsPV[bsPVIdx]) << std::endl;
                                                std::cout << " SKIPPING THIS BPM... " << std::endl;
                                                continue;
                                        }
                                        
                                        
                                        dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].set(v);

                                        dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].setStatus(ICAFE_NORMAL);

                                        //std::cout << " readback of value that was set = " << dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].val[0].d << std::endl;
                                                
                                        }
                                        else if (size==2) {
                                                unsigned short iv = 0;
                                                //memcpy(&iv, cmd, size);
                                                        //big endian
                                                        for (size_t n=0; n < size; n++) {
                                                                iv = (iv << 8) + data[n];
                                                        }
                                                        //little endian
                                                        //for (size_t n = size; n >= 0; n--) {
                                                        //      iv = (iv << 8) + data[n];
                                                        //}
                                        
                                        //This is BPM Data      - VALID/INVALID 
                                                        
                                                        //std::cout << "uint val (1 means valid) " << iv << std::endl;
                                                        
                                          /*
                                                std::copy(data, data + 32, reinterpret_cast<char *>(&i));
                                                std::cout << "uint16  val " << i << std::endl;
                                                std::copy(data, data + 8, reinterpret_cast<char *>(&i));
                                                std::cout << "uint16  val " << i << std::endl;
                                                std::copy(data, data + 4, reinterpret_cast<char *>(&i));
                                                std::cout << "uint16  val " << i << std::endl;
                                                std::copy(data, data + 2, reinterpret_cast<char *>(&i));
                                                std::cout << "uint16  val " << i << std::endl;
                                                std::copy(data, data + 1, reinterpret_cast<char *>(&i));
                                                std::cout << "uint16  val " << i << std::endl;
                                                
                                                */
                                                        ++bsPVIdx;
                                                
                                                         
                                        if (dbpm.getPVIdx(bsPV[bsPVIdx]) <0)  {
                                          std::cout << " WARNING--> THIS CHANNEL WAS NOT REQUESTED IN CONFIGURATION FILE " << std::endl;
                                                std::cout << " bsPV index = " <<  bsPVIdx << std::endl;
                                          std::cout << " pv from bs = " << bsPV[bsPVIdx] << std::endl;
                                          std::cout << " Illegal index Value =" << dbpm.getPVIdx(bsPV[bsPVIdx]) << std::endl;
                                                std::cout << " SKIPPING THIS BPM ENUM TYPE " << std::endl;
                                                continue;
                                        }
                                                
                                                                
                                                                
                                                                
                                                        if (iv==1) {
                                                                dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].set((std::string) "VALID");
                                                                
                                                        }       
                                                        else {
                                                                dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].set((std::string) "INVALID");
                                                        }
                                                        
                                                                //std::cout << "value DBPM = " << dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].val[0].str << std::endl;


                                                                dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].setStatus(ICAFE_NORMAL);
                                        }       
                                
                                        
                                }
                                else if (subMessage > 1 && subMessage%2 ==1) {
                                        //std::cout << "timestamp " << std::endl;
                                        unsigned int a=0,b=0;
                                        for (size_t n=0; n < size/2; n++) {
                                                                a = (a << 8) + data[n];
                                        }
                                        for (size_t n=size/2; n < size; n++) {
                                                                b = (b << 8) + data[n];
                                        }
                                        
                                        //std::cout << "a " << a << " b " << b << std::endl;
                                        
                                         
                                        if (dbpm.getPVIdx(bsPV[bsPVIdx]) <0)  {
                                          std::cout << " WARNING--> THIS CHANNEL WAS NOT REQUESTED IN CONFIGURATION FILE " << std::endl;
                                                std::cout << " bsPV index = " <<  bsPVIdx << std::endl;
                                          std::cout << " pv from bs = " << bsPV[bsPVIdx] << std::endl;
                                          std::cout << " Illegal index Value =" << dbpm.getPVIdx(bsPV[bsPVIdx]) << std::endl;
                                                std::cout << " SKIPPING THIS BPM TIMESTAMP " << std::endl;
                                                continue;
                                        }
                                        
                                        
                                        
                                        dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].ts.secPastEpoch=a;
                                        dbpm.pvd[dbpm.getPVIdx(bsPV[bsPVIdx])].ts.nsec=b;
                                }
                                


                                        //std::cout << "subMessage above: " << subMessage << std::endl;

                                        ++subMessage;

      
                } //ifelse
                                
      
        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);

        zmq_msg_close (&message);
                                
                //std::cout << "value of more " << more << std::endl;
                                
        if (!more) {
                                  dbpm.status=ICAFE_NORMAL;
                                 // std::cout << "subMessage total: " << subMessage << std::endl;
                                 // std::cout << " of which 0 size: " << nZeroSize << std::endl;
                                 // std::cout << " Percentage good: " << (subMessage-nZeroSize-2)*100/(subMessage-2);
                                 // puts ("//------------------------------------END-------------------------------------//");
                                         
            break;      //  Last message part
                                }
    } //while 1
                
                inDumpFlag=0;
        
                if (subMessage>2) {
                        dbpm.status=ICAFE_NORMAL;
                        //std::cout << "subMessage total: " << subMessage << std::endl;
                        //std::cout << " of which 0 size: " << nZeroSize << std::endl;
                        //std::cout << " Percentage good: " << (subMessage-nZeroSize-2)*100/(subMessage-2);
                        //puts ("//------------------------------------END-------------------------------------//");
                }


                //std::cout << "end of loop " << std::endl;
                
                return;
}


static void z_bsread (BSDataHolder &bsd){
#define __METHOD__  "z_bsread (BSDataHolder &bsd)"
        unsigned int localDebug=0;
        
        bsd.overallStatus=ICAFE_NORMAL;
        
        //Data type index
        if (!bsdtInsertFlag) {
                CAFE_BSHELPER::bsdtInsert();
                bsdtInsertFlag=true;
        } 
        bsreadContainer_set_by_name & name_index = CAFE_BSHELPER::bsdt.get<by_bsName> ();
        bsreadContainer_set_by_name::iterator it_name;
        unsigned int bsdtIndex;
         
        if(localDebug)puts ("//START----------------------------------------//\n");
        int subMessage=0;
        void * socket = bsd.subscriber;
        int64_t more =0;           //  Multipart detection
        size_t more_size = sizeof (more);
        //Keep a count of the number of MULTIpart messages with zero data
        int nZeroSize=0;
        
        unsigned long ts,tnsec;
                
        //Matching PV Index to BSChannel in bsd
        int bsPVIdx=-1;
        
        #if HAVE_JSON
        Json::Value parsedFromString;
        Json::Reader reader;    
        Json::FastWriter fastWriter;
        #endif
        bool parsingSuccessful; 
        bool mainHeaderReceived=false;  
        bool newHash=false;
        
        //subMessage is incremeneted at end of multi-part message
        
        //The Loop
        while (1) {

                //Process all parts of the message
                zmq_msg_t message;
                zmq_msg_init (&message);
                size_t size = zmq_msg_recv (&message, socket, 0);
                if(localDebug)printf ("[%03lu] \n", size);
                if (size == -1) {
                        std::cout << " Error is " << zmq_errno() << " " << zmq_strerror(zmq_errno()) << std::endl;
                        //Resource unavailable means that there is nothing to read now
                        zmq_msg_close (&message);
                        bsd.overallStatus=ICAFE_ERRNO_BASE+zmq_errno(); 
                        for (size_t i=0; i < bsd.getNPV(); ++i) {
                                if(bsd.getBSChannel(i).isBSEnabled()) {
                                        bsd.pvd[i].set(0);
                                        bsd.pvd[i].setStatus(bsd.overallStatus);
                                        bsd.pvd[i].setAlarmStatus(-1);
                                        bsd.pvd[i].setAlarmSeverity(-1);
                                        bsd.pvd[i].ts.secPastEpoch=0;
                                        bsd.pvd[i].ts.nsec        =0;
                                        bsd.pvd[i].setPulseID(0);
                                }
                        }       
                        break;
                } 
                else if (size == 0) {
                        if(localDebug)std::cout << " Data of Zero SIZE for submessage " << subMessage << std::endl;
                        ++nZeroSize;
                        //Avoid timestamp blob in count
                        if (nZeroSize%2==1) {
                                ++bsPVIdx;  //Increment when data
                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(0);
                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].setStatus(ECAFE_BSREAD_MULTIPART_MESS_NODATA);
                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].setPulseID(bsd.getPulse_id());
                        }
                        else { //Zero Timestamp
                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].ts.secPastEpoch=0;
                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].ts.nsec        =0;
                        }
                        if (bsd.overallStatus == ICAFE_NORMAL) {
                                bsd.overallStatus=ECAFE_BSREAD_MULTIPART_MESS_NODATA;
                        }
                } 
                else { 
                
                        //  Dump the message as text or binary
                        if(localDebug)printf ("[%03lu] \n", size);
                        
                        char *data = (char*)zmq_msg_data (&message);
                        
                        //Check if it is text and is so if it is the main header or not
                        
                        //First message is always the Main Header
                        
                        if (!mainHeaderReceived) {
                                //std::cout << "MAIN HEADER ================================================================================= " << std::endl;
                                parsingSuccessful=reader.parse(data, parsedFromString);
                                if (parsingSuccessful) {
                                        
                                        if (localDebug==1) {
                                                Json::StyledWriter styledWriter;
                                                std::cout << "STYLED: --------------------------------" << std::endl;
                                                std::cout << styledWriter.write(parsedFromString) << std::endl;
                                                std::cout << "----------------------------------" << std::endl;
                                                std::cout << "HEADER TYPE" << std::endl;
                                                std::cout << parsedFromString["htype"].asString() << std::endl;
                                        }
                                        
                                        if (fastWriter.write(parsedFromString["htype"]).find("bsr_m-1.1") != std::string::npos) { 
                                                
                                                if (localDebug) {
                                                        std::cout << "hash " << parsedFromString["hash"].asString() << std::endl;
                                                        std::cout << "p id " << parsedFromString["pulse_id"].asUInt64() << std::endl;
                                                        std::cout << "g ts " << parsedFromString["global_timestamp"] << std::endl;
                                                        std::cout << "comp " << parsedFromString["dh_compression"].asString() << std::endl;
                                                        std::cout << "sec  " << parsedFromString["global_timestamp"]["sec"].asUInt() << std::endl;
                                                        std::cout << "nsec " << parsedFromString["global_timestamp"]["ns"].asUInt() << std::endl;
                                                }
                                                //required
                                                bsd.setHtype( parsedFromString["htype"].asString() );
                                                
                                                if (bsd.getHash().compare(parsedFromString["hash"].asString()) !=0 ){
                                                        bsd.setHash( parsedFromString["hash"].asString() );
                                                        newHash=true;
                                                        //std::cout << " NEW HASHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHh " << std::endl;
                                                }
                                                bsd.setPulse_id(parsedFromString["pulse_id"].asUInt64());
                                                //optional
                                                if (!(parsedFromString["global_timestamp"]).empty()) {
                                                        bsd.setGlobal_timestamp(parsedFromString["global_timestamp"]["sec"].asUInt(),
                                                                                                                                                        parsedFromString["global_timestamp"]["ns"].asUInt());
                                                }
                                                //Check for Data Compression here!!
                                                if (!(parsedFromString["dh_compression"].asString()).empty()) {                                                         
                                                        bsd.setDH_compression(parsedFromString["dh_compression"].asString() );                                          
                                                } 
                                        }
                                        else if (fastWriter.write(parsedFromString["htype"]).find("bsr_reconnect-1.0") != std::string::npos) {
                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                std::cout << "RECONNECT CONTROL MESSAGE " << std::endl;
                                                std::cout << "HAS NEW SOURCE ADDRESS: " <<  fastWriter.write(parsedFromString["address"])  << std::endl;
                                                bsd.globalBSZmqStream = fastWriter.write(parsedFromString["address"]).c_str();
                                                dataHeaderReceived=false;
                                        }
                                        else if (fastWriter.write(parsedFromString["htype"]).find("bsr_stop-1.0") != std::string::npos) {
                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                std::cout << "STOP CONTROL MESSAGE " << std::endl;
                                                std::cout << "RECEIVED: " <<  fastWriter.write(parsedFromString["htype"])  << std::endl;
                                                std::cout << "From BS Documentation: Message can be ignored as source will send from same address after startup" << std::endl;
                                                dataHeaderReceived=false;
                                        }
                                        else {
                                                //Maybe this is not a header!??
                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                std::cout << "HEADER IS UNRECOGNIZED: SOMETHING OTHER THAN DOCUMENETED! " << std::endl;
                                                std::cout << parsedFromString["htype"] << std::endl;
                                                std::cout << "EXIT PROGRAM " << std::endl;
                                                dataHeaderReceived=false;
                                                exit(1);
                                        }
                                        mainHeaderReceived=true;
                                } //parsing successful
                                else {
                                        std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                        std::cout << "PARSING FAILED !! " << std::endl;
                                        std::cout << "CANNOT PROCEED WITHOUT MAIN HEADER INFORMATION! " << std::endl;
                                        bsd.overallStatus=ECAFE_BSREAD_PARSEFAIL_MAINHEADER;
                                        
                                        //Close message and break from loop
                                        ++subMessage;
                                                
                                        std::cout << "SUBMESSAGE " << subMessage << std::endl;
                                        more=0;           //  Multipart detection
                                        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                                        zmq_msg_close (&message);
                                        break;
                                }
                        } //ifMainHeaderReceived
                        
                        
                        else if ( (mainHeaderReceived && !dataHeaderReceived) || (newHash && dataHeaderReceived)) {
                                //std::cout << "DATA HEADER ================================================================================= " << std::endl;
                                //std::cout << "mh = " << mainHeaderReceived << " dh " << dataHeaderReceived << " nh " << newHash  <<std::endl;
                                newHash=false;
                                //Is it compressed or not??
                                if (bsd.getDH_compression().compare("bitshuffle_lz4")==0) {
                                        /*
                                        int uncompressedSize=0; int blockSize=0;
                                        unfoldPreBlob(data, uncompressedSize, blockSize);
                                        //decompress
                                        int src_size=uncompressedSize/sizeof(char);
                                        //use bitshuffle terminology
                                        char* regen_buffer = (char *) malloc(src_size*sizeof(char));
                                        if (regen_buffer == NULL) {
                                                bsd.overallStatus=ECAFE_BITSHUFF_ALLOCMEM;
                                                return;
                                        }
                                        
                                        int sizeBlob = size-BSREAD_PREBLOB_BYTES;
                                        if (sizeBlob <1) {sizeBlob=size;} //Should not happen
                                        char * dataBlob = new char[sizeBlob];
                                        int ij=0; 
                                        for (int char_nbr = BSREAD_PREBLOB_BYTES; char_nbr < size; char_nbr++) {
                                                dataBlob[ij]=data[char_nbr];
                                                ++ij;
                                        }
                                        //ECAFE_BITSHUFF_DECOMPRESS
                                        //ECAFE_BITSHUFF_REALLOCMEM
                                        const int decompressed_size = bshuf_decompress_lz4((const char *) dataBlob, regen_buffer, uncompressedSize, sizeof(char), 0);
                                        delete [] dataBlob;
                                        if(localDebug) std::cout << " " << " decompressed_size " << decompressed_size << std::endl;
                                        if (decompressed_size < 0) {bsd.overallStatus=ECAFE_BITSHUFF_DECOMPRESS; return;}
                                        else if (decompressed_size == 0){    
                                                if(localDebug)std::cout << "0 decompressed size! Should have a positive number for successful deompression?" <<std::endl;
                                        }
                                        else {    
                                                if(localDebug)std::cout << "We successfully decompressed some data!" <<std::endl;
                                        }
                                        // Not only does a positive return value mean success,
                                        // value returned == number of bytes regenerated from compressed_data stream.
                                        // We can use this to realloc() *regen_buffer to free up memory 
                                        //Do this to remove spurious trailing characters
                                        regen_buffer = (char *)realloc(regen_buffer, decompressed_size);
                                        if (regen_buffer == NULL) {
                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                std::cout << "WARNING with status code: " << ECAFE_BITSHUFF_REALLOCMEM << std::endl;
                                                std::cout << "Failed to reallocate memory for compressed data. Not a show stopper" << std::endl;
                                        }
                                        */
                                        char * regen_buffer = NULL;
                                        int status=CAFE_BSHELPER::bitshuffleDecompress(data, regen_buffer, size, sizeof(char));
                                        if (status==ICAFE_NORMAL) { 
                                                parsingSuccessful=reader.parse((const char *) regen_buffer, parsedFromString);
                                        } else 
                                        { 
                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                std::cout << "Data Header Compression Algorithmn: " << bsd.getDH_compression() << " FAILED! " << std:: endl;
                                                std::cout << "With Status Code: " << status << std:: endl;
                                                std::cout << "Discarding remainder of multi-part message! " << status << std:: endl;
                                                bsd.overallStatus=status;
                                                free(regen_buffer);
                                                
                                                //Close message and break from loop
                                                ++subMessage;
                                                
                                                std::cout << "SUBMESSAGE " << subMessage << std::endl;
                                                more=0;           //  Multipart detection
                                                zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                                                zmq_msg_close (&message);
                                                break;
                                        }
                                        
                                } //bitshuffle_lz4
                                else if (bsd.getDH_compression()=="none") {
                                        parsingSuccessful=reader.parse(data, parsedFromString);
                                }
                                else {
                                        std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                        std::cout << "Data Header Compression Algorithmn: " << bsd.getDH_compression() << " not recognized " << std:: endl;
                                        std::cout << "Assuming no compression for data header " << std::endl;
                                        parsingSuccessful=reader.parse(data, parsedFromString);
                                }
                                
                                if (parsingSuccessful) {
                                
                                        if (localDebug==1) {
                                                Json::StyledWriter styledWriter;
                                                std::cout << "STYLED: --------------------------------" << std::endl;
                                                std::cout << styledWriter.write(parsedFromString) << std::endl;
                                                std::cout << "----------------------------------" << std::endl;
                                                std::cout << "HEADER TYPE" << std::endl;
                                                std::cout << parsedFromString["htype"].asString() << std::endl;
                                        }
                                        
                                        if (fastWriter.write(parsedFromString["htype"]).find("bsr_d-1.1") != std::string::npos) { 
                                        
                                                if (localDebug) std::cout << "chan " << parsedFromString["channels"] << std::endl;
                                                
                                                bsrdV.clear();
                                                bsrdV.reserve(parsedFromString["channels"].size());
                                                
                                                for (Json::Value::ArrayIndex i=0; i < parsedFromString["channels"].size(); ++ i) {
                                                        
                                                        //New line is spit out after parsedFromString[]
                                                        //std::cout << "name:" << parsedFromString["channels"][i]["name"].asString()     << " //"<< std::endl;
                                                        //std::cout << "enco:" << parsedFromString["channels"][i]["encoding"].asString() << " //"<< std::endl;
                                                        //std::cout << "type:" << parsedFromString["channels"][i]["type"].asString()     << " //" <<std::endl;
                                                        //std::cout << "shape isArray " << parsedFromString["channels"][i]["shape"].isArray() << std::endl;
                                                        Json::Value iv;
                                                        Json::Value::ArrayIndex inpv=0;
                                                        
                                                        //std::cout << "Value as int " << parsedFromString["channels"][i]["shape"][inpv].asDouble() << std::endl;
                                                        //std::cout << "SIZE " << parsedFromString["channels"][i]["shape"].size() << std::endl;
                                                        //std::cout << "COMP " << parsedFromString["channels"][i]["compression"].asString() << std::endl;
                                                        std::vector<int> lshape;
                                                        for (Json::Value::ArrayIndex il=0; il < parsedFromString["channels"][i]["shape"].size(); ++ il) {   
                                                                //std::cout << "Value as int// " << parsedFromString["channels"][i]["shape"][il].asInt() << std::endl;
                                                                lshape.push_back(std::max(1, parsedFromString["channels"][i]["shape"][il].asInt()) ); 
                                                        }
                                                        
                                                        std::string NAME = parsedFromString["channels"][i]["name"].asString();
                                                        std::string ENCO = parsedFromString["channels"][i]["encoding"].asString(); // big default=little  
                                                        if (ENCO.empty()) {ENCO="little"; } //default
                                                        std::string TYPE = parsedFromString["channels"][i]["type"].asString();
                                                        if (TYPE.empty()) {TYPE="float64"; } //default
                                                        std::string COMP = parsedFromString["channels"][i]["compression"].asString();
                                                        if (COMP.empty()) {COMP="none"; } //default
                                                        
                                                        //std::cout << "TYPE " << TYPE << std::endl;
                                                        //std::cout << "TYPE: SIZE= " << TYPE.size() << std::endl; 
                                                        //std::cout << TYPE.substr(0,TYPE.size()) << std::endl;
                                                        
                                                        BSChannel bsc(NAME);
                                                        
                                                        //Need to find correct entry
                                                        //int idx= bsd.getIdxFromName(NAME);                                                    
                                                        //BSChannel bsc=  bsd.getBSChannel(idx); // instead of i
                                                        
                                                        bsc.setShape(lshape);
                                                        
                                                        //Need to fill in Channel Name, encoding and Type...
                                                        bsc.setType(TYPE);
                                                        bsc.setEncoding(ENCO);
                                                        bsc.setCompression(COMP);
                                                        
                                                        bsrdV.push_back(bsc);
                                                        
                                                        if ( bsd.getPVIdx(NAME) < 0 ) {
                                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                                std::cout << " WARNING--> THIS CHANNEL WAS NOT REQUESTED IN CONFIGURATION FILE " << std::endl;                                                  
                                                                std::cout << " pv requested = " << NAME << std::endl;
                                                                std::cout << " Illegal index Value =" << bsd.getPVIdx(NAME) << std::endl;
                                                                std::cout << " NO FURTHER ACTION REQUIRES " << std::endl;                                                               
                                                        }
                                                        else {
                                                                //Ensure there is sufficent space to hold returned data,, e.g. may be an array
                                                                bsd.pvd[ bsd.getPVIdx(NAME)].setNelem(bsc.getNelem());
                                                        }
                                                        int idx= bsd.getIdxFromName(NAME);
                                                        bsd.setBSChannel(idx, bsc);
                                                        
                                                        //std::cout << "The No elements are " << bsd.getBSChannel(i).getNelem() << std::endl;                                                   
                                                        //There may be more channels coming through this stream than requested
                                                        //therefore first fill a local vector<BSChannel>
                                                        //and then map this to that in BSDataHolder!
                                                        //It is conceivable that they do not match exactly
                                                
                                                } // for JsonValue
                                                
                                                if (localDebug) {
                                                        for (int i=0; i< bsrdV.size(); ++i) {
                                                                std::cout << i << " " <<  bsrdV[i].getName() << " " << bsd.getPVIdx( bsrdV[i].getName()) << std::endl;
                                                        }
                                                        std::cout << "No of channels " << bsrdV.size() << " " << parsedFromString["channels"].size() << std::endl;
                                                        std::cout << "----------------------------------" << std::endl;
                                                }
                                        } //bsr_d
                                  dataHeaderReceived=true; //set FLAG   
                                } //parsingsuccessful
                                else {
                                        std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                        std::cout << "PARSING FAILED !! " << std::endl;
                                        std::cout << "CANNOT PROCEED WITHOUT DATA HEADER INFORMATION! " << std::endl;
                                        bsd.overallStatus=ECAFE_BSREAD_PARSEFAIL_DATAHEADER;
                                        
                                        //Close message and break from loop
                                        ++subMessage;
                                                
                                        std::cout << "SUBMESSAGE " << subMessage << std::endl;
                                        more=0;           //  Multipart detection
                                        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                                        zmq_msg_close (&message);
                                        break;
                                }
                        
                        } //dataHeader
                        
                        else if (subMessage==1) {
                                //header again so we skip this as hash is the same!
                                //Do nothing
                                if(localDebug) std::cout << "HEADER INFO ALREADY REGISTERED " << std::endl;
                        }
                        else if (subMessage%2 ==1) 
                        {
                                //std::cout << "timestamp " << std::endl;
                                
                                if (bsd.getPVIdx(bsrdV[bsPVIdx].getName()) <0)  {
                                        std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                        std::cout << " WARNING--> THIS CHANNEL WAS NOT REQUESTED IN CONFIGURATION FILE " << std::endl;
                                        std::cout << " bsPV index = " <<  bsPVIdx << std::endl;
                                        std::cout << " pv from bs = " << bsrdV[bsPVIdx].getName() << std::endl;
                                        std::cout << " Expect illegal BSDataHolder index value =" << bsd.getPVIdx(bsrdV[bsPVIdx].getName()) << std::endl;
                                        std::cout << " SKIPPING THIS SUBMESSAGE... " << subMessage << std::endl;
                                }
                                else {
                                        ts=0; tnsec=0;
                                        //long test=1518777482;
                                        //printBits(sizeof(test), &test);
                                  //printBits(size, &data);
                                        //size==16
                                        /*
                                        std::cout << "SIZE OF TS " << size << std::endl;                                
                                  for (size_t n=size/2; n < size; n++) {
                                          char jh = data[n];            
                                                printBits(sizeof(jh), &jh);
                                        }
                                        std::cout << "SIZE OF TS " << size << std::endl;
                                        */
                                        /*(=
                                        char * data2 =new char[size/2];
                                        for (size_t n=0; n < size/2; n++) {
                                                data2[n]=data[n];
                                        }
                                        std::cout << "LLLONG " << bsetLongLong.unfoldScalar(data2, "big") << std::endl;
                                        delete [] data2;
                                        */
                                        for (size_t n=0; n < size/2; n++) {
                                          char jh = data[n];            
                                                //printBits(sizeof(jh), &jh);
                                                if ( (n+1) < size/2 ) {
                                                if ( data[n+1] & 0x80) {
                                                        data[n]=data[n]+1;
                                                }
                                                }
                                                //if (n==4) data[n]=data[n]+1; //Just what is the compiler playing at when shifting bits!??
                                                //if (n==5) data[n]=data[n]+1; //ditto
                                                ts = (ts<<8) + data[n];                                                                         
                                        }
                                        
                                        //printBits(sizeof(ts), &ts);
                                        //if(localDebug)std::cout << "ts original way " << ts  << std::endl;
                                  //unsigned short idx=0;
                                        //int ts_int=( ( (data[idx+7]<<0) & 0x7f) | ((data[idx+7]<<0) & 0x80)  |  (data[idx+6]<<8) | (data[idx+5]<<16) |  (data[idx+4]<<24) 
                                        //int ts_int=(  (data[idx+7]<<0)   |  (data[idx+6]<<8) | (data[idx+5]<<16) |  (data[idx+4]<<24) 
                                        //      | ((long long) data[idx+3]<<32) | ((long long) data[idx+2]<<40) | ((long long) data[idx+1]<<48) | ((long long) data[idx]<<56) ) ;
                                                
                                        //std::cout << "ts_int= " << ts_int << std:: endl;
                                        //idx=4;
                                        //long long ts_long=( ( (data[idx+3]<<0) ) | ((data[idx+3]<<0) & 0x80)  |  ( (data[idx+2]<<8) ) | ((data[idx+2]<<8) & 0x80)  |  
                                        //                                                                              ( (data[idx+1]<<16) ) | ((data[idx+1]<<16) & 0x80)  |  ( (data[idx]<<24) ) | ((data[idx]<<24) & 0x80) );  
                        
                                                
                                        //std::cout << "ts_long= " << ts_long << std:: endl;
                                        
                                        
                                        
                                  for (size_t n=size/2; n < size; n++) {
                                          char jh = data[n];            
                                                //printBits(sizeof(jh), &jh);
                                                //if (n==12) data[n]=data[n]+1;
                                                if ( (n+1) < size ) {
                                                if ( data[n+1] & 0x80 ) {
                                                        data[n]=data[n]+1;
                                                        //std::cout << "------------" << std::endl;
                                                        //jh = data[n]; 
                                                        //printBits(sizeof(jh), &jh);
                                                        //std::cout << "------------" << std::endl;
                                                }
                                                }
                                                tnsec = (tnsec<<8) + data[n];
                                        }
                                          //printBits(sizeof(tnsec), &tnsec);
                                        //if(localDebug)std::cout << "tns original way " << tnsec  << std::endl;
                                        
                                        //idx=8;
                                        //tnsec=( ( (data[idx+7]<<0) & 0x7f) | ((data[idx+7]<<0) & 0x80)  |  (data[idx+6]<<8) | (data[idx+5]<<16) |  (data[idx+4]<<24) 
                                        //      | ((long long) data[idx+3]<<32) | ((long long) data[idx+2]<<40) | ((long long) data[idx+1]<<48) | ((long long) data[idx]<<56) ) ;
                                        
                                        
                                        if(localDebug)std::cout << "ts " << ts << " tnsec " << tnsec << std::endl;
                                        //Add to bsd
                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].ts.secPastEpoch=ts;
                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].ts.nsec        =tnsec;
                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].setBeamEventNo(bsd.getPulse_id());
                                }
                        }
                        else {
                                //std::cout << " NOW COMES THE DATA " << std::endl;
                                
                                //if (localDebug) {
                                //      for (int i=0; i< bsrdV.size(); ++i) {
                                //              std::cout << i << " " <<  bsrdV[i].getName() << " " << bsd.getPVIdx(bsrdV[i].getName()) << std::endl;
                                //      }
                                //      std::cout << "No of channels " << bsrdV.size()  << std::endl;
                                //      std::cout << "-----------------------" << std::endl;
                                //}
                                
                                //Now comes the data
                                ++bsPVIdx;
                                
                                if (bsd.getPVIdx(bsrdV[bsPVIdx].getName()) <0)  {
                                        
                                        std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                        std::cout << " WARNING--> THIS CHANNEL WAS NOT REQUESTED IN CONFIGURATION FILE " << std::endl;
                                        std::cout << " bsPV index = " <<  bsPVIdx << std::endl;
                                        std::cout << " pv from bs = " << bsrdV[bsPVIdx].getName() << std::endl;
                                        std::cout << " Expect illegal BSDataHolder index value =" << bsd.getPVIdx(bsrdV[bsPVIdx].getName()) << std::endl;
                                        std::cout << " SKIPPING THIS SUBMESSAGE... " << subMessage << std::endl;
                                        
                                        more=0;           //  Multipart detection
                                        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                                        zmq_msg_close (&message);
                                        ++subMessage;
                                        if (!more) {
                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                std::cout << " No more submessages! " << std::endl;
                                                std::cout << " This message implies that pv was not accompanied by a timestamp!! Something not quite right. " << std::endl;  
                                                break;      //  Last message part
                                        }
                                        continue; //top of while
                                }
                                
                                if(localDebug) {
                                        std::cout << "index= " << bsPVIdx <<  std::endl;
                                        std::cout << "Name = " << bsrdV[bsPVIdx].getName() <<  " " << bsrdV[bsPVIdx].getType() << std::endl;
                                        std::cout << "IDx  = " << bsd.getPVIdx(bsrdV[bsPVIdx].getName()) << std::endl;
                                }
                                
                                //Find enum of data type
                                it_name = name_index.find(bsrdV[bsPVIdx].getType());
                                if ( (it_name) != name_index.end() ) {
                                        bsdtIndex=(*it_name).by_bsID;
                                }
                                //Do we need to compress the data?
                                bool compressedFlag=false;
                                BSChannel bsc=bsd.getBSChannel(bsrdV[bsPVIdx].getName());
                                char * regen_buffer;
                                
                                if ( bsc.getCompression().compare("bitshuffle_lz4")==0)  {
                                
                                        //Decompress
                                        
                                        int status=CAFE_BSHELPER::bitshuffleDecompress(data, regen_buffer, size, getByteSize(bsdtIndex));
                                        if (status==ICAFE_NORMAL) { 
                                                parsingSuccessful=reader.parse((const char *) regen_buffer, parsedFromString);                                  
                                        }       else 
                                        { 
                                                free(regen_buffer);
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].setStatus(status);
                                                bsd.overallStatus=status;
                                                //skip message
                                                more=0;           //  Multipart detection
                                                zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                                                zmq_msg_close (&message);
                                                ++subMessage;
                                                if (!more) {
                                                        std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                        std::cout << " No more submessages! " << std::endl;
                                                        std::cout << " This message implies that pv was not accompanied by a timestamp!! Something not quite right. " << std::endl;  
                                                        break;      //  Last message part
                                                }
                                                continue; //top of while
                                        }
                                        compressedFlag=true;
                                        
                                }
                                else {
                                        //No compression required
                                        regen_buffer = data; //new char[size];
                                        //regen_buffer = new char[size];
                                        //for (int i=0; i<size; ++i) {
                                                //regen_buffer[i]=data[i];
                                        //}
                                        //std::cout << size << " /// " << bsrdV[bsPVIdx].getNelem()<< std::endl;
                                }
                                
                                //std::cout << "Compression//0//--> " << bsd.getBSChannel(0).getCompression() << std::endl;
                                //std::cout << "Compression//1//--> " << bsd.getBSChannel(1).getCompression() << std::endl;
                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].setStatus(ICAFE_NORMAL);
                                
                                //char,bool=1, short=2, int =4, long,long long=8
                                
                                switch (bsdtIndex) {
                                        case CAFE_BSHELPER::BS_STRING:
                                                {
                                                if (bsrdV[bsPVIdx].getNelem()>1) { 
                                                        std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                        std::cout << "An array of strings has not been envisioned for bsREAD" << std::endl;
                                                        std::cout << "Assuming one string element of max size PVNAME_SIZE" << std::endl;
                                                }
                                                if (size > MAX_STRING_SIZE) {
                                                        std::cout << "Datatype is string. Size " << size << " too large; trimming to MAX_STRING_SIZE" << std::endl;
                                                }
                                                
                                                std::string str="";;
                                                for (int i=0; i<std::min((int)size, (int)MAX_STRING_SIZE); ++i) {
                                                        if (data[i] != '\0') {
                                                                str.append(1, data[i] );
                                                                //std::cout << str << " [" << i << "] " << std::endl;
                                                        }
                                                }       
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(str);
                                                if(localDebug)std::cout << " readback of string value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].str << std::endl;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_FLOAT32:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetFloat.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of float scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].f << std::endl;
                                                        break;
                                                }                                                       
                                                //std::vector<float> Vbs( bsrdV[bsPVIdx].getNelem()); //Must allocate memory here
                                                //bsetFloatV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);                 
                                                //for (int ik=0; ik<Vbs.size(); ++ik) std::cout << Vbs[ik] << " [" << ik << "] " << std::endl;
                                                float * ff = bsetFloat.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(ff); 
                                                if(localDebug)std::cout << " readback of float value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].f << std::endl;
                                                delete [] ff;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_FLOAT64:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetDouble.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of double scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].d << std::endl;
                                                        break;
                                                }
                                                //std::vector<double> Vbs( bsrdV[bsPVIdx].getNelem()); //Must allocate memory here
                                                //bsetDoubleV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                double * dd =  bsetDouble.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(dd); 
                                                if(localDebug)std::cout << " readback of double value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].d << std::endl;
                                                delete [] dd;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_INT64:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetLongLong.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of uint64 scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsLongLong()  << std::endl;
                                                        break;
                                                }
                                                //std::vector<long long> Vbs; 
                                                //bsetLongLongV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                long long * ll=bsetLongLong.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());                                             
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(ll); 
                                                if(localDebug) {
                                                        std::cout << " readback of int64 value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsLongLong() << std::endl;
                                                        //for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                        //      std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsLongLong(i) << " [" << i << "] " ;
                                                        //}
                                                        //std::cout << std::endl;
                                                }
                                                delete [] ll;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_UINT64: 
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetULongLong.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of uint64 scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsULongLong()  << std::endl;
                                                        break;
                                                }
                                                //std::vector<unsigned long long> Vbs; 
                                                //bsetULongLongV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                unsigned long long * ull=bsetULongLong.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());                                          
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(ull); 
                                                if(localDebug) {
                                                        std::cout << " readback of uint64 value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsULongLong() << std::endl;
                                                        
                                                        //for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                        //      std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsULongLong() << " [" << i << "] " ;
                                                        //}
                                                        std::cout << std::endl;
                                                }
                                                delete [] ull;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_INT32:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetInt.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of int32 scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].l  << std::endl;
                                                        break;
                                                }
                                                //std::vector<int> Vbs; 
                                                //bsetIntV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                int * l=bsetInt.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());                                         
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(l); 
                                                if(localDebug) {
                                                        std::cout << " readback of int32 value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].l << std::endl;
                                                        for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                                std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[i].l << " [" << i << "] " ;
                                                        }
                                                        std::cout << std::endl;
                                                }
                                                delete [] l;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_UINT32:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetUInt.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of uint32 scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsULong()  << std::endl;
                                                        break;
                                                }
                                                //std::vector<unsigned int> Vbs; 
                                                //bsetUIntV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                unsigned int * ul=bsetUInt.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());                                              
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(ul); 
                                                if(localDebug) {
                                                        std::cout << " readback of uint32 value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsULong() << std::endl;
                                                        //for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                        //      std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getAsULong(i) << " [" << i << "] " ;
                                                        //}
                                                        //std::cout << std::endl;
                                                }
                                                delete [] ul;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_INT16:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetShort.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of short scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].s << std::endl;
                                                        break;
                                                }
                                                //std::vector<short> Vbs; 
                                                //bsetShortV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                
                                                short * s=bsetShort.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());
                                                //std::cout <<  "SHORT VALUE " << s[0] << std::endl;
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(s); 
                                                if(localDebug) {
                                                        std::cout << " readback of short value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].s << std::endl;
                                                        //std::cout << " nelements in pvdata " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].getNelem() << std::endl;
                                                        //for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                        //      std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[i].s << " [" << i << "] " ;
                                                        //}
                                                        //std::cout << std::endl;
                                                }       
                                                delete [] s;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_UINT16:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetUShort.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of ushort scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].us << std::endl;
                                                        break;
                                                }
                                                //std::vector<unsigned short> Vbs; 
                                                //bsetUShortV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                unsigned short * us=bsetUShort.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(us); 
                                                if(localDebug) {
                                                        std::cout << " readback of ushort value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].us << std::endl;
                                                        //for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                        //      std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[i].us << " [" << i << "] " ;
                                                        //}
                                                        //std::cout << std::endl;
                                                }       
                                                delete [] us;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_INT8:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetChar.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of char scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].ch << std::endl;
                                                        break;
                                                }
                                                //std::vector<char> Vbs; 
                                                //bsetCharV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                char * ch=bsetChar.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());                                              
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(ch); //set(ch); 
                                                if(localDebug) {
                                                        std::cout << " readback of char value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].ch << std::endl;
                                                        //for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                        //      std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[i].ch << " [" << i << "] " ;
                                                        //}
                                                        //std::cout << std::endl;
                                                }
                                                delete [] ch;
                                                break;
                                                }
                                        case CAFE_BSHELPER::BS_UINT8:
                                        case CAFE_BSHELPER::BS_BOOL:
                                                {
                                                if ( bsrdV[bsPVIdx].getNelem()==1) {
                                                        bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set( bsetUChar.unfoldScalar(regen_buffer, bsrdV[bsPVIdx].getEncoding()) );
                                                        if(localDebug)std::cout << " readback of unsigned char scalar that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].ch << std::endl;
                                                        break;
                                                }
                                                //std::vector<unsigned char> Vbs; 
                                                //bsetUCharV.unfold(regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding(),Vbs);
                                                unsigned char * uch=bsetUChar.unfold (regen_buffer, bsrdV[bsPVIdx].getNelem(), bsrdV[bsPVIdx].getEncoding());   
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].set(uch); 
                                                if(localDebug) {
                                                        std::cout << " readback of unsigned char value that was set = " << bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[0].ch << std::endl;
                                                        //for (int i=0; i < bsrdV[bsPVIdx].getNelem(); ++i) {
                                                        //      std::cout <<  bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].val[i].ch << " [" << i << "] " ;
                                                        //}
                                                        //std::cout << std::endl;
                                                }       
                                                delete [] uch;
                                                break;
                                                }
                                        default:
                                                std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
                                                std::cout << "ECAFE_INVALID_SWITCH_CASE" << std::endl;
                                                print_out_by<by_bsID>(bsdt);
                                                bsd.pvd[bsd.getPVIdx(bsrdV[bsPVIdx].getName())].setStatus(ECAFE_INVALID_SWITCH_CASE);
                                }
                                if (compressedFlag) {free(regen_buffer);}
                        }
                } //top else
                
                
                // THIS HAS TO BE COPIED TO BEFORE CONTINUE!
                
                //Continue goes to top of while
                //Why not add submessage here
                ++subMessage;
                
                //std::cout << "SUBMESSAGE " << subMessage << std::endl;
                more=0;           //  Multipart detection
                zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
                zmq_msg_close (&message);
                if (!more) {
                        if(localDebug)puts ("//END----------------------------------------//");
                        break;      //  Last message part
                }
        
        } //while 1
        

//First two messages are main header and data header
        if (subMessage>1) {
                bsd.overallStatus=ICAFE_NORMAL;
                bsd.setNChannels((subMessage-2)/2);
                bsd.setNNullData(nZeroSize/2);
                bsd.setPGoodData((subMessage-nZeroSize-2)*100/(std::max((subMessage-2),1)) );
                if(localDebug) {
                        std::cout << " No of submessages: " << subMessage << std::endl;
                        std::cout << " of which 0 size have: " << nZeroSize << std::endl;
                        std::cout << " No of channels: " << bsd.getNChannels()<< std::endl;
                        std::cout << " of which 0 size have: " << bsd.getNNullData() << std::endl;
                        std::cout << " Percentage good: " << bsd.getPGoodData() <<std::endl;
                        
                        puts ("//------------------------------------END-------------------------------------//");
                        
                }
        }
        
        return;
        
                        
#undef __METHOD__
}

#endif  //  ZBSREAD_H