public
Last active

PipeStream with Rx that works!

  • Download Gist
gistfile1.cs
C#
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317
using System;
using System.IO;
using System.IO.Pipes;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Runtime.Serialization;
using System.Runtime.Serialization.Formatters.Binary;
using System.Security.AccessControl;
using System.Security.Principal;
using System.Threading;
 
namespace YourNameSpace
{
public static class PipeStreamObservable
{
private static readonly PipeSecurity PipeSecurity;
 
private static readonly ThreadLocal<BinaryFormatter> BinaryFormatter
= new ThreadLocal<BinaryFormatter>(() => new BinaryFormatter());
 
static PipeStreamObservable()
{
PipeSecurity = new PipeSecurity();
PipeSecurity.AddAccessRule(
new PipeAccessRule(WindowsIdentity.GetCurrent().User, PipeAccessRights.FullControl, AccessControlType.Allow)
);
PipeSecurity.AddAccessRule(
new PipeAccessRule(
new SecurityIdentifier(WellKnownSidType.WorldSid, null), PipeAccessRights.ReadWrite,
AccessControlType.Allow
)
);
}
 
public static void Write<T>(this PipeStream stream, T type)
{
BinaryFormatter.Value.Serialize(stream, type);
}
 
public static IObservable<T> Create<T>(out NamedPipeServerStream stream, string pipeName,
EventHandler onConnected = null)
{
stream = new NamedPipeServerStream(pipeName, PipeDirection.InOut,
-1, PipeTransmissionMode.Byte,
PipeOptions.Asynchronous, 4096, 4096, PipeSecurity);
 
var serverStream = stream;
return Observable.Using(() => serverStream, resource => Observable.Create<T>(o =>
{
 
var currentStateSubscription = new SerialDisposable();
 
return NewThreadScheduler.Default.Schedule(
new FormatterIterator<T>(resource, new BinaryFormatter(), onConnected), (state, self) =>
{
var observable = state.ReadNext();
if (observable != null)
currentStateSubscription.Disposable =
observable.Subscribe(
str =>
{
self(state);
o.OnNext(str);
},
ex =>
{
currentStateSubscription
.Dispose();
o.OnError(ex);
resource.Close();
},
() =>
{
currentStateSubscription
.Dispose();
o.OnCompleted();
resource.Close();
});
else
o.OnError(new Exception("Scheduling aborted."));
}
);
}));
}
 
public static IObservable<T> Create<T>(out NamedPipeClientStream stream,
string server,
string pipeName,
EventHandler onConnected = null)
{
stream = new NamedPipeClientStream(server, pipeName, PipeDirection.InOut,
PipeOptions.Asynchronous);
 
var clientStream = stream;
return Observable.Using(() => clientStream, resource => Observable.Create<T>(o =>
{
 
var currentStateSubscription = new SerialDisposable();
return NewThreadScheduler.Default.Schedule(
new FormatterIterator<T>(resource, new BinaryFormatter(), onConnected), (state, self) =>
{
var observable = state.ReadNext();
if (observable != null)
currentStateSubscription.Disposable =
observable.Subscribe(
str =>
{
self(state);
o.OnNext(str);
},
ex =>
{
currentStateSubscription
.Dispose();
o.OnError(ex);
resource.Close();
},
() =>
{
currentStateSubscription
.Dispose();
o.OnCompleted();
resource.Close();
});
else
o.OnError(new Exception("Scheduling aborted."));
}
);
}));
}
 
private delegate bool StreamHandler<T>(Stream input, out T output);
 
public interface IIterator<out T>
{
IObservable<T> ReadNext();
}
 
private abstract class ReaderState<T> : IIterator<T>
{
public abstract IObservable<T> ReadNext();
}
 
private class ReadReadyState<T> : ReaderState<T>
{
private readonly PipeStream _stream;
private readonly StreamHandler<T> _handler;
 
 
internal ReadReadyState(PipeStream stream, StreamHandler<T> handler)
{
_stream = stream;
_handler = handler;
}
 
public override IObservable<T> ReadNext()
{
return Observable.Create<T>(o =>
{
try
{
if (_stream.IsConnected)
{
T value;
if (_handler(_stream, out value))
o.OnNext(value);
else
o.OnCompleted();
}
}
catch (Exception e)
{
o.OnError(e);
}
 
return Disposable.Empty;
});
}
}
 
private class ServerStreamReader<T> : IIterator<T>
{
private ReaderState<T> _currentState;
private readonly NamedPipeServerStream _stream;
private readonly StreamHandler<T> _handler;
 
internal ServerStreamReader(NamedPipeServerStream stream, StreamHandler<T> handler, EventHandler onConnected)
{
_stream = stream;
_handler = handler;
_currentState = new ConnectionWaitState<T>(this, onConnected);
}
 
private class ConnectionWaitState<T1> : ReaderState<T1>
{
private readonly ServerStreamReader<T1> _parent;
private readonly EventHandler _onConnected;
internal ConnectionWaitState(ServerStreamReader<T1> parent, EventHandler onConnected)
{
_parent = parent;
_onConnected += onConnected;
}
 
public override IObservable<T1> ReadNext()
{
try
{
_parent._stream.WaitForConnection();
if (_onConnected != null)
_onConnected(this, EventArgs.Empty);
_parent._currentState = new ReadReadyState<T1>(_parent._stream, _parent._handler);
return _parent._currentState.ReadNext();
}
catch (Exception)
{
}
return null;
}
}
 
public IObservable<T> ReadNext()
{
return _currentState.ReadNext();
}
}
 
private class ClientStreamReader<T> : IIterator<T>
{
private ReaderState<T> _currentState;
private readonly NamedPipeClientStream _stream;
private readonly StreamHandler<T> _handler;
 
internal ClientStreamReader(NamedPipeClientStream stream, StreamHandler<T> handler, EventHandler onConnected)
{
_stream = stream;
_handler = handler;
_currentState = new ConnectionWaitState<T>(this, onConnected);
}
 
private class ConnectionWaitState<T1> : ReaderState<T1>
{
private readonly ClientStreamReader<T1> _parent;
private readonly EventHandler _onConnected;
 
internal ConnectionWaitState(ClientStreamReader<T1> parent, EventHandler onConnected)
{
_parent = parent;
_onConnected += onConnected;
}
 
public override IObservable<T1> ReadNext()
{
try
{
_parent._stream.Connect(5000);
if (_onConnected != null)
_onConnected(this, EventArgs.Empty);
_parent._currentState = new ReadReadyState<T1>(_parent._stream, _parent._handler);
return _parent._currentState.ReadNext();
}
catch (Exception)
{
}
return null;
}
}
 
public IObservable<T> ReadNext()
{
return _currentState.ReadNext();
}
}
 
private sealed class FormatterIterator<T> : IIterator<T>
{
private readonly IIterator<T> _iterator;
private readonly IFormatter _formatter;
 
public FormatterIterator(NamedPipeServerStream source, IFormatter formatter, EventHandler onConnected)
{
_iterator = new ServerStreamReader<T>(source, DeserializeWithFormatter, onConnected);
_formatter = formatter;
}
 
public FormatterIterator(NamedPipeClientStream source, IFormatter formatter, EventHandler onConnected)
{
_iterator = new ClientStreamReader<T>(source, DeserializeWithFormatter, onConnected);
_formatter = formatter;
}
 
public IObservable<T> ReadNext()
{
return _iterator.ReadNext();
}
 
private bool DeserializeWithFormatter(Stream stream, out T value)
{
try
{
value = (T) _formatter.Deserialize(stream);
return true;
}
catch (Exception)
{
value = default(T);
return false;
}
}
}
}
}
gistfile2.txt
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
namespace YourNameSpace
{
public interface IPipeStreamObserver<in T> : IObserver<T>
{
PipeStream PipeStream { get; set; }
void OnConnected();
}
public class IpcClient<T>
{
public IPipeStreamObserver<T> Observer { get; set; }
private readonly AutoResetEvent _auto = new AutoResetEvent(false);
private readonly string _server;
private readonly string _name;
 
public IpcClient(string server, string name, IPipeStreamObserver<T> observer)
{
Observer = observer;
_server = server;
_name = name;
}
 
public IDisposable Create()
{
NamedPipeClientStream pipe;
var observable = PipeStreamObservable.Create<T>(out pipe, _server,_name, (sender, args) => _auto.Set());
Observer.PipeStream = pipe;
observable.Subscribe(
Observer.OnNext,
ex =>
{
Observer.OnError(ex);
pipe.Close();
_auto.Set();
},
() =>
{
Observer.OnCompleted();
pipe.Close();
});
_auto.WaitOne();
if (pipe.IsConnected)
Observer.OnConnected();
return pipe;
}
}
public class IpcServer<T>
{
private bool _running;
private readonly AutoResetEvent _auto = new AutoResetEvent(false);
private readonly List<PipeStream> _pipes = new List<PipeStream>();
private readonly string _name;
private Task _task;
 
public IpcServer(string name)
{
_name = name;
_task = Task.Factory.StartNew(() => { });
}
 
public void IpcServerPipeCreate<TPipeStreamObserver>(Task task)
where TPipeStreamObserver : IPipeStreamObserver<T>, new()
{
NamedPipeServerStream pipe;
var observable = PipeStreamObservable.Create<T>(out pipe, _name, (sender, args) => _auto.Set());
 
var observer = new TPipeStreamObserver() {PipeStream = pipe};
observable.Subscribe(
observer.OnNext,
ex =>
{
observer.OnError(ex);
pipe.Close();
lock (_pipes)
{
_pipes.Remove(pipe);
}
},
() =>
{
observer.OnCompleted();
pipe.Close();
lock (_pipes)
{
_pipes.Remove(pipe);
}
});
_auto.WaitOne();
lock (_pipes)
{
if (_running)
_pipes.Add(pipe);
}
 
if (_running)
{
task.ContinueWith(IpcServerPipeCreate<TPipeStreamObserver>);
observer.OnConnected();
}
else
{
pipe.Close();
}
 
}
 
public Task Start<TPipeStreamObserver>() where TPipeStreamObserver : IPipeStreamObserver<T>, new()
{
_running = true;
_task = _task.ContinueWith(IpcServerPipeCreate<TPipeStreamObserver>);
return _task;
}
 
public void Stop()
{
_running = false;
_task.ContinueWith(Stop);
}
 
private void Stop(Task task)
{
lock (_pipes)
{
_running = false;
_pipes.ForEach(pipe => pipe.Close());
}
_auto.Set();
for (;;)
{
int count;
lock (_pipes)
{
count = _pipes.Count;
}
if (count == 0)
break;
}
}
}
}

