thinkbeforecoding · July 21, 2011 16:01
diff --git a/Program.cs b/Program.cs
 using System;
 using System.IO;
 using System.Net;
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;

 namespace NodeSharp
 {
    class Program
   {
      private const string InTemplate =
 @"<html>
   <head><title>node#</title></head>
   <body>
      Text received : {0}
   </body>
 </html>";
      
      private const string OutTemplate =
 @"<html>
   <head><title>node#</title></head>
   <body>
      Text sent : {0}
   </body>
 </html>";
      
      static void Main()
      {
         var observableOut = GetObservableListener(new HttpListener {Prefixes = {"http://+:80/NodeSharp/Out/"}});
         var observableIn = GetObservableListener(new HttpListener {Prefixes = {"http://+:80/NodeSharp/In/"}});

         var inputText = observableIn
          .Do( c =>
                                          {
                                    using (var writer = new StreamWriter(c.Response.OutputStream))
                                       writer.WriteLine(InTemplate, c.Request.QueryString["text"]);
                                          }
            )
         .Select(c => c.Request.QueryString["text"]);


         Observable.When(observableOut.And(inputText).Then((c, t) => new {Context = c, Text = t}))
            .Subscribe(
               c =>
                  {
                     using (var writer = new StreamWriter(c.Context.Response.OutputStream))
                        writer.WriteLine(OutTemplate, c.Text);
                  }

            );
      
         Console.ReadKey();
      }

      private static IObservable<HttpListenerContext> GetObservableListener(HttpListener listener)
      {
         return Observable.Create<HttpListenerContext>(
               observer =>
               {
                  // start listener on subscription
                  listener.Start();

                  var getContextEvent = 
                     Observable.FromAsyncPattern(
                        listener.BeginGetContext,
                        r => listener.EndGetContext(r));

                  Observable.Defer(getContextEvent)   // observable from factory
                     .Repeat()                  // repeat once done
                     .ObserveOn(Scheduler.TaskPool)   // execute query in taskpool thread
                     .Subscribe(observer);

                  return Disposable.Create(listener.Stop);   // stop when unsubscribing
               }
            );
      }
   }
 }
	using System;
	using System.IO;
	using System.Net;
	using System.Reactive.Concurrency;
	using System.Reactive.Disposables;
	using System.Reactive.Linq;

	namespace NodeSharp
	{
	class Program
	{
	private const string InTemplate =
	@"<html>
	<head><title>node#</title></head>
	<body>
	Text received : {0}
	</body>
	</html>";

	private const string OutTemplate =
	@"<html>
	<head><title>node#</title></head>
	<body>
	Text sent : {0}
	</body>
	</html>";

	static void Main()
	{
	var observableOut = GetObservableListener(new HttpListener {Prefixes = {"http://+:80/NodeSharp/Out/"}});
	var observableIn = GetObservableListener(new HttpListener {Prefixes = {"http://+:80/NodeSharp/In/"}});

	var inputText = observableIn
	.Do( c =>
	{
	using (var writer = new StreamWriter(c.Response.OutputStream))
	writer.WriteLine(InTemplate, c.Request.QueryString["text"]);
	}
	)
	.Select(c => c.Request.QueryString["text"]);


	Observable.When(observableOut.And(inputText).Then((c, t) => new {Context = c, Text = t}))
	.Subscribe(
	c =>
	{
	using (var writer = new StreamWriter(c.Context.Response.OutputStream))
	writer.WriteLine(OutTemplate, c.Text);
	}

	);

	Console.ReadKey();
	}

	private static IObservable<HttpListenerContext> GetObservableListener(HttpListener listener)
	{
	return Observable.Create<HttpListenerContext>(
	observer =>
	{
	// start listener on subscription
	listener.Start();

	var getContextEvent =
	Observable.FromAsyncPattern(
	listener.BeginGetContext,
	r => listener.EndGetContext(r));

	Observable.Defer(getContextEvent) // observable from factory
	.Repeat() // repeat once done
	.ObserveOn(Scheduler.TaskPool) // execute query in taskpool thread
	.Subscribe(observer);

	return Disposable.Create(listener.Stop); // stop when unsubscribing
	}
	);
	}
	}
	}