Bnaya Eshet : Rx

The 3rd day of the SDP

bnaya — Wed, 08 May 2013 04:17:58 GMT

The 3rd day of the SDP was over and I want to thanks all the attendants.
my yesterday session was about C# 5 async and await.

today I will have a full day tutorial on Rx and TPL Dataflow.
you can download the demo code and also the presentation for second and third days from here (the link is also having the demo code for today's sessions).

anyway, I also want to recommend 2of yesterday session.

if you were attend at the conference you should have an access to the sessions video page.
first I want to recommend Ofir Makmal's session about .NET 4.5 CLR improvements.

Ofir was talking about subjects like:

Multi-core JIT
Managed Profile Guided Optimization
Auto-Ngen
Background Server GC

my favorite subject was Multi-core JIT.

the following is a snippet for Multi-core JITing.

Code Snippet
 static void Main(string[] args)
{
    ProfileOptimization.SetProfileRoot(@"C:\MyApplicationFolder");
    ProfileOptimization.StartProfile("startup.profile");
}
 

my second recommendation is Manu Cohen-Yashar's session about .NET cryptography.
Manu's talk is great for anyone who want to use the .NET cryptography library.

Open House at Microsoft

bnaya — Mon, 14 Jan 2013 07:48:22 GMT

Open House at Microsoft

yesterday I was lecturing at Microsoft about VS 2012, .NET 4.5, async/await, Rx and TPL Dataflow.

there was 90 people attended and I hope that everybody has learn something new.

the code sample for this lecture available here.

SDP 2012 - Day 2

bnaya — Mon, 19 Nov 2012 16:31:00 GMT

SDP 2012 - Day 2

the SDP 2012 conference has day 2 had completed.

I was speaking about async / await, Rx and TPL Dataflow.

you can find the materials for my lecture in here.

I will give the same lecture on Wednesday next week,
there is a chance that some place is still available,
so if you are interesting, you can check it with Sela Marketing.

Rx - Open Source

bnaya — Fri, 09 Nov 2012 22:07:57 GMT

Rx - Open Source

Microsoft had announced the open sourcing of Rx.

it is a great news for the Rx development community.

Microsoft is still committed to the library's quality while accepting external contribution into the Rx code base.
Open sourcing is the new Microsoft strategy and you can see other technology that is moving in that direction (for example ASP.NET).

personally I will consider to move Visual Rx into an official Rx code base.

central code base that is being managed by Microsoft is much better than a separate libraries because when a contribution integrate into the code base it is having the Microsoft team quality stamp and developer can easily adopt it with less worries.

Visual Rx - run as administrator

bnaya — Fri, 26 Oct 2012 22:43:23 GMT

Visual Rx - run as administrator

if you have download the Visual Rx client, make the current version required an administrator privilege.

Visual Rx - Part 6

bnaya — Fri, 26 Oct 2012 01:06:37 GMT

Visual Rx - Part 6

this post is part of the Visual Rx series and it will focus on Visual Rx Viewer Side Extensibility.

this post is dealing with advance topic, you may want to read other post on this series before reading this one.

this post refer to version 2.0.20.622.60 or higher (the extensibility model was simplify at this version).

Visual Rx support extension via plug-ins bundle contract.
a plug-ins bundle contains:

General information about the package (like title and description),
Publisher information (like name, email and web address)
Implementation for different plug-ins capabilities.

plug-ins capabilities can be one or more implementation for the following aspects:

a complete Tab (actually the Grid Tab is a plug-in that you can remove).

It can modify a marble diagram icon.

and even change the representation of single item:

it can also control the entire view of a marble diagram (this way you can control the entire diagram view).

you can see the code of this extension in the source code under the StockExchangeGraphPlugin project.

the last extensibility point is a non visual extensibility.
it can intercept an item and do what ever you want to do with it data (for example send an email or write to a file).

How to get started?

first you have to set a plug-in bundle.
as I said earlier a bundle encapsulate the information about the plug-in and the publisher with a set of plug-in's capabilities.

it is done by inheritance of the PluginBundleBase class which is implementing the IPluginBundle interface (if you want you can inherit the interface directly).

the following is the IPluginBundle definition:

Code Snippet
 public interface IPluginBundle
{
    Guid Id { get; }
    string Title { get; }
    string Description { get; }
    Version Version { get; }
    string PublisherName { get; }
    string PublisherEmail { get; }
    string SupportUrl { get; }
    string ProductUrl { get; }
    ImageSource Icon { get; }
 
    bool HasCustomConfiguration { get; }
    void EditConfiguration();
 
    IItemInterceptionPlugin InterceptionPlugin { get; }
    ILineHeaderImagePlugin LineHeaderImagePlugin { get; }
    IMarblePanelPlugin MarblePanelPlugin { get; }
    IMarbleItemPlugin MarbleItemPlugin { get; }
    ILoadResourcesPlugin LoadResourcesPlugin { get; }
    ITabPlugin TabPlugin { get; }
}
 

as you can see it start with information about the plug-in and the publisher and followed by a set of properties which return interfaces.
those interfaces represent different capabilities.

let start with the most general interface which is helping the plug-in environment rather then actually having a real plug-in functionality.

Load Resources Plugin

the LoadResourcesPlugin property return the ILoadResourcesPlugin interface. this is how you can add resources which can be latter used by other plug-ins implementation (it is very common that your plug-in representation will need resources which is not available within Visual Rx core).

for example you can load WPF ResourceDictionary which will contain anything needed by your plug-in (images, styles, templates, ext.).

Code Snippet
 public class LoadResourcesPlugin : ILoadResourcesPlugin
{
    private readonly string ASSEMBLY_NAME = UIPluginHelper.GetAssemblyName();
 
    public ResourceDictionary[] GetResources()
    {
        Uri uri = UIPluginHelper.GetResourceUri(ASSEMBLY_NAME, "CustomMarble.xaml");
        ResourceDictionary myResource = new ResourceDictionary();
        myResource.Source = uri;
        return new ResourceDictionary[] { myResource };
    }
}
 

as you can see Visual Rx does have some helper method like UIPluginHelper.GetAssemblyName() which can make this task even easier.

Marble Item Plugin

the must useful capability may be the ability to change the item representation based on it's data.
this capability achieved by implementation of the MarbleItemPlugin property which return the IMarbleItemPlugin interface.

the interface is fairly simple and it look as follow:

Code Snippet
 public interface IMarbleItemPlugin
{
    DataTemplate SelectTemplate(MarbleBase item, FrameworkElement element);
}
 

it gets an item and a FrameworkElement and return a DataTemplate.
you should return null if you don't want to change the default template for the current item.

the following snippet was taken from the System.Reactive.Contrib.Monitoring.SamplePlugin which is part of the source code that available for download at codeplex.

Code Snippet
 public class PseudoTemplateSelectorPlugin : IMarbleItemPlugin
{
    public DataTemplate SelectTemplate(MarbleBase item, FrameworkElement element)
    {
        DataTemplate template = null;
        bool startWith = item.Name.ToLower().StartsWith("sec");
        bool isDivBy10 = Convert.ToInt32(item.RawValue) % 10 == 0;
        if (startWith && isDivBy10)
        {
            template = element.FindResource("CustomMarbleNextTemplate") as DataTemplate;
        }
        return template;
    }
}
 

the above implementation is replacing the template for items within marble diagrams that start with "sec" and ignoring items that doesn't divide by 10.
the result will be the fire icon representation for those items:

Line Header Image Plugin

sometime all you want to do is to change the marble diagram icon.
in this case you should use the LineHeaderImagePlugin property which return the ILineHeaderImagePlugin interface.

Code Snippet
 public interface ILineHeaderImagePlugin
{
    ImageSource Convert(string text);
}
 

the text parameter is the name of the marble diagram.
it can be use as follow (taken from the System.Reactive.Contrib.Monitoring.SamplePlugin):

Code Snippet
 public class ImageMapperPlugin : ILineHeaderImagePlugin
{
    private static readonly string ASSEMBLY_NAME = UIPluginHelper.GetAssemblyName();
    private readonly Uri STRESS_IMG = UIPluginHelper.GetPackResourceUri(ASSEMBLY_NAME, "Images/clicknrun.png");
    private readonly Uri SEC_IMG = UIPluginHelper.GetPackResourceUri(ASSEMBLY_NAME, "Images/Clock.png");
 
    public ImageSource Convert(string text)
    {
        text = text.ToLower();
        ImageSource img = null;
        if (text.StartsWith("stress"))
            img = new BitmapImage(STRESS_IMG);
        else if (text.StartsWith("sec"))
            img = new BitmapImage(SEC_IMG);
        return img;
    }
}
 

the above snippet is changing the icons of the stress and sec marble diagrams into a preloaded resource (see LoadResourcesPlugin) and ignore any other diagrams.

the result will be shown as follow:

as you can see the sec icons become a clock, stress become runner while the error icon remain the default one.

I will leave the rest of the plug-in capabilities to a latter post and switch into the plug-in deployment.

Plug-in Deployment

the plug-in are exposed though MEF (when you implement IPluginBundle or inherit PluginBundleBase your code is automatically exported).

the Visual Rx setting let you set the plug-ins discovery and activation.

the following diagram is showing the setting dialog:

the plug-in setting is divided into 2 sections:

discovery section where you can set plug-in folders (both relative and absolute path are supported).
plug-in activation section where you can enable or disable plug-ins.

Future features

the setting dialog will have a plug-in upload and download from a central plug-in repository.

each plug-in will also be able to add a custom setting.

summary

Visual Rx was built for extensibility.
you can use built-in, your own or 3rd party plug-ins which may enhance your monitoring efficiency.

Sample code

the sample code is part of the source code of Visual Rx.
you can use the sample as a starting point but remember that it is only a sample code, some of the sample has a memory leak and other issues.
you should test the code as you would test your own code.

Visual Rx Version 2.0.20622.60

bnaya — Thu, 25 Oct 2012 18:49:13 GMT

Visual Rx Version 2.0.20622.60

I have release a new version of Visual Rx just before my fight to the Build conference.

the version support .Net 4.5 and is having a new plug-in model.

this post will be follow by a post about writing plug-ins for the Visual Rx.

for the client side it is recommended to use the Nuget package (write visualrx into the Nuget search box).

if you want to learn more about .Net 4.5 parallelism, Rx and Tpl Dataflow you can check out the Sela developer practice conference, my session are already sold out, but I believe that a few places is still available for a parallel session by my colleague Noam Sheffer.

Visual Rx - Backward Compatibility

bnaya — Wed, 22 Aug 2012 12:11:18 GMT

Visual Rx - Backward Compatibility

I have just publish a new release of Visual Rx SDK which is targeting Rx versions 1.0.10621 and 1.1.11111.

this release is result of a community request which was added to the Issue Tracker. you can add other request or vote for existing one in here.

this release is also available on the NuGet Gallery.
the NuGet page for version 1.0.10621 is available for download in here and the page for version 1.1.11111 is available in here.

the same Visual Rx viewer is compatible with all versions.

Visual Rx - Part 5

bnaya — Sun, 19 Aug 2012 11:32:45 GMT

Visual Rx - Part 5

this post is part of the Visual Rx series and it will focus on controlling the Visual Rx publication on the coding part.

you can see this series TOC in here.

I was talking about enabling / disabling the Visual Rx publication (through the predefine publication channel), using the VirtualRxSettings.Enable property, in part 1 of this series.

Enable / Disable

let look at a Enable / Disable snippet:

Code Snippet
 Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
    VisualRxWcfDiscoveryProxy.Create());
 
