RTBKit: soa/logger/ThriftFlumeEventServer.cpp Source File

00001 
00006 #include "ThriftFlumeEventServer.h"
00007 
00008 
00009 
00010 uint32_t ThriftFlumeEventServer_append_args::read(::apache::thrift::protocol::TProtocol* iprot) {
00011 
00012   uint32_t xfer = 0;
00013   std::string fname;
00014   ::apache::thrift::protocol::TType ftype;
00015   int16_t fid;
00016 
00017   xfer += iprot->readStructBegin(fname);
00018 
00019   using ::apache::thrift::protocol::TProtocolException;
00020 
00021 
00022   while (true)
00023   {
00024     xfer += iprot->readFieldBegin(fname, ftype, fid);
00025     if (ftype == ::apache::thrift::protocol::T_STOP) {
00026       break;
00027     }
00028     switch (fid)
00029     {
00030       case 1:
00031         if (ftype == ::apache::thrift::protocol::T_STRUCT) {
00032           xfer += this->evt.read(iprot);
00033           this->__isset.evt = true;
00034         } else {
00035           xfer += iprot->skip(ftype);
00036         }
00037         break;
00038       default:
00039         xfer += iprot->skip(ftype);
00040         break;
00041     }
00042     xfer += iprot->readFieldEnd();
00043   }
00044 
00045   xfer += iprot->readStructEnd();
00046 
00047   return xfer;
00048 }
00049 
00050 uint32_t ThriftFlumeEventServer_append_args::write(::apache::thrift::protocol::TProtocol* oprot) const {
00051   uint32_t xfer = 0;
00052   xfer += oprot->writeStructBegin("ThriftFlumeEventServer_append_args");
00053   xfer += oprot->writeFieldBegin("evt", ::apache::thrift::protocol::T_STRUCT, 1);
00054   xfer += this->evt.write(oprot);
00055   xfer += oprot->writeFieldEnd();
00056   xfer += oprot->writeFieldStop();
00057   xfer += oprot->writeStructEnd();
00058   return xfer;
00059 }
00060 
00061 uint32_t ThriftFlumeEventServer_append_pargs::write(::apache::thrift::protocol::TProtocol* oprot) const {
00062   uint32_t xfer = 0;
00063   xfer += oprot->writeStructBegin("ThriftFlumeEventServer_append_pargs");
00064   xfer += oprot->writeFieldBegin("evt", ::apache::thrift::protocol::T_STRUCT, 1);
00065   xfer += (*(this->evt)).write(oprot);
00066   xfer += oprot->writeFieldEnd();
00067   xfer += oprot->writeFieldStop();
00068   xfer += oprot->writeStructEnd();
00069   return xfer;
00070 }
00071 
00072 uint32_t ThriftFlumeEventServer_close_args::read(::apache::thrift::protocol::TProtocol* iprot) {
00073 
00074   uint32_t xfer = 0;
00075   std::string fname;
00076   ::apache::thrift::protocol::TType ftype;
00077   int16_t fid;
00078 
00079   xfer += iprot->readStructBegin(fname);
00080 
00081   using ::apache::thrift::protocol::TProtocolException;
00082 
00083 
00084   while (true)
00085   {
00086     xfer += iprot->readFieldBegin(fname, ftype, fid);
00087     if (ftype == ::apache::thrift::protocol::T_STOP) {
00088       break;
00089     }
00090     switch (fid)
00091     {
00092       default:
00093         xfer += iprot->skip(ftype);
00094         break;
00095     }
00096     xfer += iprot->readFieldEnd();
00097   }
00098 
00099   xfer += iprot->readStructEnd();
00100 
00101   return xfer;
00102 }
00103 
00104 uint32_t ThriftFlumeEventServer_close_args::write(::apache::thrift::protocol::TProtocol* oprot) const {
00105   uint32_t xfer = 0;
00106   xfer += oprot->writeStructBegin("ThriftFlumeEventServer_close_args");
00107   xfer += oprot->writeFieldStop();
00108   xfer += oprot->writeStructEnd();
00109   return xfer;
00110 }
00111 
00112 uint32_t ThriftFlumeEventServer_close_pargs::write(::apache::thrift::protocol::TProtocol* oprot) const {
00113   uint32_t xfer = 0;
00114   xfer += oprot->writeStructBegin("ThriftFlumeEventServer_close_pargs");
00115   xfer += oprot->writeFieldStop();
00116   xfer += oprot->writeStructEnd();
00117   return xfer;
00118 }
00119 
00120 uint32_t ThriftFlumeEventServer_close_result::read(::apache::thrift::protocol::TProtocol* iprot) {
00121 
00122   uint32_t xfer = 0;
00123   std::string fname;
00124   ::apache::thrift::protocol::TType ftype;
00125   int16_t fid;
00126 
00127   xfer += iprot->readStructBegin(fname);
00128 
00129   using ::apache::thrift::protocol::TProtocolException;
00130 
00131 
00132   while (true)
00133   {
00134     xfer += iprot->readFieldBegin(fname, ftype, fid);
00135     if (ftype == ::apache::thrift::protocol::T_STOP) {
00136       break;
00137     }
00138     switch (fid)
00139     {
00140       default:
00141         xfer += iprot->skip(ftype);
00142         break;
00143     }
00144     xfer += iprot->readFieldEnd();
00145   }
00146 
00147   xfer += iprot->readStructEnd();
00148 
00149   return xfer;
00150 }
00151 
00152 uint32_t ThriftFlumeEventServer_close_result::write(::apache::thrift::protocol::TProtocol* oprot) const {
00153 
00154   uint32_t xfer = 0;
00155 
00156   xfer += oprot->writeStructBegin("ThriftFlumeEventServer_close_result");
00157 
00158   xfer += oprot->writeFieldStop();
00159   xfer += oprot->writeStructEnd();
00160   return xfer;
00161 }
00162 
00163 uint32_t ThriftFlumeEventServer_close_presult::read(::apache::thrift::protocol::TProtocol* iprot) {
00164 
00165   uint32_t xfer = 0;
00166   std::string fname;
00167   ::apache::thrift::protocol::TType ftype;
00168   int16_t fid;
00169 
00170   xfer += iprot->readStructBegin(fname);
00171 
00172   using ::apache::thrift::protocol::TProtocolException;
00173 
00174 
00175   while (true)
00176   {
00177     xfer += iprot->readFieldBegin(fname, ftype, fid);
00178     if (ftype == ::apache::thrift::protocol::T_STOP) {
00179       break;
00180     }
00181     switch (fid)
00182     {
00183       default:
00184         xfer += iprot->skip(ftype);
00185         break;
00186     }
00187     xfer += iprot->readFieldEnd();
00188   }
00189 
00190   xfer += iprot->readStructEnd();
00191 
00192   return xfer;
00193 }
00194 
00195 void ThriftFlumeEventServerClient::append(const ThriftFlumeEvent& evt)
00196 {
00197   send_append(evt);
00198 }
00199 
00200 void ThriftFlumeEventServerClient::send_append(const ThriftFlumeEvent& evt)
00201 {
00202   int32_t cseqid = 0;
00203   oprot_->writeMessageBegin("append", ::apache::thrift::protocol::T_CALL, cseqid);
00204 
00205   ThriftFlumeEventServer_append_pargs args;
00206   args.evt = &evt;
00207   args.write(oprot_);
00208 
00209   oprot_->writeMessageEnd();
00210   oprot_->getTransport()->writeEnd();
00211   oprot_->getTransport()->flush();
00212 }
00213 
00214 void ThriftFlumeEventServerClient::close()
00215 {
00216   send_close();
00217   recv_close();
00218 }
00219 
00220 void ThriftFlumeEventServerClient::send_close()
00221 {
00222   int32_t cseqid = 0;
00223   oprot_->writeMessageBegin("close", ::apache::thrift::protocol::T_CALL, cseqid);
00224 
00225   ThriftFlumeEventServer_close_pargs args;
00226   args.write(oprot_);
00227 
00228   oprot_->writeMessageEnd();
00229   oprot_->getTransport()->writeEnd();
00230   oprot_->getTransport()->flush();
00231 }
00232 
00233 void ThriftFlumeEventServerClient::recv_close()
00234 {
00235 
00236   int32_t rseqid = 0;
00237   std::string fname;
00238   ::apache::thrift::protocol::TMessageType mtype;
00239 
00240   iprot_->readMessageBegin(fname, mtype, rseqid);
00241   if (mtype == ::apache::thrift::protocol::T_EXCEPTION) {
00242     ::apache::thrift::TApplicationException x;
00243     x.read(iprot_);
00244     iprot_->readMessageEnd();
00245     iprot_->getTransport()->readEnd();
00246     throw x;
00247   }
00248   if (mtype != ::apache::thrift::protocol::T_REPLY) {
00249     iprot_->skip(::apache::thrift::protocol::T_STRUCT);
00250     iprot_->readMessageEnd();
00251     iprot_->getTransport()->readEnd();
00252   }
00253   if (fname.compare("close") != 0) {
00254     iprot_->skip(::apache::thrift::protocol::T_STRUCT);
00255     iprot_->readMessageEnd();
00256     iprot_->getTransport()->readEnd();
00257   }
00258   ThriftFlumeEventServer_close_presult result;
00259   result.read(iprot_);
00260   iprot_->readMessageEnd();
00261   iprot_->getTransport()->readEnd();
00262 
00263   return;
00264 }
00265 
00266 bool ThriftFlumeEventServerProcessor::process(boost::shared_ptr< ::apache::thrift::protocol::TProtocol> piprot, boost::shared_ptr< ::apache::thrift::protocol::TProtocol> poprot, void* callContext) {
00267 
00268   ::apache::thrift::protocol::TProtocol* iprot = piprot.get();
00269   ::apache::thrift::protocol::TProtocol* oprot = poprot.get();
00270   std::string fname;
00271   ::apache::thrift::protocol::TMessageType mtype;
00272   int32_t seqid;
00273 
00274   iprot->readMessageBegin(fname, mtype, seqid);
00275 
00276   if (mtype != ::apache::thrift::protocol::T_CALL && mtype != ::apache::thrift::protocol::T_ONEWAY) {
00277     iprot->skip(::apache::thrift::protocol::T_STRUCT);
00278     iprot->readMessageEnd();
00279     iprot->getTransport()->readEnd();
00280     ::apache::thrift::TApplicationException x(::apache::thrift::TApplicationException::INVALID_MESSAGE_TYPE);
00281     oprot->writeMessageBegin(fname, ::apache::thrift::protocol::T_EXCEPTION, seqid);
00282     x.write(oprot);
00283     oprot->writeMessageEnd();
00284     oprot->getTransport()->writeEnd();
00285     oprot->getTransport()->flush();
00286     return true;
00287   }
00288 
00289   return process_fn(iprot, oprot, fname, seqid, callContext);
00290 }
00291 
00292 bool ThriftFlumeEventServerProcessor::process_fn(::apache::thrift::protocol::TProtocol* iprot, ::apache::thrift::protocol::TProtocol* oprot, std::string& fname, int32_t seqid, void* callContext) {
00293   std::map<std::string, void (ThriftFlumeEventServerProcessor::*)(int32_t, ::apache::thrift::protocol::TProtocol*, ::apache::thrift::protocol::TProtocol*, void*)>::iterator pfn;
00294   pfn = processMap_.find(fname);
00295   if (pfn == processMap_.end()) {
00296     iprot->skip(::apache::thrift::protocol::T_STRUCT);
00297     iprot->readMessageEnd();
00298     iprot->getTransport()->readEnd();
00299     ::apache::thrift::TApplicationException x(::apache::thrift::TApplicationException::UNKNOWN_METHOD, "Invalid method name: '"+fname+"'");
00300     oprot->writeMessageBegin(fname, ::apache::thrift::protocol::T_EXCEPTION, seqid);
00301     x.write(oprot);
00302     oprot->writeMessageEnd();
00303     oprot->getTransport()->writeEnd();
00304     oprot->getTransport()->flush();
00305     return true;
00306   }
00307   (this->*(pfn->second))(seqid, iprot, oprot, callContext);
00308   return true;
00309 }
00310 
00311 void ThriftFlumeEventServerProcessor::process_append(int32_t seqid, ::apache::thrift::protocol::TProtocol* iprot, ::apache::thrift::protocol::TProtocol* oprot, void* callContext)
00312 {
00313   void* ctx = NULL;
00314   if (eventHandler_.get() != NULL) {
00315     ctx = eventHandler_->getContext("ThriftFlumeEventServer.append", callContext);
00316   }
00317   ::apache::thrift::TProcessorContextFreer freer(eventHandler_.get(), ctx, "ThriftFlumeEventServer.append");
00318 
00319   if (eventHandler_.get() != NULL) {
00320     eventHandler_->preRead(ctx, "ThriftFlumeEventServer.append");
00321   }
00322 
00323   ThriftFlumeEventServer_append_args args;
00324   args.read(iprot);
00325   iprot->readMessageEnd();
00326   uint32_t bytes = iprot->getTransport()->readEnd();
00327 
00328   if (eventHandler_.get() != NULL) {
00329     eventHandler_->postRead(ctx, "ThriftFlumeEventServer.append", bytes);
00330   }
00331 
00332   try {
00333     iface_->append(args.evt);
00334   } catch (const std::exception& e) {
00335     if (eventHandler_.get() != NULL) {
00336       eventHandler_->handlerError(ctx, "ThriftFlumeEventServer.append");
00337     }
00338     return;
00339   }
00340 
00341   if (eventHandler_.get() != NULL) {
00342     eventHandler_->asyncComplete(ctx, "ThriftFlumeEventServer.append");
00343   }
00344 
00345   return;
00346 }
00347 
00348 void ThriftFlumeEventServerProcessor::process_close(int32_t seqid, ::apache::thrift::protocol::TProtocol* iprot, ::apache::thrift::protocol::TProtocol* oprot, void* callContext)
00349 {
00350   void* ctx = NULL;
00351   if (eventHandler_.get() != NULL) {
00352     ctx = eventHandler_->getContext("ThriftFlumeEventServer.close", callContext);
00353   }
00354   ::apache::thrift::TProcessorContextFreer freer(eventHandler_.get(), ctx, "ThriftFlumeEventServer.close");
00355 
00356   if (eventHandler_.get() != NULL) {
00357     eventHandler_->preRead(ctx, "ThriftFlumeEventServer.close");
00358   }
00359 
00360   ThriftFlumeEventServer_close_args args;
00361   args.read(iprot);
00362   iprot->readMessageEnd();
00363   uint32_t bytes = iprot->getTransport()->readEnd();
00364 
00365   if (eventHandler_.get() != NULL) {
00366     eventHandler_->postRead(ctx, "ThriftFlumeEventServer.close", bytes);
00367   }
00368 
00369   ThriftFlumeEventServer_close_result result;
00370   try {
00371     iface_->close();
00372   } catch (const std::exception& e) {
00373     if (eventHandler_.get() != NULL) {
00374       eventHandler_->handlerError(ctx, "ThriftFlumeEventServer.close");
00375     }
00376 
00377     ::apache::thrift::TApplicationException x(e.what());
00378     oprot->writeMessageBegin("close", ::apache::thrift::protocol::T_EXCEPTION, seqid);
00379     x.write(oprot);
00380     oprot->writeMessageEnd();
00381     oprot->getTransport()->writeEnd();
00382     oprot->getTransport()->flush();
00383     return;
00384   }
00385 
00386   if (eventHandler_.get() != NULL) {
00387     eventHandler_->preWrite(ctx, "ThriftFlumeEventServer.close");
00388   }
00389 
00390   oprot->writeMessageBegin("close", ::apache::thrift::protocol::T_REPLY, seqid);
00391   result.write(oprot);
00392   oprot->writeMessageEnd();
00393   bytes = oprot->getTransport()->writeEnd();
00394   oprot->getTransport()->flush();
00395 
00396   if (eventHandler_.get() != NULL) {
00397     eventHandler_->postWrite(ctx, "ThriftFlumeEventServer.close", bytes);
00398   }
00399 }
00400 
00401 
00402