Merge pull request #31 from Microsoft/develop

LINQ and Rx core concepts sample

Author: Tuatan

Doc/HotObservables.md

@@ -18,6 +18,7 @@ Both definitions are useful in many real-world situations, but there is a subtle
 
 Let's start with trivial implementation of Hot observable:
 
+```csharp
     class HotObservable : IObservable<int>
     {
         Subject<int> _subject = new Subject<int>();
@@ -33,19 +34,22 @@ Let's start with trivial implementation of Hot observable:
         {
             return _subject.Subscribe(observer);
         }
-    }
+    }
+```
+
 And do a simple query that relies on order: 
 
-    var o = new HotObservable();
+```csharp
+    var o = new HotObservable();
 
     var pairs =
         from f in o
         from n in o.Take(1)
         select new { First = f, Next = n };
 
     pairs.Subscribe(Console.WriteLine);
-    o.Run();
- 
+    o.Run();
+``` 
 
 The output is:
 
@@ -68,16 +72,17 @@ Here:
 
 Now let's try the same query with Cold observable:
 
-    var data = new int[] { 0, 1, 2, 3, 4 };
+```csharp
+    var data = new int[] { 0, 1, 2, 3, 4 };
     var o = data.ToObservable();
 
     var pairs =
         from f in o
         from n in o.Take(1)
         select new { First = f, Next = n };
 
-    pairs.Subscribe(Console.WriteLine);
-
+    pairs.Subscribe(Console.WriteLine);
+```
 
 The output is:
 
@@ -89,13 +94,15 @@ The output is:
 
 This result is exactly the same as LINQ to Objects on the original data as IEnumerable collection:
 
+```csharp
     var data = new int[] { 0, 1, 2, 3, 4 };
     var pairs =
         from f in data
         from n in data.Take(1)
         select new { First = f, Next = n };
 
-    foreach (var p in pairs) Console.WriteLine(p);
+    foreach (var p in pairs) Console.WriteLine(p);
+```
 
 Unfortunately, it is not easy to think about it as a marble diagram - it would look like arrows return back in time to start from 0. 
 
@@ -129,14 +136,16 @@ Supporting Cold observables on the other hand requires that the events are store
 
 Let's consider example of Join query on IIS traces:
 
-	var requests = from b in begin 
+```csharp	
+var requests = from b in begin 
 			   from e in end.Where(e=>e.Header.ActivityId == b.Header.ActivityId).Take(1)
 			   select new
 			   {
 					b.Url,
 					e.HttpStatus,
 					Duration = (e.Header.Timestamp - b.Header.Timestamp).TotalMilliseconds
 				};
+ ```
 
 If "begin" and "end" were small in-memory collections this query could be done using LINQ to Objects. But if they refer to the same trace (say, 1TB etl file on disk):
 
@@ -174,7 +183,8 @@ This allows users to:
 
 Assuming **single, typed** input sequence the cold observable paradox can be avoided by turning the to hot:
 
-	var data = new int[] { 0, 1, 2, 3, 4 };
+```	
+var data = new int[] { 0, 1, 2, 3, 4 };
     var o = data.ToObservable().Publish();
 
     var pairs =
@@ -184,6 +194,7 @@ Assuming **single, typed** input sequence the cold observable paradox can be avo
 
     pairs.Subscribe(Console.WriteLine);
     o.Connect(); 
+```
 
 Behind the scenes, Publish() uses a Subject. 
 

Samples/LinqRxConcepts/AplogEnumerable.txt