Here is how it is used:

class ServerObserver : IPipeStreamObserver<string>
{

    public void OnNext(string value)
    {
        value += " World!";
        Console.WriteLine(value);

        PipeStream.Write(value);
    }

    public void OnError(Exception error)
    {

    }

    public void OnCompleted()
    {

    }

    public PipeStream PipeStream { get; set; }

    public void OnConnected()
    {

    }
}

class ClientObserver : IPipeStreamObserver<string>
{

    public void OnNext(string value)
    {
        Console.WriteLine(value);

        PipeStream.Write("Hello ");
    }

    public void OnError(Exception error)
    {
        Console.WriteLine(error);
    }

    public void OnCompleted()
    {

    }

    public PipeStream PipeStream { get; set; }

    public void OnConnected()
    {
        PipeStream.Write("Hello ");
    }
}

class Program
{

    static void Main(string[] args)
    {

        var server = new IpcServer<string>("test");

        server.Start<ServerObserver>();

        var client1 = new IpcClient<string>(".", "test", new ClientObserver()).Create();

        Console.ReadLine();
        client1.Dispose();
        Console.WriteLine("Client1 disposed!");

        var client2 = new IpcClient<string>(".", "test", new ClientObserver()).Create();

        Console.ReadLine();
        client2.Dispose();
        Console.WriteLine("Client2 disposed!");

        var client3 = new IpcClient<string>(".", "test", new ClientObserver()).Create();

        Console.ReadLine();
        client3.Dispose();
        Console.WriteLine("Client3 disposed!");

        Console.ReadLine();
        server.Stop();
        Console.WriteLine("Server stopped!");
        Console.ReadLine();
    }
}

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.