如何将基于代理的并发与免费的 monad 结合起来？

Question

Continuing my rediscovery 的免费 monad，我决定将我正在编写的一个小程序转换为使用免费 monad。

该程序收集通过 MQTT 从我拥有的一些空气传感器接收到的消息，并将结果写入数据库。我还没有实现任何错误处理，稍后会出现。

open System
open uPLibrary.Networking.M2Mqtt
open uPLibrary.Networking.M2Mqtt.Messages
open MySql.Data.MySqlClient


let connString = "server=...."

type agentMessage =
    |Mqtt of string * DateTime
    |ExitCode
    |WaitForExit of AsyncReplyChannel<unit>
    |NotifyDbReady

type agentState =
    {
        MqttList : (string * DateTime) list
        ReplyOpt : AsyncReplyChannel<unit> option
        DbReady : bool
    }

type SensorReading =
    {
        SensorName:string
        SensorDate:string
        ReadingName:string
        ReadingValue:string
    }

let getSensorReading (s:string, dt:DateTime) =
    match s.Split("@") with
    |ar when ar.Length = 2 ->
        ar.[1].Split("|")
        |> Array.choose (fun r ->
            match r.Split(":") with
            |sar when sar.Length = 2 ->
                {
                    SensorName = ar.[0]
                    SensorDate = dt.ToString("yyyy-MM-dd HH:mm:ss")
                    ReadingName = sar.[0]
                    ReadingValue = sar.[1]
                } |> Some
            |sar ->
                printfn "Reading string [%s] has %i values" r sar.Length
                None
        ) |> Some
    |ar ->
        printfn "Message [%s] has %i values" s ar.Length
        None

let getSensorReadings lst = lst |> List.toArray |> Array.choose getSensorReading |> Array.collect id

let getLoadReadingsSql readings =
    readings |> Array.map (fun r ->
        sprintf "('%s', '%s', '%s', %s)" r.SensorName r.SensorDate r.ReadingName r.ReadingValue
    ) |> String.concat ", "
    |> sprintf "insert into sensor_reading_staging(sensor_name, sensor_date, reading_name, reading_value) values %s"
    |> (fun sql -> 
        MySqlHelper.ExecuteNonQuery(connString, sql) |> ignore
        MySqlHelper.ExecuteNonQuery(connString, "call process_sensor_reading_staging()") |> ignore
    )

let agent = MailboxProcessor.Start (fun inbox ->
    let processSensorReadingList lst =
        getSensorReadings lst
        |> getLoadReadingsSql

        inbox.Post NotifyDbReady

    let rec messageLoop oldState = async{
        let! msg = inbox.Receive()
        let newState =
            match msg with
            |Mqtt (s, dt) when oldState.DbReady ->
                async{(s, dt) :: oldState.MqttList |> processSensorReadingList } |> Async.Start
                {oldState with MqttList = []; DbReady = false}
            |Mqtt (s, dt) ->
                {oldState with MqttList = (s, dt) :: oldState.MqttList}
            |NotifyDbReady when oldState.MqttList.Length > 0 ->
                async{oldState.MqttList |> processSensorReadingList } |> Async.Start
                {oldState with MqttList = []; DbReady = false}
            |NotifyDbReady ->
                {oldState with DbReady = true}
            |WaitForExit rep ->
                {oldState with ReplyOpt = Some rep}
            |ExitCode ->
                oldState.ReplyOpt |> Option.map (fun rep -> rep.Reply()) |> ignore
                {oldState with ReplyOpt = None}

        return! messageLoop newState
    }
    messageLoop {MqttList = []; ReplyOpt = None; DbReady = true}
)

[<EntryPoint>]
let main argv =
    let client = new MqttClient(argv.[0])
    printfn "Connecting to broker"
    client.Connect(Guid.NewGuid().ToString()) |> ignore

    client.Subscribe([|"AirSensorReadings"|], [|MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE|]) |> ignore

    use __ =
        client.MqttMsgPublishReceived
        |> Observable.subscribe(fun evArgs -> (evArgs.Message |> Array.map char |> String.Concat, DateTime.Now) |> Mqtt |> agent.Post)

    agent.PostAndReply (fun rep -> WaitForExit rep)

    //the agent never actually receives the exit message
    0 // return an integer exit code

为此，我正在关注Mark Seemann's excellent blog post。

所以我想我需要使用返回类型为单位的每个函数，并将其推送到解释器。

这好像是

1. 从 MQTT 接收消息
2. 写入数据库
3. 邮寄给代理

前两个似乎很容易。最后一个让我有点摸不着头脑。 这个练习产生的 AST 本质上是一个顺序的指令列表。没有什么并发的。但是发布到代理的消息可能随时发生。 那么如何结合这两个概念呢？会不会是每条收到的消息都会生成一个新的 AST，最终产生一个纯单元？换句话说，只有上面列表中的第 2 项被建模为指令。 或者我是否会为代理和碰巧写入数据库的工作完全独立的 AST？

Answer 1

好的，我想出了一个我认为将这两个概念结合起来的不错的方法。以下代码是自包含的。