VisualRxInitResult infos = info.Result;
Trace.WriteLine(infos);
 
var xs = Observable.Interval(TimeSpan.FromSeconds(0.5))
    .Take(10);
xs = xs.Monitor("Enable / Disable", 1);
xs.Subscribe(
    v => VisualRxSettings.Enable = v < 3 || v > 6);
 

lines 1,2 define the publication channel to Wcf Discovery.
lines 4,5 is waiting to the channel construction completion and log it to the Trace.
lines 7,8 create a datum stream.
line 9 is monitoring the stream.

line 11, within the Subscribe method I toggle the publication (on / off), you should be aware that the current value has been already sent by the time that the toggling logic took place, which mean that the toggling will affect the next value.

the toggle state for values 4-7 is off therefore the Visual Rx Viewer will show the stream as follow:

enabling or disabling the publication is totally at your control.

Partial publication

real-life scenario may demand a finer control over the datum's publication.
you may want to enable or disable a partial publication based on the datum's stream name, keywords, channel key or even the datum itself.

this can be achieved by using the VisualRxSetting.AddFilter API.

the following snippet is showing how to split different streams into different channels:

Code Snippet
 Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
    VisualRxWcfDiscoveryProxy.Create(),
    VisualRxWcfQueuedProxy.Create(),
    VisualRxTraceSourceProxy.Create());
 
