<edit: There's a major problem in the code. It is basically the one in the comment regarding backpressure. I'll rework the code within a a few days...
It's time for a quick code review, good points, examples, handholding if anyone has time to spare for commenting and, yes, even nitpicking too. I wrote a short piece of code, purpose of which is to
- Read a resource on some predefined intervals. This resource can be at least a file or a network resource, so preferably the source and how to read it could be configured and be testable. The contents can text in JSON or XML or a CSV file or bytes or whatever. At least JSON for the purposes for this example. I assume all data can be read in one go.
- When the resource has been polled and data received, it needs to be transformed into some nice, strongly typed form.
- After the data has been transformed into a strongly typed form, it gets processed in various ways. It would be nice if one could have a notion of "tubes" here, which would work in asynchronously and independently so that an error in one "tube" could be just logged whereas processing in other ones go on.
- After processing data it's being tucked into various places. Like various queues (or a queue), files and so forth.
The code follows, but I'm not satisfied with it. For one, even if the idea of getData works with dataPoller, I need to add the Async.RunSynchronously in case I use an async method, or if not, omit it. Is there a way to do this either way independent of how getData is defined?
I've been reading about the excellent chap Scott Wlaschin's excellent railway oriented programming tutorial and am thinking how could apply the lessons here. For one, extending this excellent looking framework of thought to async isn't really clear to me (as written in the end of the slides). I'm also thinking if I could employ TPL.Dataflow or Hopac, which looks like being a fit (defining processing DAGs) to this kind of a processing...
... But really, I'd be happy even if someone would advise me if there's a neater way to write this poller (the Async.RunSynchronously part). The rest is something I know I'll be tempted to tackle on the cost of writing a program that actually does something.
Currently I have a module in a file CommonLibrary.fs like so
namespace CommonLibrary
[<AutoOpen>]
module Infrastructure =
open System.Reactive
open System.Reactive.Linq
//The current version of Rx does not have a notion of backpressure so if call durations rise, so will the process memory consumption too.
let dataPoller(interval, source) = Observable.Interval(interval) |> Observable.map(fun _ -> source)
[<AutoOpen>]
module DomainTypes =
open System
open FSharp.Data
type Data = {
someData: string
}
type DataPieces = {
contents: seq<Data> option
}
type DataPiecesProvider = JsonProvider<Sample = "DataPiece.json", Culture = "en-US">
let getData uri = async {
try
let! data = DataPiecesProvider.AsyncLoad(uri)
//Omitted for brevity, but basically here's a bunch of parsing code,
//which spans for a about ten lines.
//let datas =
return { contents = Some(datas) }
with _ -> return { contents = None }
}
And then I have the main file, Program.fs where I call these facilities as follows
open System
open System.IO
open System.Threading
open System.Text
open CommonLibrary
[<Literal>]
let pollingUri = "http://xyz/json"
[<EntryPoint>]
let main argv =
let exampleDataPump =
dataPoller(TimeSpan.FromSeconds(1.0), Async.RunSynchronously(getData pollingUri))
|> Observable.subscribe(fun i -> printfn "%A" (i.contents.Value |> Seq.head))
Thread.Sleep(System.Threading.Timeout.Infinite);
//These aren't never reached, but for the sake of completeness.
exampleDataPump.Dispose()
0
<edit 2014-05-04: So, I have a version two of the code. The original idea was to ask things (as you can see) of it, but I believe I have a better story.
Wheareas this version too could be improved in many ways (for one, it lacks documentation), I believe it has the problematic nature of relying on exceptions to count for retrying, well, in exceptional conditions. A better strategy could be to catch everything at the source, make it a domain event DU (taking cues from Scott's slides) and then handling them in the Rx pipeline.
In any event, here's the newest code. I'll work on the DU version too and add it here. Hopefully in the near future -- sooner than these edits. Then there's something to compare and discuss if there's something to discuss.
In CommonLibrary.fs
let getData uri = async {
let! data = DataPiecesProvider.AsyncLoad(uri)
//Omitted for brevity, but basically here's a bunch of parsing code,
//which spans for a about ten lines.
//let datas = ...
return datas //seq<Data>
[<Extension>]
type ObservableExtensions =
[<Extension>]
[<CompiledName("PascalCase")>]
static member inline retryAfterDelay<'TSource, 'TException when 'TException :> System.Exception>(source: IObservable<'TSource>, retryDelay: int -> TimeSpan, maxRetries, scheduler: IScheduler): IObservable<'TSource> =
let rec go(source: IObservable<'TSource>, retryDelay: int -> TimeSpan, retries, maxRetries, scheduler: IScheduler): IObservable<'TSource> =
source.Catch<'TSource, 'TException>(fun ex ->
if maxRetries <= 0 then
Observable.Throw<'TSource>(ex)
else
go(source.DelaySubscription(retryDelay(retries), scheduler), retryDelay, retries + 1, maxRetries - 1, scheduler))
go(source, retryDelay, 1, maxRetries, scheduler)
type DataPumpOperations =
static member dataPump(source: Async<_>): IObservable<_> = Async.StartAsTask(source).ToObservable()
static member dataPump(source: seq<_>): IObservable<_> = source |> Observable.toObservable
In Program.fs
let constantRetryStrategy(retryCount: int) =
TimeSpan.FromSeconds(1.0)
let maxRetries = 5
let ds = (getData pollingUri)
let t = Observable.Create(fun i -> (DataPumpOperations.dataPump ds).retryAfterDelay(constantRetryStrategy, maxRetries, Scheduler.Default).Subscribe(i)).Delay(TimeSpan.FromSeconds(1.0)).Repeat().Subscribe(fun i -> printfn "%A" (i |> Seq.head).origin)
