let rnd = new System.Random() // used for generating random messages

///cfg: the configution passed in by storm
///runner: a spout runner function (passed in from topology)
let spout runner cfg = 
    //define the function that will return the emitter function
    //emit: a function that emits message to storm
    let createEmitter emit = fun () -> async { tuple [ rnd.Next(0, 100) ] |> emit } //the "next" function
    //run the spout
    createEmitter |> runner

///cfg: the configuration passed in from Storm
///runner: passed in from topology
///log: passed in from topology
///emit: passed in from topology
let addOneBolt runner log emit cfg = 
    //define the function that will return the consumer function
    let createAdder = 
        //accept/consume tuple function
        fun (msg : Json) -> 
            async { 
                log "msg" (sprintf "%A" msg)
                // note the tuple fields are accessible as Jval array items
                tuple [ msg?tuple.[0].ValI + 1 ] |> emit
            }
    //run the bolt
    createAdder |> runner

///cfg: the configution passed in by storm
///runner: passed in from topology
///log: passed in from topology
let resultBolt runner log cfg = 
    //define the function that will return the consumer function
    let createReader = 
        //accept messages function
        fun (msg : Json) -> async { log "x" (sprintf "%A" msg?tuple.[0].ValI) }
    //run the bolt
    createReader |> runner

open StormDSL

let topology = 
    { TopologyName = "FstSample"
      Spouts = 
          [ { Id = "SimpleSpout"
              Outputs = [ Default [ "number" ] ]
              Spout = Local { Func = spout Storm.simpleSpoutRunner }
              Config = JsonNull
              Parallelism = 1 } ]
      Bolts = 
          [ { Id = "AddOneBolt"
              Outputs = [ Default [ "number" ] ]
              Inputs = [ DefaultStream "SimpleSpout", Shuffle ]
              Bolt = Local { Func = addOneBolt Storm.autoAckBoltRunner Logging.log Storm.emit}
              Config = JsonNull
              Parallelism = 2 }
            { Id = "ResultBolt"
              Outputs = []
              Inputs = [ DefaultStream "AddOneBolt", Shuffle ]
              Bolt = Local { Func = resultBolt Storm.autoAckBoltRunner Logging.log }
              Config = JsonNull
              Parallelism = 2 } ] }

open NUnit.Framework
open System
open StormTest

let conf = 
    { PidDir = ""
      TaskId = ""
      Json = jval "" }

[<Test>]
let ``spout emits``() = 
    let results = ref []
    let s = spout (simpleSpoutRunner (fun x -> results := x :: results.Value) [ next; next ]) conf
    Async.RunSynchronously s
    match results.Value with
    | x :: y :: [] -> ()
    | _ -> Assert.Fail "Must have emitted exactly two!"

[<Test>]
let ``bolt adds``() = 
    let results = ref []
    let b = 
        addOneBolt (autoAckBoltRunner [ tuple [ 123 ] ] Console.WriteLine) 
            (fun tag desc -> Console.WriteLine("{0}: {1}", tag, desc)) (fun t -> results := t :: results.Value) conf
    Async.RunSynchronously(b, 2000)
    match results.Value with
    | x :: [] -> Assert.AreEqual(124, x?tuple.[0].ValI)
    | _ -> Assert.Fail "Head, no tail, that's the deal!"

[<Test>]
let ``bolt consumes``() = 
    let results = ref []
    let b = 
        resultBolt (autoAckBoltRunner [ tuple [ 123 ] ] Console.WriteLine) 
            (fun tag desc -> results := (tag, desc) :: results.Value) conf
    Async.RunSynchronously(b, 2000)
    match results.Value with
    | x :: [] -> Assert.AreEqual(("x", sprintf "%A" 123), x)
    | _ -> Assert.Fail "Head, no tail, that's the deal!"