VisualRxSettings.AddFilter((marble, channelKey) =>
    {
        bool enable = false;
        switch (marble.Name)
        {
            case "Stream A":
                enable = channelKey == VisualRxWcfDiscoveryProxy.KIND;
                break;
            case "Stream B":
                enable = channelKey == VisualRxTraceSourceProxy.KIND;
                break;
            default:
                enable = channelKey == VisualRxWcfQueuedProxy.KIND;
                break;
        }
        return enable;
    });
 
VisualRxInitResult infos = info.Result;
Trace.WriteLine(infos);
 
var xs = Observable.Interval(TimeSpan.FromSeconds(0.5))
   .Take(10);
xs = xs.Monitor("Stream A", 1);
var ys = Observable.Interval(TimeSpan.FromSeconds(1))
   .Take(10);
ys = ys.Monitor("Stream B", 2);
 
xs.Subscribe();
ys.Subscribe();
 

the result will be that the Visual Rx Viewer will monitor "Stream A" while "Stream B" will be log to the TraceSource under the "Visual Rx" setting.

the Viewer will present the following:

while the TraceSource output is going to the default listener (therefore it will go to the output window on debugging).
the output will look as follow:

Code Snippet
 <MarbleBase z:Id="i1" i:type="MarbleNext">
  <BinaryValue i:nil="true" />
  <DateCreatedUtc>2012-08-13T12:07:31.8190414Z</DateCreatedUtc>
  <FormattedValue>0</FormattedValue>
  <IndexOrder>2</IndexOrder>
  <Keywords />
  <Kind>OnNext</Kind>
  <MachineName i:nil="true" />
  <Name>Stream B</Name>
  <Offset>PT0.4960332S</Offset>
  <Options>None</Options>
  <StringValue>0</StringValue>
  <ThreadId>14</ThreadId>
</MarbleBase>
 
...
 
VisualRx Information: 0 :
<MarbleBase z:Id="i1" i:type="MarbleComplete">
  <BinaryValue i:nil="true" />
  <DateCreatedUtc>2012-08-13T12:07:40.9455634Z</DateCreatedUtc>
  <FormattedValue i:nil="true" />
  <IndexOrder>2</IndexOrder>
  <Keywords />
  <Kind>OnCompleted</Kind>
  <MachineName i:nil="true" />
  <Name>Stream B</Name>
  <Offset>PT9.6220625S</Offset>
  <Options>None</Options>
  <StringValue i:nil="true" />
  <ThreadId>14</ThreadId>
</MarbleBase>
 

if you want you can even filter by the actual data,
for example you can omit item that doesn't relevant to the current monitoring session.

the following code omit publication for datum that doesn't belong to the "Extreme" category.

first I will use the following declarations:

Code Snippet
 public enum Category
{ 
    BallGame,
    Snow,
    Extreme,
    Water,
    Weapon
}
 
public class Item
{
    public Item(long value)
    {
        Value = value;
        this.Category = (Category)(value % 5);
    }
    public long Value { get; private set; }
    public Category Category { get; private set; }
}
 

the following snippet does the filtering:

Code Snippet
 Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
    VisualRxWcfDiscoveryProxy.Create());
 
VisualRxSettings.AddFilter((marble, channelKey) =>
    {
        Item item = marble.RawValue as Item;
        if (item != null && marble.Name == "Items")
            return item.Category == Category.Extreme;
        return true; // it is best practice to return true 
                     // because this way you don't 
                     // interfere with other filters
    });
 
VisualRxInitResult infos = info.Result;
Trace.WriteLine(infos);
 
var xs = Observable.Interval(TimeSpan.FromSeconds(0.5))
    .Select(v => new Item(v))
   .Take(10);
xs = xs.Monitor("Items", 1);
xs = xs.Monitor("Other monitoring", 1); // actually monitor the same aspect
 
xs.Subscribe();
 

focusing on the filtering at lines 4-12, you can see that I filtered out datum that belong to the "Items" monitor and doesn't have the "Extreme" category.
in order to show the differences in the Visual Rx Viewer,
I was monitoring twice (lines 20, 21) once with the "Items" name which will be filtered and the second time is actually monitoring the same data but this time with "Other monitoring" name.
the result will be as follow:

you can download the source code and check the tests App.Config file to see how can you set the configuration for the TraceSource channel.

the following is one option where most of the listener are disabled:

