c_cpp/aio-test/nsync.cpp

   1 /*
   2  * nsync.cpp
   3  *
   4  *  Created on: Jul 5, 2010
   5  *      Author: Aqua
   6  */
   7
   8 #include"nsync.h"
   9
  10 using namespace std;
  11 using namespace boost;
  12
  13 namespace bp = ::boost::process;
  14 namespace ba = ::boost::asio;
  15
  16 #define BUFFER_SIZE 0xFFFFFF
  17
  18 int ImportFileList(nSyncBuf_s & buf, char * fn)
  19 {
  20         char errStr[4096];
  21         int count(0);
  22
  23         if(access(fn, R_OK) == -1)
  24         {
  25                 snprintf(errStr, 4095, "[%s][%s][%d]: File->%s", __FILE__, __FUNCTION__, __LINE__, fn);
  26                 perror(errStr);
  27                 return -1;
  28         }
  29         ifstream I(fn);
  30         string strTmp;
  31         while(true)
  32         {
  33                 getline(I, strTmp, '\n');
  34                 if(!I)
  35                         break;
  36                 if(access(strTmp.c_str(), R_OK) == -1)
  37                 {
  38                         snprintf(errStr, 4095, "[%s][%s][%d]: File->%s", __FILE__, __FUNCTION__, __LINE__, strTmp.c_str());
  39                         perror(errStr);
  40                         continue;
  41                 }
  42                 buf.fnList.push_back(strTmp);
  43                 ++count;
  44         }
  45
  46         return count;
  47 }
  48
  49 int InsertFile(nSyncBuf_s & buf, char * fn)
  50 {
  51         char errStr[4096];
  52
  53         if(access(fn, R_OK) == -1)
  54         {
  55                 snprintf(errStr, 4095, "[%s][%s][%d]: File->%s", __FILE__, __FUNCTION__, __LINE__, fn);
  56                 perror(errStr);
  57                 return -1;
  58         }
  59         else
  60                 buf.fnList.push_back(fn);
  61         return 1;
  62 }
  63
  64 bp::child StartChild(nSyncBuf_s & buf)
  65 {
  66         string exec = bp::find_executable_in_path("zcat");
  67
  68         vector<string> args;
  69         args.push_back(bp::find_executable_in_path("zcat"));
  70         args.push_back("-f");
  71
  72         for(uint i(0); i < buf.fnList.size(); ++i)
  73         {
  74                 args.push_back(buf.fnList[i]);
  75         }
  76
  77         bp::context ctx;
  78         ctx.stdout_behavior = bp::capture_stream();
  79         ctx.stderr_behavior = bp::inherit_stream();
  80         ctx.environment = bp::self::get_environment();
  81
  82         return bp::launch(exec, args, ctx);
  83 }
  84
  85 void endRead(nSyncBuf_s &, const boost::system::error_code & ec, size_t bytes_transferred);
  86
  87 void nSyncRead(nSyncBuf_s & buf)
  88 {
  89         buf.in.async_read_some(boost::asio::buffer(buf.buffer, BUFFER_SIZE),
  90                 boost::bind(&endRead, boost::ref(buf), ba::placeholders::error, ba::placeholders::bytes_transferred));
  91 }
  92
  93 void endRead(nSyncBuf_s & buf, const boost::system::error_code & ec, size_t bytes_transferred)
  94 {
  95         if(!ec)
  96         {
  97                 if(buf.atomEnabler)
  98                         buf.lock();
  99                 buf.Routine(buf, bytes_transferred);
 100                 nSyncRead(buf);
 101         }
 102 }
 103
 104 void nSync(nSyncBuf_s & buf)
 105 {
 106         if(buf.fnList.empty())
 107         {
 108                 perror("Empty filename vector. Program Terminated");
 109                 exit(EXIT_FAILURE);
 110         }
 111         bp::child c = StartChild(buf);
 112
 113         buf.in.assign(c.get_stdout().handle().release());
 114
 115         nSyncRead(buf);
 116
 117         buf.io_service.run();
 118
 119         c.wait();
 120 }
 121
 122 nSyncBuf_s::nSyncBuf_s(_ConsumeBufferRoutine_f func, bool atomic) : atomEnabler(atomic), Routine(func), in(io_service)
 123 {
 124         buffer = new char[BUFFER_SIZE];
 125         if(!buffer)
 126         {
 127                 perror("[nSyncBuf_s]: Buffer allocation error.");
 128                 throw bad_alloc();
 129         }
 130 }
 131
 132 nSyncBuf_s::~nSyncBuf_s()
 133 {
 134         if(buffer)
 135         {
 136                 delete[] buffer;
 137         }
 138         else
 139         {
 140                 perror("[nSyncBuf_s]: Null ptr of buffer when destruction.");
 141                 throw bad_alloc();
 142         }
 143 }
 144
 145 inline void nSyncBuf_s::lock()
 146 {
 147         atom.lock();
 148 }
 149
 150 inline void nSyncBuf_s::unlock()
 151 {
 152         atom.unlock();
 153 }
 154
 155 void nSyncBuf_s::release()
 156 {
 157         atom.unlock();
 158 }
 159
 160 nSyncBufAry_t nSyncBuf_s::ptr()
 161 {
 162         return buffer;
 163 }
 164
 165 nSyncBufAry_t nSyncBuf_s::operator *()
 166 {
 167         return buffer;
 168 }
 169