open System

type SensorReading =
    {
        SensorName:string
        SensorDate:string
        ReadingName:string
        ReadingValue:string
    }

type AgentState =
    {
        MqttList : (string * DateTime) list
        DbReady : bool
    }

type AgentMessage =
    |Mqtt of string * DateTime
    |NotifyDbReady
    |ExitCode

type Instruction<'a> =
    |ReceiveMessage of (AgentMessage -> 'a)
    |InsertReadings of SensorReading [] * 'a
    |PostMessage of AgentMessage * 'a

type Program<'a> =
    |Free of Instruction<Program<'a>>
    |Pure of 'a

let mapI (f:'a -> 'b) : Instruction<'a> -> Instruction<'b> = function
    |ReceiveMessage next -> ReceiveMessage (next >> f)
    |InsertReadings (x, next) -> InsertReadings (x, next |> f)
    |PostMessage (x, next) -> PostMessage (x, next |> f)

let rec bind f = function
    |Free instruction -> instruction |> mapI (bind f) |> Free
    |Pure x -> f x

let map f = bind (f >> Pure)

let receiveMessage = Pure |> ReceiveMessage |> Free
let insertReadings lst = (lst, Pure ()) |> InsertReadings |> Free
let postMessage msg = (msg, Pure ()) |> PostMessage |> Free

type programBuilder() =
    member this.Bind(x, f) = bind f x
    member this.Return x = Pure x
    member this.ReturnFrom x = x
    member this.Zero() = Pure ()

let program = programBuilder()

let getSensorReading (s:string, dt:DateTime) =
    match s.Split("@") with
    |ar when ar.Length = 2 ->
        ar.[1].Split("|")
        |> Array.choose (fun r ->
            match r.Split(":") with
            |sar when sar.Length = 2 ->
                {
                    SensorName = ar.[0]
                    SensorDate = dt.ToString("yyyy-MM-dd HH:mm:ss")
                    ReadingName = sar.[0]
                    ReadingValue = sar.[1]
                } |> Some
            |_ -> None
        ) |> Some
    |_ -> None

let getSensorReadings lst = lst |> List.toArray |> Array.choose getSensorReading |> Array.collect id

let rec processMessage oldState = program {
    let! msg = receiveMessage
    match msg with
    |Mqtt (s, dt) when oldState.DbReady ->
        do! (s, dt) :: oldState.MqttList |> getSensorReadings |> insertReadings
        return! {oldState with MqttList = []; DbReady = false} |> processMessage
    |Mqtt (s, dt) ->
        return! {oldState with MqttList = (s, dt) :: oldState.MqttList} |> processMessage
    |NotifyDbReady when oldState.MqttList.Length > 0 ->
        do! oldState.MqttList |> getSensorReadings |> insertReadings
        return! {oldState with MqttList = []; DbReady = false} |> processMessage
    |NotifyDbReady ->
        return! {oldState with DbReady = true} |> processMessage
    |ExitCode -> return ()
}

let startAgent insertToDb = 
    MailboxProcessor.Start (fun inbox ->
    let rec interpret = function
        |Pure x -> x
        |Free (ReceiveMessage next) ->
            inbox.Receive() |> Async.RunSynchronously |> next |> interpret
        |Free (InsertReadings (lst, next)) ->
            lst |> insertToDb (fun () -> inbox.Post NotifyDbReady)
            next |> interpret
        |Free (PostMessage (msg,next)) ->
            inbox.Post msg
            next |> interpret

    async{{MqttList = []; DbReady = true} |> processMessage |> interpret}
)
let printReadingsToConsole notify readings =
    async{
        readings |> Array.map (fun r ->
            sprintf "('%s', '%s', '%s', %s)" r.SensorName r.SensorDate r.ReadingName r.ReadingValue
        ) |> String.concat "\n" |> printfn "%s"
        do! Async.Sleep 5000
        notify()
    } |> Async.Start

[<EntryPoint>]
let main argv =

    let agent = startAgent printReadingsToConsole

    async{
        let r = new Random()
        let getRandomMessage() =
            let sensorName = r.Next() |> sprintf "Sensor%i"
            let r1 = r.Next() |> sprintf "Reading%i"
            let r2 = r.Next() |> sprintf "Reading%i"
            let v1 = r.NextDouble()
            let v2 = r.NextDouble()
            sprintf "%s@%s:%f|%s:%f" sensorName r1 v1 r2 v2

        let rec sendTestMessages () = async{
            printfn "Sending Test Message"
            (getRandomMessage(), DateTime.Now) |> Mqtt |> agent.Post
            do! Async.Sleep 500
            do! sendTestMessages()
        }
        do! sendTestMessages()
    } |> Async.Start

    Console.ReadKey() |> ignore
    agent.Post ExitCode

    0 // return an integer exit code

这个想法是邮箱处理器成为解释器。接收到消息时发生的所有逻辑都定义为 AST，并且解释器选择何时对消息进行异步响应。

解释器的一切都是完全纯粹的，所有的 IO 都被推到边缘。