TCP Qbservable Provider

1.0 Alpha

  • Targets the .NET Framework 4.5 Beta.
  • Written entirely in C# 5 and VS 11 Beta.
  • Depends on Rx 2.0 Beta and Rxx 2.0 Beta, both of which are included in the download.
    • (Note that Rxx 2.0 Beta is not yet officially released and will most likely change).

Overview

The TCP Qbservable Provider library enables you to easily expose IQbservable<T> services over TCP. When a client subscribes, its query is serialized to the server as an expression tree and then executed on the server.

For more in-depth information, see this series of blog posts.

Warning: This provider allows clients to execute arbitrary code on your server.
This is only an alpha release, so security hasn't been fully considered. Do not expose a Qbservable TCP service on a public server or client without taking the necessary precautions to secure them first.

Features

  • Simple server factory methods for hosting a Qbservable TCP service.
    • Supports hosting any IObservable<T> query as a TCP service (hot or cold).
    • Supports hosting any IQbservable<T> query as a TCP service.
  • Simple client factory methods for acquiring a Qbservable TCP service.
    • You must only specify the end point address and the expected return type. The result is an IQbservable<T> that you can query and Subscribe.
    • All Qbservable Rx operators are supported.
  • Automatically serialized Expression trees.
    • Dynamic expressions and debug info expressions are not supported. All other types of expressions are supported.
  • Automatically serialized anonymous types.
  • Immediate evaluation of local members and closures (optional; default behavior)
    • Compiler-generated methods are executed locally and replaced with their return values before the expression is transmitted to the server. This includes iterator blocks, which are serialized as List<T>.
    • Evaluation assumes that local methods are never executed for their side-effects. Actions (void-returning methods) cause an exception. Do not depend upon the order in which members are invoked.
  • Full duplex communication (optional; default behavior for IObservable<T> closures)
    • Must opt-in on server.
    • May opt-in on client for full duplex communication of all local members; automatic for IObservable<T> closures.
    • Duplex communication automatically supports iterator blocks.
  • Designed with extensibility in mind; e.g., supports custom Qbservable service providers, protocols and sinks.

Simple Examples

The following example creates a cold observable sequence that generates a new notification every second and exposes it as an IQbservable<long> service over TCP port 3205 on the local computer.
Server
IObservable<long> source = Observable.Interval(TimeSpan.FromSeconds(1));

var service = source.ServeQbservableTcp(new IPEndPoint(IPAddress.Loopback, 3205));

using (service.SubscribeEither(
	client => Console.WriteLine("Client shutdown."),
	ex => Console.WriteLine("Error: {0}", ex.Message),
	ex => Console.WriteLine("Fatal error: {0}", ex.Message),
	() => Console.WriteLine("This will never be printed because a service host never completes.")))
{
	Console.ReadKey();
}
The following example creates a LINQ query over the IQbservable<long> service that is created by the previous example. Subscribing to the query on the client causes the query to be serialized to the server and executed there. In other words, the where clause is actually executed on the server so that the client only receives the data that it requested without having to do any filtering itself. The client will receive the first six values, one per second. The server then filters out the next 2 values - it does not send them to the client. Finally, the remaining values are sent to the client until either the client or the server disposes of the subscription.
Client
var client = new QbservableTcpClient<long>(new IPEndPoint(IPAddress.Loopback, 3205));

IQbservable<long> query =
	from value in client.Query()
	where value <= 5 || value >= 8
	select value;

using (query.Subscribe(
	value => Console.WriteLine("Client observed: " + value),
	ex => Console.WriteLine("Error: {0}", ex.Message),
	() => Console.WriteLine("Completed")))
{
	Console.ReadKey();
}

Getting Started

  1. Download the latest release and unzip it.
  2. All of the assemblies that you need to build your own applications are located in the bin folder.
To run the examples:
  1. Run QbservableServer.exe.
    1. The server will start hosting example Qbservable services as soon as the console application begins.
    2. Pressing a key at any time will stop the server.
  2. Run QbservableClient.exe.
    1. You can run several client console applications at the same time.
  3. When the client console application starts, press any key to connect to the server. The client will begin running the first example.
  4. Press any key to stop the current example and start the following example.
To build the source code:
  1. Download the source code from the Source Code tab.
    1. The solution is located under the QbservableProvider directory, but it depends upon the relative path to the Main branch. Make sure that both paths exist.
  2. You must build the Rxx 2.0 Beta project in the Main branch in Debug mode. See How To Build the Source Code for instructions.
  3. Open the QbservableProvider solution in VS 11 Beta.
  4. Set the QbservableServer project as the startup project.
  5. Build and run. The server will start as soon as the console application begins.
  6. Set the QbservableClient project as the startup project.
  7. Build and run. You can run several client console applications at the same time.
  8. When the client console application starts, press any key to connect to the server.
Tip: To see the original and rewritten expression trees, run the client application with the debugger attached and look at the Output window.

Planning

  1. Research building on top of WCF to support advanced configuration, customization, extensibility, standardization and additional transports such as WebSockets over HTTP.
  2. Improve expression tree serialization; e.g., fix bugs and write unit tests.
  3. Consider security.
  4. Consider memory/performance.
  5. Support querying from Silverlight 5 and Windows Phone 7.5 apps. (This may require WebSockets though.)
  6. Support querying from RxJS clients.

Last edited May 5, 2012 at 10:10 PM by davedev, version 12