I'm using TSimpleServer in C++ and wanted to use glib's main loop
implementation, mainly for the ability to schedule other things to be
done whenever the thrift server is idle. Is there a clean way to do
this? My desperate solution was to hack Thrift to give outside code
access to its file descriptors, so that glib's main loop can poll
Thrift's socket and let it process a request when necessary.
I've attached my patch for the curious, but it's ugly and was the bare
minimum just to get my code doing what I wanted.
Is there a better way to achieve this?
Index: lib/cpp/src/server/TSimpleServer.cpp
===================================================================
--- lib/cpp/src/server/TSimpleServer.cpp (revision 924958)
+++ lib/cpp/src/server/TSimpleServer.cpp (working copy)
@@ -42,12 +42,14 @@
shared_ptr<TProtocol> inputProtocol;
shared_ptr<TProtocol> outputProtocol;
- try {
- // Start the server listening
- serverTransport_->listen();
- } catch (TTransportException& ttx) {
- cerr << "TSimpleServer::run() listen(): " << ttx.what() << endl;
- return;
+ if(!oneAtATime_) {
+ try {
+ // Start the server listening
+ serverTransport_->listen();
+ } catch (TTransportException& ttx) {
+ cerr << "TSimpleServer::run() listen(): " << ttx.what() << endl;
+ return;
+ }
}
// Run the preServe event
@@ -56,7 +58,7 @@
}
// Fetch client from server
- while (!stop_) {
+ for (int i = 0; !stop_ && (i == 0 || !oneAtATime_); ++i) {
try {
client = serverTransport_->accept();
inputTransport = inputTransportFactory_->getTransport(client);
Index: lib/cpp/src/server/TSimpleServer.h
===================================================================
--- lib/cpp/src/server/TSimpleServer.h (revision 924958)
+++ lib/cpp/src/server/TSimpleServer.h (working copy)
@@ -37,21 +37,25 @@
TSimpleServer(boost::shared_ptr<TProcessor> processor,
boost::shared_ptr<TServerTransport> serverTransport,
boost::shared_ptr<TTransportFactory> transportFactory,
- boost::shared_ptr<TProtocolFactory> protocolFactory) :
+ boost::shared_ptr<TProtocolFactory> protocolFactory,
+ bool oneAtATime) :
TServer(processor, serverTransport, transportFactory, protocolFactory),
- stop_(false) {}
+ oneAtATime_(oneAtATime), stop_(false) {}
TSimpleServer(boost::shared_ptr<TProcessor> processor,
boost::shared_ptr<TServerTransport> serverTransport,
boost::shared_ptr<TTransportFactory> inputTransportFactory,
boost::shared_ptr<TTransportFactory> outputTransportFactory,
boost::shared_ptr<TProtocolFactory> inputProtocolFactory,
- boost::shared_ptr<TProtocolFactory> outputProtocolFactory):
+ boost::shared_ptr<TProtocolFactory> outputProtocolFactory,
+ bool oneAtATime):
TServer(processor, serverTransport,
inputTransportFactory, outputTransportFactory,
inputProtocolFactory, outputProtocolFactory),
- stop_(false) {}
+ oneAtATime_(oneAtATime), stop_(false) {}
+ bool oneAtATime_;
+
~TSimpleServer() {}
void serve();
Index: lib/cpp/src/transport/TServerSocket.cpp
===================================================================
--- lib/cpp/src/transport/TServerSocket.cpp (revision 924958)
+++ lib/cpp/src/transport/TServerSocket.cpp (working copy)
@@ -18,6 +18,7 @@
*/
#include <cstring>
+#include <vector>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/poll.h>
@@ -259,6 +260,19 @@
// The socket is now listening!
}
+std::vector<struct pollfd> TServerSocket::getFDs() {
+ struct pollfd fds[2];
+ std::memset(fds, 0 , sizeof(fds));
+ fds[0].fd = this->serverSocket_;
+ fds[0].events = POLLIN;
+ if (intSock2_ >= 0) {
+ fds[1].fd = this->intSock2_;
+ fds[1].events = POLLIN;
+ }
+
+ return std::vector<struct pollfd>(fds, fds+(this->intSock2_ >= 0 ? 2 : 1));
+}
+
shared_ptr<TTransport> TServerSocket::acceptImpl() {
if (serverSocket_ < 0) {
throw TTransportException(TTransportException::NOT_OPEN, "TServerSocket not listening");
Index: lib/cpp/src/transport/TServerSocket.h
===================================================================
--- lib/cpp/src/transport/TServerSocket.h (revision 924958)
+++ lib/cpp/src/transport/TServerSocket.h (working copy)
@@ -23,6 +23,9 @@
#include "TServerTransport.h"
#include <boost/shared_ptr.hpp>
+#include <vector>
+#include <sys/poll.h>
+
namespace apache { namespace thrift { namespace transport {
class TSocket;
@@ -53,6 +56,8 @@
void interrupt();
+ std::vector<struct pollfd> getFDs();
+
protected:
boost::shared_ptr<TTransport> acceptImpl();
Index: lib/cpp/src/transport/TServerTransport.h
===================================================================
--- lib/cpp/src/transport/TServerTransport.h (revision 924958)
+++ lib/cpp/src/transport/TServerTransport.h (working copy)
@@ -24,6 +24,9 @@
#include "TTransportException.h"
#include <boost/shared_ptr.hpp>
+#include <vector>
+#include <sys/poll.h>
+
namespace apache { namespace thrift { namespace transport {
/**
@@ -74,6 +77,8 @@
*/
virtual void close() = 0;
+ virtual std::vector<struct pollfd> getFDs() = 0;
+
protected:
TServerTransport() {}
