[THRIFT-1558] Named Pipe and Anonymous Pipe transport for Windows

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.9
• Component/s: C++ - Library
• Labels: anonymous named pipe thrift transport

Environment

Microsoft Windows / Visual Studio

• Patch Info: Patch Available
• Assignee: Roger Meier
• Reporter: Peace C
• Created: 10/Apr/12 22:33
• Updated: 11/Oct/12 02:11
• Resolved: 06/Oct/12 06:11

Description

This patch adds named & anonymous pipe transport for Windows. The new classes do not affect *NIX builds. We've been using this code on Windows & OSX for 6 months and I'm fairly confident in it. It has not been hammered by automated tests and I welcome stress testing to wring out any bugs.

The TPipe and TPipeServer classes are generally modeled after TSocket and TSocketServer. The server of course uses TPipeServer to set up the server side then instantiates TPipe for communications. The client instantiates TPipe transport for connection to the server.

Here are some code snippet examples from functions we've built. Variables such as 'pipename' are passed in to the functions. Error handling has been omitted.

//-----------------------------------------------------------------------
// ---- Server ----
//-----------------------------------------------------------------------
#ifdef _WIN32
pipename = "\\\\.\\pipe
" + pipename;
boost::shared_ptr<TServerTransport> transport(new TPipeServer(pipename, 1024, NumThreads)); //Named pipe
#else //Mac, *NIX
unlink(pipename.c_str());
boost::shared_ptr<TServerTransport> transport(new TServerSocket(pipename));
#endif

boost::shared_ptr<TServer> server;
boost::shared_ptr<MyHandler> handler(new MyHandler());
boost::shared_ptr<TProcessor> processor(new MyProcessor(handler));
boost::shared_ptr<TTransportFactory> tfactory(new TBufferedTransportFactory());
boost::shared_ptr<TProtocolFactory> pfactory(new TBinaryProtocolFactory());

if(NumThreads <= 1)

{ //Single-threaded server server.reset(new TSimpleServer(processor, transport, tfactory, pfactory)); }

else

{ //Multi-threaded server boost::shared_ptr<ThreadManager> threadManager = ThreadManager::newSimpleThreadManager(NumThreads); boost::shared_ptr<PlatformThreadFactory> threadFactory = boost::shared_ptr<PlatformThreadFactory>(new PlatformThreadFactory()); threadManager->threadFactory(threadFactory); threadManager->start(); server.reset(new TThreadPoolServer(processor, transport, tfactory, pfactory, threadManager)); }

printf("Starting the 'server'...\n");
server->serve();
printf("done.\n");

//-----------------------------------------------------------------------
// ---- Client ----
//-----------------------------------------------------------------------
#ifdef _WIN32
pipename = "\\\\.\\pipe
" + pipename;
boost::shared_ptr<TTransport> pipe(new TPipe(pipename));
transport.reset(new TBufferedTransport(pipe));
#else //Mac, *nix
boost::shared_ptr<TTransport> socket(new TSocket(pipename));
transport.reset(new TBufferedTransport(socket));
#endif
boost::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));

client.reset(new MyClient(protocol));