WCF Duplex Client as Observer

Setup Rx observer over the wire requires WCF Duplex callback:
(1) ServiceContract need to define IObserverCallback, instead of subscrbe method lamda.
(2) Subscribe and unsubscrib() should be one-way to avoid potential blocking
(3) If Session is required then Subscribe/Unsubscrib need to be initiating/terminating
(4) Note that IO.Subcribe is non-blocking (v.s. deprecated ForEach) and its only purpose is to set up OnNext on Service side which call Client Side OnNext.
(5) Service Instancing Context must be PerSession or Single. PerCall cannot do Unsubscribe since state variable bool stopIO is not maintained. 
On the other hand, Single => Unsubscribe will stop all Sessions.
(6) Threading Issues:  WCF service as Single-Thread by default,So reentry will cause deadlock and therefore disallowed ( e.g calling Callback during Service Operation).
 So mark Service Behavior ConcurrencyMode Reentrant/Mutiple
(7) netTcp Binding is Duplex (its name did not say Dual)
(8) Update Svc Reference on Client will trigger WCF host for WCF Lib project to run for debugging/testing.
(9) Client side Auto Gen Service Reference will force Service Constructor taking in InstanceContext, which in term require Callback contract implemntation.
Service Side uses OperationContext.Current.GetCallbackChannel<IObservserCallback>();
(10) Note that Callback contract name on client is IService1Callback <> IObservserCallback service side Callback Contract Name.
(11) CPU <10% when run test with 1ms data push, no network travel.
(12) To signal Complete on an IO, use TakeWhile and check a non-blocking State Variable like stopIO=true.

Contract and Service Definition

    [ServiceContract(CallbackContract = typeof(IObservserCallback),SessionMode= SessionMode.Required)]
    public interface IService1
    {
        [OperationContract(IsOneWay=true, IsInitiating=true)]
        void Subscribe();
        [OperationContract(IsOneWay = true,IsTerminating=true)]
        void Unsubscribe();
    }

    public interface IObservserCallback
    {
        [OperationContract]
        void OnNext(CompositeType data);
    }

    [DataContract]
    public class CompositeType
    {
        double _value = 30.45;
        string _Name = "Name1";

        [DataMember]
        public double Value
        {
            get { return _value; }
            set { _value = value; }
        }

        [DataMember]
        public string Name
        {
            get { return _Name; }
            set { _Name = value; }
        }
    }


Service

   // [ServiceBehavior(InstanceContextMode=InstanceContextMode.Single)]
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.PerSession,ConcurrencyMode=ConcurrencyMode.Reentrant)]
    public class Service1 : IService1
    {
        bool stopIO;  //Session State
        public void Subscribe()
        {
            stopIO = false;
            IObservserCallback client = OperationContext.Current.GetCallbackChannel<IObservserCallback>();
            var IO = Observable.Interval(TimeSpan.FromMilliseconds(1)).TakeWhile(x => { return !stopIO; });
            IO.Subscribe(x =>   // This is non-blocking
            {
              if(client!=null)
                  client.OnNext(new CompositeType() { Name ="Name"+ x.ToString(), Value=(new Random()).NextDouble()*30 });
            });
        }

        public void Unsubscribe()
        {
            stopIO = true;
        }   
    }

Client side Subscription and Observer

    class Program
    {
        static void Main(string[] args)
        {
            ServiceReference1.Service1Client c = new ServiceReference1.Service1Client(new InstanceContext(new callbackObserver()));
            c.Subscribe();
            Console.WriteLine("Wait for Data. To stop IO, hit return");
            Console.ReadLine();
            c.Unsubscribe();
            Console.ReadLine();
        }
    }

    class callbackObserver : ConsoleApplication1.ServiceReference1.IService1Callback
    {
        public void OnNext(CompositeType data)
        {
            Console.WriteLine("Data: " + data.Name +" "+data.Value.ToString());
        }
    }

Drag Canvas around using Rx