Code Snippet
 <system.diagnostics>
  <sources>
    <!-- Visual Rx TraceSource publication -->
    <source name="VisualRx" switchName="defaultswitch">
      <listeners>
        <!--<remove name="Default"/>-->
        <!--<add name="consoleListener" />-->
        <!--<add name="xmlFileMonitorListener"/>-->
        <!--<add name="customListener"/>-->
      </listeners>
    </source>
  </sources>
  <switches>
    <!-- Critical, Error, Warning, Information, Verbose
         ActivityTracing: Start, Stop, Suspend, Resume, Transfer
    -->
    <add name="defaultswitch" value="Verbose"/>
  </switches>
  <sharedListeners>
    <add name="customListener"
        type="System.Reactive.Contrib.TestMonitor.CustomTraceListener,System.Reactive.Contrib.TestMonitor"
         traceOutputOptions="ProcessId, DateTime"/>
    <add name="consoleListener"
        type="System.Diagnostics.ConsoleTraceListener"
         traceOutputOptions="ProcessId, DateTime"/>
    <add name="eventlogListener"
        type="System.Diagnostics.EventLogTraceListener"
        initializeData="Application">
      <filter type="System.Diagnostics.EventTypeFilter"
        initializeData="Error"/>
    </add>
    <add name="textFileListener"
        type="System.Diagnostics.TextWriterTraceListener"
        initializeData="RxContrib.log" />
    <add name="xmlFileMonitorListener"
        type="System.Diagnostics.XmlWriterTraceListener"
        initializeData="Monitor.xml" />
    <add name="defaultListener"
        type="System.Diagnostics.DefaultTraceListener" traceOutputOptions="ProcessId"/>
    <!--<add name="nlog" type="NLog.NLogTraceListener, NLog" />-->
  </sharedListeners>
</system.diagnostics>
 

if you want to filter by keyword just ask about the marble.Keywords:

Code Snippet
 VisualRxSettings.AddFilter((marble, channelKey) =>
    marble.Keywords.Contains("Key1"));
 

the following snippet:

Code Snippet
 Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
    VisualRxWcfDiscoveryProxy.Create());
 
VisualRxSettings.AddFilter((marble, channelKey) =>
    marble.Keywords.Contains("Key1"));
 
VisualRxInitResult infos = info.Result;
Trace.WriteLine(infos);
 
var xs = Observable.Interval(TimeSpan.FromSeconds(0.5))
   .Take(10);
xs = xs.Monitor("Stream A", 1, "Key1", "Key3", "Key4");
var ys = Observable.Interval(TimeSpan.FromSeconds(1))
   .Take(10);
ys = ys.Monitor("Stream B", 2, "Key2", "Key3");
 
xs.Subscribe();
ys.Subscribe();
 

will be present in the viewer where only the "Stream A" will be shown on the Visual Rx Viewer because "Stream B" doesn't have the "Key1" keyword:

Clear Filters

at any time you can wipe-out all the filters by using the ClearFilters API (this one can help for testability).

Code Snippet
       VirtualRxSettings.ClearFilters(); 
     

Summary

the filtering capability can keep a real-life system performance unaffected by data that doesn't relevant to the current monitoring session.

Future Feature

one of the feature that I want to add in the future is a remote tool that will control simple filtering like which keywords is actively monitored.

Visual Rx - Part 4

bnaya — Wed, 15 Aug 2012 23:40:12 GMT

Visual Rx - Part 4

this post is the 4th of the Visual Rx post series.
this post will focus on the communication channel settings.

you can see this series TOC in here.

at the time I'm writing this post there is 4 available communication channel (out of the box).

Wcf discovery - will try to dynamically locate the end point of the viewer service (or any other discoverable service which apply to the Visual Rx contract).
this is the easiest proxy and it is recommended to local network scenarios.
Wcf queued - which submit the datum though MSMQ channel (the default viewer is compatible with this channel).
Trace Source - publish the data though the standard .NET Trace Source. using the "VisualRx" trace source name (for internal messages it use "VisualRx.Log")
the app.config setting will be discuss on latter posts.
Wcf Fixed Address - is a new one, it is let you specify a Wcf Binding and End-Point (and gives you the full flexibility of the Wcf library).

Visual Rx Viewer side:

at the viewer side you simply dealing with a traditional Wcf application configuration (for setting a Wcf channels).
from version 2.0.20622.10 the default application setting is enabling Wcf discovery over named-pipe and Wcf fixed address over basic http.
other setting like MSMQ and Wcf discovery over TCP is remarked.
you can see the full application setting in here.
but the part that you are rely care about is:

Code Snippet
 <!-- used by VisualRxWcfQueuedProxy -->
<!--<endpoint address="net.msmq://localhost/private/VisualRx" binding="netMsmqBinding"
          bindingConfiguration="netMsmq"
          contract="System.Reactive.Contrib.Monitoring.Contracts.IVisualRxServiceAdapter" />-->
<!-- used by VisualRxWcfFixedAddressProxy -->
<endpoint address="VisualRx" binding="basicHttpBinding"
          contract="System.Reactive.Contrib.Monitoring.Contracts.IVisualRxServiceAdapter" />
 
<!-- used by VisualRxWcfDiscoveryProxy -->
<!--<endpoint name="tcpMonitor" address="Monitor"
          binding="netTcpBinding"
          contract="System.Reactive.Contrib.Monitoring.Contracts.IVisualRxServiceAdapter"/>-->
<!-- used by VisualRxWcfDiscoveryProxy -->
<endpoint name="namepipeMonitor" address="Monitor" 
          binding="netNamedPipeBinding" 
          contract="System.Reactive.Contrib.Monitoring.Contracts.IVisualRxServiceAdapter"/>
 

removing the remark from lines 2-4 will enable MSMQ channel (make sure that MSMQ is install on the machine).

lines 6-8 is the Wcf fixed address over http.

lines 10-16 is the Wcf discovery endpoints (the TCP endpoint is remarked).

actually this is all you have to do from a viewer perspective.

Visual Rx Code side

at the code side you should choose which channel proxy do you want to use at the setting initialization of VisualRxSetting.

you can choose multiple channels at once, but you have to make sure that you don't create a duplication in terms of a viewer will get the same result, from multiple channels, which will lead to duplication of data at the viewer side.

the following snippet is enabling all 3 Wcf channel types:

Code Snippet
 Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
    VisualRxWcfDiscoveryProxy.Create(),
    VisualRxWcfFixedAddressProxy.CreateDefaultHttp(Environment.MachineName),
    VisualRxWcfQueuedProxy.Create());
 