@@ -0,0 +1,77 @@
+// Copied from: https://github.com/Microsoft/Tx/blob/master/Source/Tx.Core/ByteArrayExtensions.cs
+
+using System.Text;
+
+namespace System.Reactive
+{
+    public static class ByteArrayExtensions
+    {
+        /// <summary>
+        /// Displays the byte array as hex-dump, like within a debugger
+        /// </summary>
+        /// <param name="bytes">array to display as hex</param>
+        /// <returns></returns>
+        public static string ToHexDump(this byte[] bytes)
+        {
+            StringBuilder sb = new StringBuilder();
+            int lineOffset = 0;
+
+            while (lineOffset < bytes.Length)
+            {
+                // output line offset from the start of the buffer
+                sb.Append(lineOffset.ToString("x4"));
+                sb.Append(": ");
+
+                // output hex dump
+                int endOffset = Math.Min(lineOffset + 16, bytes.Length);
+                int index = 0;
+                for (int byteOffset = lineOffset; byteOffset < endOffset; byteOffset++)
+                {
+                    if (index == 8)
+                        sb.Append(' '); 
+                    
+                    index++;
+                    sb.Append(bytes[byteOffset].ToString("x2"));
+                    sb.Append(' ');
+                }
+
+                // fill in the blanks if we cut off without completing entire line
+                int lineLength = endOffset - lineOffset;
+                if (lineLength < 16)
+                {
+                    for (int i = lineLength; i < 16; i++ )
+                    {
+                        if (index == 8)
+                            sb.Append(' ');
+
+                        index++;
+                        
+                        sb.Append("   ");
+                    }
+                }
+
+                sb.Append("  ");
+
+                // output character dump
+                index = 0;
+                for (int byteOffset = lineOffset; byteOffset < endOffset; byteOffset++)
+                {
+                    index++;
+                    byte b = bytes[byteOffset];
+                    if (b > 32)
+                        sb.Append((char)b);
+                    else
+                        sb.Append('.');
+
+                    if (index == 8)
+                        sb.Append(' ');
+                }
+
+                sb.AppendLine();
+                lineOffset += 16;
+            }
+
+            return sb.ToString();
+        }
+    }
+}

Samples/LinqRxConcepts/CSharp/ByteArrayExtensions.cs

@@ -0,0 +1,62 @@
+using System.Collections;
+using System.Collections.Generic;
+
+namespace System.Linq
+{
+    /// <summary>
+    /// This is mock-up implementation of .Where. It is called Filter to avoid name collision
+    /// </summary>
+    class Filter<T> : IEnumerable<T>
+    {
+        Func<T, bool> _filter;
+        IEnumerable<T> _input;
+
+        public Filter(IEnumerable<T> input, Func<T, bool> filter)
+        {
+            _input = input;
+            _filter = filter;
+        }
+        public IEnumerator<T> GetEnumerator()
+        {
+            return new Enumerator(this, _input.GetEnumerator());
+        }
+
+        IEnumerator IEnumerable.GetEnumerator()
+        {
+            return new Enumerator(this, _input.GetEnumerator());
+        }
+
+        class Enumerator : IEnumerator<T>
+        {
+            Filter<T> _parent;
+            IEnumerator<T> _input;
+            public Enumerator(Filter<T> parent, IEnumerator<T> input)
+            {
+                _parent = parent;
+                _input = input;
+            }
+            public bool MoveNext()
+            {
+                while (true)
+                {
+                    if (!_input.MoveNext())
+                        return false;
+
+                    if (_parent._filter(_input.Current))
+                        return true;
+                }
+            }
+            public T Current  { get { return _input.Current; } }
+            object IEnumerator.Current { get { return _input.Current; } }
+
+            public void Dispose()
+            {
+            }
+
+            public void Reset()
+            {
+                throw new NotImplementedException();
+            }
+        }
+    }
+}

Samples/LinqRxConcepts/LinqPoc/Filter.cs