WPF UI can be Drag around using mouseMove stream Buffer of two element calculation of Delta:
(1) Observe 3 streams: mouseDown, mouseUp, mouseMove. The first two are point event and need duration to join to mouseMove.
(2) The duration is skip until mouseDown take until mouseUp
(3) Delta is the difference between two mouseMove sampling inside Buffer(2,2). Note that Buffer(count, skip) skip=count is standard, skip>count skip and skip<count is overlapping. Buffer means pile up multiple point into one tally





        public MainWindow()
        {
            InitializeComponent();

            var mouseDown = from evt in Observable.FromEventPattern<MouseButtonEventHandler, MouseButtonEventArgs>(h => cv.MouseDown += h, h => cv.MouseDown -= h)
                            select evt.EventArgs.GetPosition(this);
            var mouseUp = from evt in Observable.FromEventPattern<MouseButtonEventHandler, MouseButtonEventArgs>(h => MouseUp += h, h => MouseUp -= h)
                          select evt.EventArgs.GetPosition(this);
            var mouseMove = from evt in Observable.FromEventPattern<MouseEventHandler, MouseEventArgs>(h => MouseMove += h, h => MouseMove -= h)
                            select evt.EventArgs.GetPosition(this);

            var q = Observable.Join(
                mouseDown,
                mouseMove.Buffer(2, 2),
                 _ => mouseMove.SkipUntil(mouseDown).TakeUntil(mouseUp),
                _ => Observable.Empty<Unit>(),
                (d, consecPts) =>new Point(consecPts[1].X - consecPts[0].X, consecPts[1].Y - consecPts[0].Y));

            q.Subscribe(delta =>
            {
                Canvas.SetLeft(cv, Canvas.GetLeft(cv) + delta.X);
                Canvas.SetTop(cv, Canvas.GetTop(cv) + delta.Y);
            });

// Side Note: To allow resizing do the following and using Microsoft sample Resizing Adorner:
           // a = AdornerLayer.GetAdornerLayer(g); //g=Canvas for layout
           // a.Add(new ResizingAdorner(cv));  // cv= any UIElement inside g
        }
         // AdornerLayer adornerLayer;
    }


 public class ResizingAdorner : Adorner
  {
    Thumb topLeft, topRight, bottomLeft, bottomRight;

    VisualCollection visualChildren;

    public ResizingAdorner(UIElement adornedElement)
      : base(adornedElement)
    {
      visualChildren = new VisualCollection(this);
        
      BuildAdornerCorner(ref topLeft, Cursors.SizeNWSE);
      BuildAdornerCorner(ref topRight, Cursors.SizeNESW);
      BuildAdornerCorner(ref bottomLeft, Cursors.SizeNESW);
      BuildAdornerCorner(ref bottomRight, Cursors.SizeNWSE);

      bottomLeft.DragDelta  += new DragDeltaEventHandler(HandleBottomLeft);
      bottomRight.DragDelta += new DragDeltaEventHandler(HandleBottomRight);
      topLeft.DragDelta     += new DragDeltaEventHandler(HandleTopLeft);
      topRight.DragDelta    += new DragDeltaEventHandler(HandleTopRight);
    }

    #region Handle Resizing from Thumb

    void HandleBottomRight(object sender, DragDeltaEventArgs args)
    {
      FrameworkElement adornedElement = this.AdornedElement as FrameworkElement;
      Thumb hitThumb = sender as Thumb;

      if (adornedElement == null || hitThumb == null) return;
      FrameworkElement parentElement = adornedElement.Parent as FrameworkElement;
      EnforceSize(adornedElement);

      adornedElement.Width = Math.Max(adornedElement.Width + args.HorizontalChange, hitThumb.DesiredSize.Width);
      adornedElement.Height = Math.Max(args.VerticalChange + adornedElement.Height, hitThumb.DesiredSize.Height);
    }

    void HandleBottomLeft(object sender, DragDeltaEventArgs args)
    {
      FrameworkElement adornedElement = AdornedElement as FrameworkElement;
      Thumb hitThumb = sender as Thumb;

      if (adornedElement == null || hitThumb == null) return;

      // Ensure that the Width and Height are properly initialized after the resize.
      EnforceSize(adornedElement);

      adornedElement.Width = Math.Max(adornedElement.Width - args.HorizontalChange, hitThumb.DesiredSize.Width);
      adornedElement.Height = Math.Max(args.VerticalChange + adornedElement.Height, hitThumb.DesiredSize.Height);
    }

    void HandleTopRight(object sender, DragDeltaEventArgs args)
    {
      FrameworkElement adornedElement = this.AdornedElement as FrameworkElement;
      Thumb hitThumb = sender as Thumb;

      if (adornedElement == null || hitThumb == null) return;
      FrameworkElement parentElement = adornedElement.Parent as FrameworkElement;

      EnforceSize(adornedElement);

      adornedElement.Width = Math.Max(adornedElement.Width + args.HorizontalChange, hitThumb.DesiredSize.Width);
      adornedElement.Height = Math.Max(adornedElement.Height - args.VerticalChange, hitThumb.DesiredSize.Height);
    }

    void HandleTopLeft(object sender, DragDeltaEventArgs args)
    {
      FrameworkElement adornedElement = AdornedElement as FrameworkElement;
      Thumb hitThumb = sender as Thumb;

      if (adornedElement == null || hitThumb == null) return;
      EnforceSize(adornedElement);
      adornedElement.Width = Math.Max(adornedElement.Width - args.HorizontalChange, hitThumb.DesiredSize.Width);
      adornedElement.Height = Math.Max(adornedElement.Height - args.VerticalChange, hitThumb.DesiredSize.Height);
    }

    #endregion

    // Arrange thumbs relative to Adorner Center
    protected override Size ArrangeOverride(Size finalSize)
    {
      double desiredWidth  = AdornedElement.DesiredSize.Width;
      double desiredHeight = AdornedElement.DesiredSize.Height;
    
      double adornerWidth = this.DesiredSize.Width;
      double adornerHeight = this.DesiredSize.Height;

      topLeft.Arrange(new Rect(-adornerWidth / 2, -adornerHeight / 2, adornerWidth, adornerHeight));
      topRight.Arrange(new Rect(desiredWidth - adornerWidth / 2, -adornerHeight / 2, adornerWidth, adornerHeight));
      bottomLeft.Arrange(new Rect(-adornerWidth / 2, desiredHeight - adornerHeight / 2, adornerWidth, adornerHeight));
      bottomRight.Arrange(new Rect(desiredWidth - adornerWidth / 2, desiredHeight - adornerHeight / 2, adornerWidth, adornerHeight));

      return finalSize;
    }

 
    void BuildAdornerCorner(ref Thumb cornerThumb, Cursor customizedCursor)
    {
      if (cornerThumb != null) return;

      cornerThumb = new Thumb();

      // Set some arbitrary visual characteristics.
      cornerThumb.Cursor = customizedCursor;
      cornerThumb.Height = cornerThumb.Width = 10;
      cornerThumb.Opacity = 0.40;
      cornerThumb.Background = new SolidColorBrush(Colors.MediumBlue);

      visualChildren.Add(cornerThumb); // Must add to the tree to arrange
    }


    void EnforceSize(FrameworkElement adornedElement)
    {
      if (adornedElement.Width.Equals(Double.NaN)) 
        adornedElement.Width = adornedElement.DesiredSize.Width;
      if (adornedElement.Height.Equals(Double.NaN))
        adornedElement.Height = adornedElement.DesiredSize.Height;

      FrameworkElement parent = adornedElement.Parent as FrameworkElement;
      if (parent != null)
      {
        adornedElement.MaxHeight = parent.ActualHeight;
        adornedElement.MaxWidth = parent.ActualWidth;
      }
    }

    protected override int VisualChildrenCount { get { return visualChildren.Count; } }
    protected override Visual GetVisualChild(int index) { return visualChildren[index]; }
  }