each channel creation is having a few overloads.
the sample above is using the most simplistic ones.

Wcf Discovery

the Wcf discovery is the most easy channel, because you don't have to configure a fix address and the proxy will locate the viewer dynamically.

the downside of this channel is that it is limit to a local network boundary and if that network has a vast number of machine the discover process could be slow and heavy.

you can set the timeout for discovery both in the code and the viewer sides.

the viewer setting is take place within the application configuration at the following section:

Code Snippet
  
  <standardEndpoints>
    <udpDiscoveryEndpoint>
      <!-- modify the udpDiscoveryEndpoint -->
      <standardEndpoint name="adhocDiscoveryEndpointConfiguration" maxResponseDelay="00:00:00.001"/>
    </udpDiscoveryEndpoint>
  </standardEndpoints>
</system.serviceModel>
 

it is define the maximum time that the viewer will randomly delay its answer (this number has to be lower then the proxy timeout setting which will be discuss in the next section), Wcf is using this setting in order to reduce the network pressure.

at the proxy side you can use the default creation which is using a 3 seconds timeout and retry to discover new viewer at a 30 minutes interval.

but you can pass a VisualRxWcfDiscoverySettings parameter and set those setting manually.

Code Snippet
 var setting = new VisualRxWcfDiscoverySettings
{
     DiscoveryTimeoutSeconds = 2,
     RediscoverIntervalMinutes = 15
};
VisualRxSettings.CollectMachineInfo = true;
Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
     VisualRxWcfDiscoveryProxy.Create(setting));
 

the Wcf discovery proxy will try to locate both named-pipe and TCP end-points (no need to any special setting for that).

Wcf Fixed End-Point

the Wcf fixed end-point is having 3 creation options:

Code Snippet
 VisualRxWcfFixedAddressProxy CreateDefaultHttp(string host);
VisualRxWcfFixedAddressProxy Create(Binding binding, string uri);
VisualRxWcfFixedAddressProxy Create(Binding binding, EndpointAddress endpoint);
 

CreateDefaultHttp is the most simple one and it operate over a basic http binding. all you have to do is to supply the Machine name or IP address of the viewer machine.
Create is having 2 overloads which is getting any Wcf binding and the end-point address of the viewer. it is very flexible in terms of the binding, it can take advantage of any binding setting (like security, session or even a restful like webHttpBinding).

MSMQ

the Wcf Queued binding is operating over MSMQ and it is having the following overloads:

Code Snippet
 VisualRxWcfQueuedProxy Create();
VisualRxWcfQueuedProxy Create(TimeSpan timeToLive);
VisualRxWcfQueuedProxy Create(NetMsmqBinding binding);
VisualRxWcfQueuedProxy Create(NetMsmqBinding binding, string queuePath);
 

the default creation is using a private queue named VisualRx and time to live of 30 minutes (time to live define the period that the message will live in the queue before it will become stale).
the second overload let you control the time to live setting.
the 3rd one is let you a full control over the binding.
and the 4th is also let you control over the queue path.

that all for this post the TraceSource will be discuss in further posts.

Summary

communication between your run-time streams and the viewer is fairly flexible and you actually can add your own custom proxy by implementing the IVisualRxProxy interface.

Visual Rx - TOC

bnaya — Sun, 12 Aug 2012 22:43:34 GMT

Visual Rx - TOC

Visual Rx is a bundle of API and a Viewer which can monitor and visualize Rx datum stream (at run-time).

the current post in the Visual Rx series includes:

Part 1: Code side.
Part 2: Viewer side.
Part 3: How to start using Visual Rx.
Part 4: Setting the communication channels.
Part 5: publication filtering.

you can vote for feature at the CodePlex site under the Issue Tracker tab.

you can download the SDK and the Viewer from here.

Visual Rx - Part 3

bnaya — Sun, 12 Aug 2012 22:37:55 GMT

Visual Rx - Part 3

this post is part of the Visual Rx series and it will focus on how to start working with Visual Rx.

you can see this series TOC in here.

as I was mention in previous posts Visual Rx is divided into 2 parts:

SDK - which is the code side on which you define the VisualRxSetting and use Monitor extension method.
Viewer - which is listening to the SDK communication channel and visualize the Rx stream.

both parts available at the download page of Visual Rx on CodePlex.

the SDK is also available at NuGet.
the NuGet package include the Rx dlls and the the Visual Rx dlls.

you can get it either directly from the NuGet site:

or from visual studio add reference (after you will install the NuGet Visual Studio extension which will add "Manage NuGet Packages"):

the Viewer is also available in 2 flavor:

XCopy - which is a zip that contain the viewer exe + dependencies.
MSI - a setup file which will install the viewer on your machine.
in the future it may be also available though clickonce.

Where can I get sample code?

you can download the source code from CodePlex and
run the MS-Tests under the System.Reactive.Contrib.Monitoring.UnitTests project (most test is under the Demo folder).

finally you can vote for feature at the CodePlex site under the Issue Tracker tab.

Summary

I hope that you find the Visual Rx useful.
please vote for feature so I will know what does the community value most.

on later post will discuss more advance scenarios.

Visual Rx - Part 2

bnaya — Sun, 12 Aug 2012 22:36:30 GMT

Visual Rx - Part 2

this is the second post in the Visual Rx series.
this post will focus on the viewer side.

you can see this series TOC in here.

as we saw in the previous post the main view of the Visual Rx viewer is using a tab control (the default tab is the "All" tab) out of the box the viewer is also having a grid tab, additional tab will be added per keyword.
(the grid is actually a plugin and it can be remove by deleting it from the relative Plugins folder).

all tabs except of the grid tab is presenting the data using a marble diagrams.

the Visual Rx viewer is a plug-able application and you can have a custom tabs if you want to present the data in a different way (more on this topic in latter posts).

Application Level

on the top left you can see 3 buttons, Pause, Clear and About.

the Clear button will clear the current presenting data on all tabs.

