src/Transport/THttpClient.cs \
src/Transport/THttpHandler.cs \
src/Transport/TMemoryBuffer.cs \
+ src/Transport/TNamedPipeClientTransport.cs \
+ src/Transport/TNamedPipeServerTransport.cs \
src/TProcessor.cs \
src/TException.cs \
src/TApplicationException.cs
<Compile Include="Transport\TFramedTransport.cs" />
<Compile Include="Transport\THttpClient.cs" />
<Compile Include="Transport\THttpHandler.cs" />
+ <Compile Include="Transport\TNamedPipeClientTransport.cs" />
+ <Compile Include="Transport\TNamedPipeServerTransport.cs" />
<Compile Include="Transport\TServerSocket.cs" />
<Compile Include="Transport\TServerTransport.cs" />
<Compile Include="Transport\TSocket.cs" />
--- /dev/null
+\feffusing System.IO.Pipes;
+
+namespace Thrift.Transport
+{
+ public class TNamedPipeClientTransport : TTransport
+ {
+ private NamedPipeClientStream client;
+ private string ServerName;
+ private string PipeName;
+
+ public TNamedPipeClientTransport(string pipe)
+ {
+ ServerName = ".";
+ PipeName = pipe;
+ }
+
+ public TNamedPipeClientTransport(string server, string pipe)
+ {
+ ServerName = (server != "") ? server : ".";
+ PipeName = pipe;
+ }
+
+ public override bool IsOpen
+ {
+ get { return client != null && client.IsConnected; }
+ }
+
+ public override void Open()
+ {
+ if (IsOpen)
+ {
+ throw new TTransportException(TTransportException.ExceptionType.AlreadyOpen);
+ }
+ client = new NamedPipeClientStream(ServerName, PipeName, PipeDirection.InOut, PipeOptions.None);
+ client.Connect();
+ }
+
+ public override void Close()
+ {
+ if (client != null)
+ {
+ client.Close();
+ client = null;
+ }
+ }
+
+ public override int Read(byte[] buf, int off, int len)
+ {
+ if (client == null)
+ {
+ throw new TTransportException(TTransportException.ExceptionType.NotOpen);
+ }
+
+ return client.Read(buf, off, len);
+ }
+
+ public override void Write(byte[] buf, int off, int len)
+ {
+ if (client == null)
+ {
+ throw new TTransportException(TTransportException.ExceptionType.NotOpen);
+ }
+
+ client.Write(buf, off, len);
+ }
+
+ protected override void Dispose(bool disposing)
+ {
+ client.Dispose();
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+\feffusing System;
+using System.Collections.Generic;
+using System.IO.Pipes;
+
+namespace Thrift.Transport
+{
+ public class TNamedPipeServerTransport : TServerTransport
+ {
+ /// <summary>
+ /// This is the address of the Pipe on the localhost.
+ /// </summary>
+ private readonly string pipeAddress;
+ NamedPipeServerStream stream = null;
+
+ public TNamedPipeServerTransport(string pipeAddress)
+ {
+ this.pipeAddress = pipeAddress;
+ }
+
+ public override void Listen()
+ {
+ // nothing to do here
+ }
+
+ public override void Close()
+ {
+ if (stream != null)
+ {
+ try
+ {
+ stream.Close();
+ stream.Dispose();
+ }
+ finally
+ {
+ stream = null;
+ }
+ }
+ }
+
+ private void EnsurePipeInstance()
+ {
+ if( stream == null)
+ stream = new NamedPipeServerStream(
+ pipeAddress, PipeDirection.InOut, 254,
+ PipeTransmissionMode.Byte,
+ PipeOptions.None, 4096, 4096 /*TODO: security*/);
+ }
+
+ protected override TTransport AcceptImpl()
+ {
+ try
+ {
+ EnsurePipeInstance();
+ stream.WaitForConnection();
+ var trans = new ServerTransport(stream);
+ stream = null; // pass ownership to ServerTransport
+ return trans;
+ }
+ catch (Exception e)
+ {
+ Close();
+ throw new TTransportException(TTransportException.ExceptionType.NotOpen, e.Message);
+ }
+ }
+
+ private class ServerTransport : TTransport
+ {
+ private NamedPipeServerStream server;
+ public ServerTransport(NamedPipeServerStream server)
+ {
+ this.server = server;
+ }
+
+ public override bool IsOpen
+ {
+ get { return server != null && server.IsConnected; }
+ }
+
+ public override void Open()
+ {
+ }
+
+ public override void Close()
+ {
+ if (server != null) server.Close();
+ }
+
+ public override int Read(byte[] buf, int off, int len)
+ {
+ if (server == null)
+ {
+ throw new TTransportException(TTransportException.ExceptionType.NotOpen);
+ }
+ return server.Read(buf, off, len);
+ }
+
+ public override void Write(byte[] buf, int off, int len)
+ {
+ if (server == null)
+ {
+ throw new TTransportException(TTransportException.ExceptionType.NotOpen);
+ }
+ server.Write(buf, off, len);
+ }
+
+ protected override void Dispose(bool disposing)
+ {
+ server.Dispose();
+ }
+ }
+ }
+}
\ No newline at end of file
{
string host = "localhost";
int port = 9090;
- string url = null;
+ string url = null, pipe = null;
int numThreads = 1;
bool buffered = false, framed = false;
framed = true;
Console.WriteLine("Using framed transport");
}
+ else if (args[i] == "-pipe") // -pipe <name>
+ {
+ pipe = args[++i];
+ Console.WriteLine("Using named pipes transport");
+ }
else if (args[i] == "-t")
{
numThreads = Convert.ToInt32(args[++i]);
threads[test] = t;
if (url == null)
{
- TTransport trans = new TSocket(host, port);
+ // endpoint transport
+ TTransport trans = null;
+ if( pipe != null)
+ trans = new TNamedPipeClientTransport(pipe);
+ else
+ trans = new TSocket(host, port);
+
+ // layered transport
if (buffered)
trans = new TBufferedTransport(trans as TStreamTransport);
if (framed)
return thing;
}
- public Dictionary<string, string> testStringMap(Dictionary<string, string> thing)
- {
- Console.WriteLine("testStringMap({");
- bool first = true;
- foreach (string key in thing.Keys)
- {
- if (first)
- {
- first = false;
- }
- else
- {
- Console.WriteLine(", ");
- }
- Console.WriteLine(key + " => " + thing[key]);
- }
- Console.WriteLine("})");
- return thing;
- }
+ public Dictionary<string, string> testStringMap(Dictionary<string, string> thing)
+ {
+ Console.WriteLine("testStringMap({");
+ bool first = true;
+ foreach (string key in thing.Keys)
+ {
+ if (first)
+ {
+ first = false;
+ }
+ else
+ {
+ Console.WriteLine(", ");
+ }
+ Console.WriteLine(key + " => " + thing[key]);
+ }
+ Console.WriteLine("})");
+ return thing;
+ }
public THashSet<int> testSet(THashSet<int> thing)
{
try
{
bool useBufferedSockets = false, useFramed = false;
- int port = 9090;
+ int port = 9090, i = 0;
+ string pipe = null;
if (args.Length > 0)
{
- port = int.Parse(args[0]);
+ i = 0;
+ if (args[i] == "-pipe") // -pipe name
+ {
+ pipe = args[++i];
+ }
+ else // default to port number (compatibility)
+ {
+ port = int.Parse(args[i]);
+ }
- if (args.Length > 1)
+ ++i;
+ if (args.Length > i)
{
- if ( args[1] == "raw" )
+ if ( args[i] == "raw" )
{
// as default
}
- else if ( args[1] == "buffered" )
+ else if ( args[i] == "buffered" )
{
useBufferedSockets = true;
}
- else if ( args[1] == "framed" )
+ else if (args[i] == "framed")
{
useFramed = true;
}
else
{
// Fall back to the older boolean syntax
- bool.TryParse(args[1], out useBufferedSockets);
+ bool.TryParse(args[i], out useBufferedSockets);
}
}
}
ThriftTest.Processor testProcessor = new ThriftTest.Processor(testHandler);
// Transport
- TServerSocket tServerSocket = new TServerSocket(port, 0, useBufferedSockets);
+ TServerTransport trans;
+ if( pipe != null)
+ {
+ trans = new TNamedPipeServerTransport(pipe);
+ }
+ else
+ {
+ trans = new TServerSocket(port, 0, useBufferedSockets);
+ }
// Simple Server
TServer serverEngine;
if ( useFramed )
- serverEngine = new TSimpleServer(testProcessor, tServerSocket, new TFramedTransport.Factory());
+ serverEngine = new TSimpleServer(testProcessor, trans, new TFramedTransport.Factory());
else
- serverEngine = new TSimpleServer(testProcessor, tServerSocket);
+ serverEngine = new TSimpleServer(testProcessor, trans);
// ThreadPool Server
// serverEngine = new TThreadPoolServer(testProcessor, tServerSocket);
testHandler.server = serverEngine;
// Run it
- Console.WriteLine("Starting the server on port " + port +
+ string where = ( pipe != null ? "on pipe "+pipe : "on port " + port);
+ Console.WriteLine("Starting the server " +where+
(useBufferedSockets ? " with buffered socket" : "") +
(useFramed ? " with framed transport" : "") +
"...");
Code Review / common / thrift.git / commitdiff