Dualization

(1) Algebra notation indicates IO  is dual of any subject =>func pattern, especially IEnumerable<T>
(2) Monad has M, bind, return and in Continuation Monad
     CM<T> = (T ->() ) -> ()
     return :: T -> CM<T>
     bind ::  CM<T> -> (T -> CM<S>) -> CM<S>




Daulality Code

interface IEnumerable<out T> {
      IEnumerator<T> GetEnumerator(); }

interface IEnumerator<out T>: IDisposable {
       bool MoveNext();
       T Current { get; } }

interface IObservable<out T> {
   IDisposable Subscribe(IObserver<T> observer);  }

interface IObserver<in T>  {
   void OnCompleted(bool done);
   void OnError(Exception exception);
   T OnNext { set; }  }



Covariant vs. Contravariant
  A <: B     A is subtype of B,for TypeContructor C
  C<A> <: C<B>   (Covariant)
  C<B> <: C<A>   (Contravariant)
   
Reading = Covar = <out T>
Writing = Contra = <in T> =pass args into array.

    interface IRequireContravariant<in T> { int Write(T t); }

    interface IRequireCovariant<out T>{ T Read(int t);  }    


IE vs. IO
(1) Concurrency: IE need to remove Concurrency and block MoveNext until Source
ready for Next Value; IO need to add Concurrency to not block source push to target
(2) Async Computation Model
         void Fib( int n, ACtion CB, Action Err, Action Cncl)
         IObserver Fib(int n);