Pause will freeze the view (without loosing data).
it is essential for analyzing a fast stream at runtime.
this is a toggle button which will change between Pause and Resume. the data continue to be collected while the viewer is in a pause mode (even those this data doesn't display).
Resume will immediately display all the data that was collected during the pause phase and resume to display a new data.

Single Tab level

under each tab you can find the "show tab actions" link (marked by yellow) which will expand a tab's level actions (those actions will affect the only current tab).

clicking on this link will expand / collapse the tab's level actions.

the tab's level actions include the following functionalities:

Clear will clear the diagrams from the current tab (while preserving other tabs diagrams).
Global Time, Private Time and Sequence are different ways of ordering the marbles on the marble line:
- Global time will show the marble on a timeline with a correlation between all marble diagrams within the current tab (this is useful when the diagram streams are related to each other, like the case of Rx combinators like Merge).
- Private time will conduct the diagram timeline on a single diagram base.
- Sequence will present the marbles in correlation to their creation order regardless of the creation time (it is useful for operators like Range or when multiple marble were produce at the same time).

on the previous post we was looking at the following code:

Code Snippet
       var inputA1 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Take(7); 
inputA1 = inputA1.Monitor("Input A1", 1, "ScenarioA", "Input"); 
var inputA2 = Observable.Interval(TimeSpan.FromSeconds(2)).Take(5); 
inputA2 = inputA2.Monitor("Input A2", 2, "ScenarioA", "Input"); 
var outputA = Observable.Merge(inputA1, inputA2); 
outputA = outputA.Monitor("Output A", 3, "ScenarioA", "Output"); 
var inputB = Observable.Interval(TimeSpan.FromSeconds(1)).Take(10); 
inputB = inputB.Monitor("Input B", 4, "ScenarioB", "Input"); 
var outputB = from item in inputB 
group item by item % 3 into g 
select g; 
outputB = outputB.MonitorGroup("Output B", 5, "ScenarioB", "Output"); 
outputA.Subscribe(); 
outputB.Subscribe(g => g.Subscribe()); 
     

the default Global Time view will present the data as follow:

as you can see the diagrams time line is in correlation with each other, Private Time will shift the diagram marble to the left:

still the time ratio were preserved on per diagram based.

the Sequence option will ignore the time ratio and present the data in a simple sequence:

let resume the the rest of the actions:

Scroll To End is a toggle button which will automatically keep the horizontal scrolling bar attached to the right end, which mean that the newest data will always be visible
the Track-bar is a zooming bar
and though the combo-box you can define the presented time scale (fast stream time scale may better present in millisecond units).

Marble diagram level

at the marble diagram level you can find a popup action bar which response to click on the marble title.

the action includes:

Refresh - which refresh the marble view.
Global Time, Private Time and Sequence on a single marble level.

the last functionality is collapse ability.
clicking on the marble collapser will collapse the marble:

Summary

the Visual Rx viewer is a reach viewer which was design to support Rx dataflow analysis.
you can take it further by using a custom plug-ins but this will be discuss on latter posts.

you can download the Viewer and the SDK from codeplex and the SDK can also be download from NuGet (search for VisualRx).

you can vote for feature at the CodePlex site under the Issue Tracker tab.

Visual Rx - Part 1

bnaya — Mon, 30 Jul 2012 22:15:05 GMT

Visual Rx - Part 1

this post is the first of series about Visual Rx.
this post series also stand as the official help of Visual Rx.

you can see this series TOC in here.

Visual Rx is a bundle of API and a Viewer which can monitor and visualize Rx datum stream (at run-time).

one of the challenge of any Rx solution is reasoning about the datum flow at run-time, the Visual Rx bundle is a perfect solution for this problem.

Visual Rx was design for minimal performance penalty, which make it a perfect candidate for production time monitoring.

Supported functionality :

Activation: monitoring can be enable and disable at runtime (at the code side).
Transport: different transport strategy can be apply (including WCF, Queue, and Trace Source).
Plug-ins: custom behaviors can be hook on different aspect of the monitoring or UI functionality, including a custom transport channel, UI painting and more (more about this one on letter post).

the architecture of the Visual Rx bundle is separate into two main components groups (those groups is having a loosely couple connection though a common contract and DTOs):

Coding extension which is responsible to the settings and the transmission of the datum information on the Rx code side.
Viewer component which is responsible to visualize the Rx datum stream.

Coding extension

Visual Rx settings

the Visual Rx setting is responsible to setup the transmission channels which will be used to submit the Rx datum stream toward the viewer (or any other end point).

out of the box Visual Rx come with the followings built-in channels:

Wcf discovery - will try to locate the end point of the viewer service (or any other discoverable service which apply to the Visual Rx contract).
this is the easiest proxy and it is recommended to local network scenarios.
Wcf queued - which submit the datum though MSMQ channel (the default viewer is compatible with this channel).
Trace Source - publish the data though the standard .NET Trace Source. using the "VisualRx" trace source name (for internal messages it use "VisualRx.Log").
the app.config setting will be discuss on latter posts.

Initialize

you can define the channel by using VirtualRxSettings Initialize static method.

Code Snippet
 Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
    VisualRxWcfDiscoveryProxy.Create(),
    VisualRxWcfQueuedProxy.Create(),
    VisualRxTraceSourceProxy.Create());
 
VisualRxInitResult infos = info.Result;
Trace.WriteLine(infos);
 

lines 1-4 setup the transmission channels to all 3 built-in channels.
it is a non blocking API which return a Task which you can either wait until the channels construction or use a ContinueWith / await for a callback operations (this sample is blocking until the channels are ready).

through the Task result you can log the channels state (lines 6-8).

this is the Initialize method signature:

Code Snippet
 Task<VisualRxInitResult> Initialize(
            params IVisualRxProxy[] proxies)
 

as you can see, you can initialize any number of proxies including a custom proxies that implement IVisualRxProxy (the proxy responsibility is to send the data though a channel).
you can call the Initialize method multiple time, each call will override the settings of the previous one.

Enable