@@ -0,0 +1,95 @@
+using System.Collections;
+using System.Collections.Generic;
+using System.Reactive;
+
+namespace System.Linq
+{
+    /// <summary>
+    /// This is operator that executes push pipeline inside pull runtime
+    /// </summary>
+    /// <typeparam name="TIn">Input event type</typeparam>
+    /// <typeparam name="TOut">Potpit event type</typeparam>
+    class PushInsidePull<TIn, TOut> : IEnumerable<TOut> 
+    {
+        IEnumerable<TIn> _input;
+        Func<IObservable<TIn>, IObservable<TOut>> _pushPipe;
+        public PushInsidePull(IEnumerable<TIn> input, Func<IObservable<TIn>, IObservable<TOut>> pushPipe) 
+        {
+            _input = input;
+            _pushPipe = pushPipe;
+        }
+        public IEnumerator<TOut> GetEnumerator()
+        {
+            return new Enumerator(this, _input.GetEnumerator(), _pushPipe);
+        }
+
+        IEnumerator IEnumerable.GetEnumerator()
+        {
+            return new Enumerator(this, _input.GetEnumerator(), _pushPipe);
+        }
+
+        class Enumerator : IEnumerator<TOut>
+        {
+            PushInsidePull<TIn, TOut> _parent;
+            IEnumerator<TIn> _input;
+
+            Subject<TIn> _subject = new Subject<TIn>();
+            IObservable<TOut> _pushOutput;
+            PushResultHolder<TOut> _result = new PushResultHolder<TOut>();
+
+            public Enumerator(PushInsidePull<TIn, TOut> parent, IEnumerator<TIn> input, Func<IObservable<TIn>, IObservable<TOut>> pushPipe)
+            {
+                _parent = parent;
+                _input = input;
+                _pushOutput = pushPipe(_subject); // this constructs the pipeline, but does not let it go yet
+                _pushOutput.Subscribe(_result); // this enables the flow... assumind someone pushed events into the _subject
+            }
+            public bool MoveNext()
+            {
+                while (true)
+                {
+                    if (!_input.MoveNext())
+                        return false;
+
+                    _result.HasValue = false;
+                    _subject.OnNext(_input.Current);
+
+                    if (_result.HasValue)
+                        return true;
+                }
+            }
+            public TOut Current { get { return _result.Value; } }
+            object IEnumerator.Current { get { return _result.Value; } }
+
+            public void Dispose()
+            {
+                ;
+            }
+
+            public void Reset()
+            {
+                throw new NotImplementedException();
+            }
+        }
+
+        class PushResultHolder<T> : IObserver<T>
+        {
+            public bool HasValue = false;
+            public T Value;
+            public void OnNext(T value)
+            {
+                Value = value;
+                HasValue = true;
+            }
+            public void OnCompleted()
+            {
+                throw new NotImplementedException();
+            }
+
+            public void OnError(Exception error)
+            {
+                throw new NotImplementedException();
+            }
+        }
+    }
+}

Samples/LinqRxConcepts/LinqPoc/Pull2PushWrapper.cs

@@ -0,0 +1,30 @@
+using System.Collections.Generic;
+
+namespace System.Linq
+{
+    public static class PullLinq
+    {
+        /// <summary>
+        /// Filter does the same exact thing as .Where, and is intended to show how .Where actually works
+        /// </summary>
+        public static IEnumerable<T> Filter<T>(this IEnumerable<T> input, Func<T, bool> filter)
+        {
+            return new Filter<T>(input, filter);
+        }
+
+        /// <summary>
+        /// Invoke a push rule for every element of pull sequence
+        /// </summary>
+        /// <typeparam name="TIn">Input event type</typeparam>
+        /// <typeparam name="TOut">Output event type</typeparam>
+        /// <param name="input">the input sequence</param>
+        /// <param name="pushPipe">function that creates IObservable pipeline</param>
+        /// <returns></returns>
+        public static IEnumerable<TOut> ReplayRealTimeRule<TIn, TOut>(
+            this IEnumerable<TIn> input, 
+            Func<IObservable<TIn>, IObservable<TOut>> pushPipe)
+        {
+            return new PushInsidePull<TIn, TOut>(input, pushPipe);
+        }
+    }
+}