(3) IE can sample source interactively while IO cannot push back high speed source. So IScheduler is define to have time specified:
          IScheduler {
              DateTimeOffset Now
              IDisposable Schedule(state, DateTimeOffset dueTime, Func action)
              IDisposable Schedule(..TimeSpan dueTime,..);
              IDisposable Schedule(TState state, Func action)  // constant push
}

Prism Region Manager, Sevice Locator and Module

Some code to demo additional features of Prism
(1) ServiceLocator.GetInstance = Container.Resolve and inject
(2) Region Manager can define region for injecting UserControl (view)
(3) Module execute injection of view
(4) Adding Module to ModuleCatalog does injection of Region Manager

App run bootstrapper

   public partial class App : Application
    {
        protected override void OnStartup(StartupEventArgs e)
        {
            OtherBootstrapper b = new OtherBootstrapper();
            b.Run(); 
        }
    }
Bootstrapper: Service Locator, ModuleCatalog

   public class OtherBootstrapper :  UnityBootstrapper
    {
        protected override void InitializeShell()
        {
            (Application.Current.MainWindow = (Window)Shell).Show();
        }

        protected override DependencyObject CreateShell()
        {
            //ServiceLocator replaced Container.Resolve
           return ServiceLocator.Current.GetInstance<ShellWin>();
          //  return Container.TryResolve<ShellWin>();         }

        protected override void ConfigureContainer()
        {
           // this will run default config such as EventAggregator, ServiceLocator and region manager
            base.ConfigureContainer();          
        }

        protected override void ConfigureModuleCatalog()
        {
            // amazingly, add ModuelInfo will do DI just like container.Resolve
            ModuleCatalog.AddModule(new ModuleInfo()
            {
                ModuleName = typeof(OtherFeatureModule).Name,
                ModuleType = typeof(OtherFeatureModule).AssemblyQualifiedName
            });
            base.ConfigureModuleCatalog();
        }
    }

   public class OtherFeatureModule : IModule
    {
        IRegionManager _mgr;
        // DI does happend when ModuleInfor Added to Catalog
        public OtherFeatureModule(IRegionManager mgr)
        {
            _mgr = mgr;
        }
        public void Initialize()
        {
            _mgr.Regions["mainReg"].Add(new uc1 ());
        }
    }

ShellWin.xaml  --- Region added to RegionManager.Region Collection
    xmlns:ps="http://www.codeplex.com/CompositeWPF"
    <Grid>
        <ContentControl ps:RegionManager.RegionName="mainReg" />
    </Grid>