at any time you can turn on / off the profiling operation by assigning the Enable property.

Code Snippet
 VirtualRxSettings.Enable = false;
 

this is one of the way you can avoid performance penalty. you can activate the data publication only when you need to actively monitor the it.
you can expose this functionality by any standard service (in letter version I will supply a tool for remote activation and deactivate the publication).

other advance setting will be discuss on latter posts.

Monitoring an IObservable<T>

for any observable you can use a Monitor extension methods in order to publish the the observable's datum information though a predefine channels.

Code Snippet
 var xs = Observable.Interval(TimeSpan.FromSeconds(1)).Take(5);
xs = xs.Monitor("Interval 1 second", 1);
xs.Subscribe();
 

the minimal overload of the Monitor method is getting 2 parameters

marble diagram name
marble diagram index order (which is handle by the viewer).

IMPORTANT: you should define the transmission channel (see Visual Rx Setting) in order to actually send the datum through the publication channel (it is you're responsibility to define the publication channel).

for example the following code snippet is setting a Wcf Discovery proxy in order of monitoring 2 observables:

Code Snippet
 Task<VisualRxInitResult> info = VisualRxSettings.Initialize(
    VisualRxWcfDiscoveryProxy.Create(),
    VisualRxWcfQueuedProxy.Create(),
    VisualRxTraceSourceProxy.Create());
 
VisualRxInitResult infos = info.Result;
Trace.WriteLine(infos);
 
var xs = Observable.Interval(TimeSpan.FromSeconds(1)).Take(5);
xs = xs.Monitor("Interval 1 second", 1);
var ys = Observable.Interval(TimeSpan.FromSeconds(0.5)).Take(10);
ys = ys.Monitor("Interval 0.5 second", 2);
  
xs.Subscribe();
ys.Subscribe();
 

the above code will be present on the VisualRx viewer as follow:

the viewer is having 2 tabs (All and Grid),
the All tab is the selected tab which present the data stream as a sequence of marbles, the distance between marble present the duration.
as you can see the 1 second interval is less intensive then the 0.5 second interval.

the Grid representation will present each datum as a line in a grid:

in real-life you may have more then a few Rx data streams which may have inter relation (you may want to have separate tab for each of those sub groups).

for this scenario Visual Rx support a keywords.

the following code snippet is setting different keywords to different datum streams:

Code Snippet
 var inputA1 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Take(7);
inputA1 = inputA1.Monitor("Input A1", 1, "ScenarioA", "Input");
var inputA2 = Observable.Interval(TimeSpan.FromSeconds(2)).Take(5);
inputA2 = inputA2.Monitor("Input A2", 2, "ScenarioA", "Input");
var outputA = Observable.Merge(inputA1, inputA2);
outputA = outputA.Monitor("Output A", 3, "ScenarioA", "Output");
 
var inputB = Observable.Interval(TimeSpan.FromSeconds(1)).Take(10);
inputB = inputB.Monitor("Input B", 4, "ScenarioB", "Input");
var outputB = from item in inputB
              group item by item % 3 into g
              select g;
outputB = outputB.MonitorGroup("Output B", 5, "ScenarioB", "Output");
 
outputA.Subscribe();
outputB.Subscribe(g => g.Subscribe());
 

line 1-6 is constructing the first scenario (Scenario A) where 2 input stream are merged into a single output stream (Output A).

you can see at line 2,4 that the monitoring is specify both "ScenarioA" and "Input" keywords,
while on line 6, the monitoring is specify "ScenarioA" and "Output" keywords.

lines 8-13 is constructing the second scenario (Scenario B) which is having a single input stream and group by manipulation for the output stream.

as you can see at line 9 the keywords for the input stream is"ScenarioB" and "Input" and
at line 13 you can see a group monitoring using the MonitorGroup extension method, which is using the "ScenarioB" and "Output" keywords.
(shortly we will survey the output of the MonitorGroup extension).

summarizing, we was attaching different keywords at different stage on the pipeline of scenario A and B.
as you will see shortly, it will help us to stay in focus about each scenario, the inputs or the outputs streams.

the Visual Rx viewer will present it as follow:

you can easily see the monitoring of the first scenario in the first 3 marble's lines, while the second scenario is present in the last 5 marble's line.

as you can record, we was using the MonitorGroup extension method. using this kind of monitoring results with one marble's line for the group's keys (the "Output B (keys)" line).
and a marble's line for each group output (because our group key was item % 3 there is only 3 group's lines Output B 1,2,3).

but in real life monitoring this tab can become a bit over packed with to much marble's line.

by using the keywords you can focus on different aspects of overall execution like each scenario, input or outputs.

you can see that except of the Grid and the All tabs the viewer has create a new tab for each keyword.

for example: the Scenario B tab will show the marble's line which was attached to the "ScenarioB" keyword:

while the Output tab will focus on the "Output" keyword:

this can really help us to stay in focus on what we actually monitoring.

More extensions:

just like the MonitorGroup which can represent a group by monitoring, there is a MonitorMany that can monitor a IObservable<IObservable<T>> (which is the result of the Window operator).
in the future I may add other monitoring operators.

Categories:

another future API will be the ability to attach a Category which will act as a scope where the Viewer can be subscribe to specific categories.
this feature will enable to open multiple viewers for different monitoring scenario.
this feature is still on the design table and in general you can vote for feature in here.

Summary

this is part one of the series which give you the basics instruction for monitoring a data stream using the Monitor extension.

you can download the Viewer and the SDK from codeplex and the SDK can also be download from NuGet (search for VisualRx).

you can vote for feature at the CodePlex site under the Issue Tracker tab.

Rx - Join

bnaya — Thu, 05 Apr 2012 00:16:56 GMT

Rx - Join

this post is an advance one so you haven't yet master the Rx basic practice you might want to start in here.

the post will focus on the Join and GroupJoin operators and we will try to get a solid understanding of what does it really means to join multiple data streams.

the Rx notion of join is quit different than the join we has used to know in SQL queries.

while the SQL query's join refer to data equality (relationship and foreign keys), the Rx's join is all about coexisting.

actually Rx is much closer to the real life notion of join.
when you meeting someone (online or in the real world) its actually mean that both of you are doing the activity at the same time.

so speaking of Rx Join, you must do that conceptual switching from data into coexistence.

because Rx's join is all about coexistence the actual data type doesn't matter any more.
you can declare a join between unrelated data type, because the data type has nothing to do with the Rx's concept of join.

for example:

assuming you have an algorithm which can define a persons mood based on it tweets an other social media text. you can join this moods stream with a weather stream and get a result stream of correlations' between weather and people's moods.

the weather time window duration (period) is from the time it was pushed until the next weather is pushed. the mood period is a point in time (period = zero).

whenever a mood coexist within a weather period the result stream projected a combined weather/mood datum.

the code snippet for scenario is something like the following code.

Code Snippet
 IObservable<Weather> ws = WeatherStream;
IObservable<Mood> ms = MoodStream;
 
IObservable<Tuple<Weather,Mood>> result =
    ws.Join(
            ms, // the join with stream 
            w => ws, // closing weather period
            m => Observable.Empty<Unit>(), // closing mood
            (w /* Weather */, m /* Mood */) => Tuple.Create(w, m));
 

lets look at the join syntax.

at line 5 we use the Join extension method on the weather stream.

the first parameter (line 6) is the mood stream which we joined with.

now we have to define when the time period window of the weather datum will end.
and this is what the second parameter (at line 7) is all about.
it is a Func that get a weather as parameter and return a stream which will close the weather period upon it next projection (OnNext or OnComplete).

because the weather stream is a hot stream the code above will close the weather period whenever a new weather will be project.
it is a one time event end it will work equally if the parameter will be w => ws.Take(1).

the 3rd parameter dictate the mood datum period,
in our case the mood is a point event which has no period, therefore the parameter return Empty stream (which is a stream that immediately complete).

the last parameter is the select operator which define
the projected data type, in our case we return a Tuple of a weather and a mood.

now we can take a breath.

the API is a bit complex, mainly because of the closing stream's parameters, but it is very powerful.

actually the closing stream concept is not unique to the Join operator, there are many other operator like Buffer, Window and GroupByUntil, which is having an overloads that get a closing stream.

the closing stream concept enable a join window notion that are more complex than a simple TimeSpan variable.

for example the closing stream can react to temperature, tweets, sound volume or any other event stream.

Group Join

actually join has another flavor which is the GroupJoin operator. it is very similar to the join but it has a different projection.

GroupJoin is grouping the secondary stream intersection as an IObservable combined with the primary stream datum.

for example:

Code Snippet
 IObservable<Tuple<Weather, IObservable<Mood>>> result =
    ws.GroupJoin(
        ms, // the join with stream                       
        // a stream which close the weather period
        value => ws,
        // a stream which close the mood period (immediate closeing)
        value => Observable.Empty<Unit>(),  // A is a point event which will catch within the main window (of A)
        // projection of a result 
        (w, observableOfMoods) =>Tuple.Create(w, observableOfMoods));
 

is is all similar to the Join operator until we goes into the projection at line 9.

it is a projection of the weather value and IObservable<Mood>, which mean that any weather datum will immediately project (as a new group) and mood datum will constantly be pushed into it's related IObservable as long as the weather life-time doesn't disposed by the closing stream.

the marble diagram of it would seem like the following diagram:

the GroupJoin operator can also be written in a pure LINQ syntax, it is goes like the following snippet:

Code Snippet
 IObservable<Tuple<Weather, IObservable<Mood>>> result =
    from w in ws
    join m in ms                  
        on ws // a stream which close the weather period
        // a stream which close the mood period (immediate closeing)
        equals Observable.Empty<Unit>()
        into g // the IObservable<Mood> which related to the current weather
    select Tuple.Create(w, g);
 

you can find the closing stream of the weather at line 4
and the moods closing stream at line 6.

Summary

the Join concept is a very powerful one but you have to do a mental switching from data correlation into coexistence correlation.

it is really worth the learning curve effort because it is bringing a hole new capabilities to the table which is not trivial at all.

Bnaya Eshet : Rx

The 3rd day of the SDP

Open House at Microsoft

Open House at Microsoft

SDP 2012 - Day 2

SDP 2012 - Day 2

Rx - Open Source

Rx - Open Source

Visual Rx - run as administrator

Visual Rx - run as administrator

Visual Rx - Part 6

Visual Rx - Part 6

How to get started?

Load Resources Plugin

Marble Item Plugin

Line Header Image Plugin

Plug-in Deployment

Future features

summary

Sample code

Visual Rx Version 2.0.20622.60

Visual Rx Version 2.0.20622.60

Visual Rx - Backward Compatibility

Visual Rx - Backward Compatibility

Visual Rx - Part 5

Visual Rx - Part 5

Enable / Disable

Partial publication

real-life scenario may demand a finer control over the datum's publication. you may want to enable or disable a partial publication based on the datum's stream name, keywords, channel key or even the datum itself.

Clear Filters

Summary

Future Feature

Visual Rx - Part 4

Visual Rx - Part 4

Visual Rx Viewer side:

Visual Rx Code side

Wcf Discovery

Wcf Fixed End-Point

MSMQ

Summary

Visual Rx - TOC

Visual Rx - TOC

Visual Rx - Part 3

Visual Rx - Part 3

Where can I get sample code?

More

Summary

Visual Rx - Part 2

Visual Rx - Part 2

Application Level

Single Tab level

Marble diagram level

Summary

Visual Rx - Part 1

Visual Rx - Part 1

Supported functionality :

Coding extension

Visual Rx settings

Initialize

Enable

Monitoring an IObservable<T>

More extensions:

Categories:

Summary

Rx - Join

Rx - Join

Group Join

Summary

real-life scenario may demand a finer control over the datum's publication.
you may want to enable or disable a partial publication based on the datum's stream name, keywords, channel key or even the datum itself.