Agent that can upgrade its functionality on the fly.

Agent that can upgrade its functionality on the fly. (F# MailboxProcessor containing function in the loop...)

Erlang-style message-passing: Agent that can upgrade its functionality on the fly.

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19: 
20: 
21: 
22: 
23: 
24: 
25: 
26: 
27: 
28: 
29: 
30: 
31: 
32: 
33: 
34: 
35: 
36: 
37:

// Note: static typing, this agent can't upgrade its state data type (so better to use obj or custom interface)...
// Also this runs only in localhost...
// So this is more a technical demo than something useful
// See slide 37: http://www.infoq.com/presentations/Message-Passing-Concurrency

open System

type Methods<'state, 'x, 'reply> = 
| Upgrade of 
    ('x*'state -> 'x*'state) // function what to do
    *('state -> 'state) // just for state conversion
| From of AsyncReplyChannel<'reply> * 'x

type UpgradableAgent<'state, 'x>() =
    let gen_server = MailboxProcessor.Start(fun msg ->
        let rec loop (state, f) =
            async { 
                let! receive = msg.Receive()
                match receive with
                | Upgrade(f1,f2) ->
                    let state1 = f2(state)
                    return! loop(state1, f1)
                | From(from, x) ->
                    let (reply, state1) = f(x,state)
                    from.Reply(reply)
                    return! loop(state1, f)
            }
        let initDoit (x,state) = (x, state)
        let initState = Unchecked.defaultof<'state> //None
        loop(initState, initDoit))
       
    member this.DoIt (item:'x) =  
        // could use also PostAndAsyncReply
        gen_server.PostAndReply(fun rep -> From(rep, item))

    member this.Upgrade functionality =  
        functionality |> Upgrade |> gen_server.Post

Some tests

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19: 
20: 
21: 
22:

let test1 =
    let server = new UpgradableAgent<int, int>()
    let myfunc = fun (x,state) -> (x+state, state) 
    let myStateConvert = fun initstate -> 5
    server.Upgrade(myfunc, myStateConvert)
    Console.WriteLine(server.DoIt(7)); // 12
    Console.WriteLine(server.DoIt(7)); // 12

    let myfunc2 = fun (x,state) -> (x+state, x+state) 
    let myStateConvert2 = fun initstate -> 5
    server.Upgrade(myfunc2, myStateConvert2)
    Console.WriteLine(server.DoIt(7)); // 12
    Console.WriteLine(server.DoIt(7)); // 19
    Console.WriteLine(server.DoIt(7)); // 26
    
let test3 =
    let server3 = new UpgradableAgent<obj, string>()
    let myfunc3 ((x:string), (state:obj)) = (x+unbox(state), box(x+unbox(state))) 
    let myStateConvert3 = fun initstate -> box(" world!")
    server3.Upgrade(myfunc3, myStateConvert3)
    Console.WriteLine(server3.DoIt("hello")); // "hello world!"
    Console.WriteLine(server3.DoIt("hello ")); // "hello hello world!"

namespace System

type Methods<'state,'x,'reply> =
| Upgrade of ('x * 'state -> 'x * 'state) * ('state -> 'state)
| From of AsyncReplyChannel<'reply> * 'x

Full name: Script.Methods<_,_,_>

union case Methods.Upgrade: ('x * 'state -> 'x * 'state) * ('state -> 'state) -> Methods<'state,'x,'reply>

union case Methods.From: AsyncReplyChannel<'reply> * 'x -> Methods<'state,'x,'reply>

type AsyncReplyChannel<'Reply>
member Reply : value:'Reply -> unit

Full name: Microsoft.FSharp.Control.AsyncReplyChannel<_>

Multiple items
type UpgradableAgent<'state,'x> =
  new : unit -> UpgradableAgent<'state,'x>
  member DoIt : item:'x -> 'x
  member Upgrade : functionality:(('x * 'state -> 'x * 'state) * ('state -> 'state)) -> unit

Full name: Script.UpgradableAgent<_,_>

--------------------
new : unit -> UpgradableAgent<'state,'x>

val gen_server : MailboxProcessor<Methods<'state,'x,'x>>

Multiple items
type MailboxProcessor<'Msg> =
  interface IDisposable
  new : body:(MailboxProcessor<'Msg> -> Async<unit>) * ?cancellationToken:CancellationToken -> MailboxProcessor<'Msg>
  member Post : message:'Msg -> unit
  member PostAndAsyncReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> Async<'Reply>
  member PostAndReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> 'Reply
  member PostAndTryAsyncReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> Async<'Reply option>
  member Receive : ?timeout:int -> Async<'Msg>
  member Scan : scanner:('Msg -> Async<'T> option) * ?timeout:int -> Async<'T>
  member Start : unit -> unit
  member TryPostAndReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> 'Reply option
  ...

Full name: Microsoft.FSharp.Control.MailboxProcessor<_>

--------------------
new : body:(MailboxProcessor<'Msg> -> Async<unit>) * ?cancellationToken:Threading.CancellationToken -> MailboxProcessor<'Msg>

static member MailboxProcessor.Start : body:(MailboxProcessor<'Msg> -> Async<unit>) * ?cancellationToken:Threading.CancellationToken -> MailboxProcessor<'Msg>

val msg : MailboxProcessor<Methods<'state,'x,'x>>

val loop : ('state * ('x * 'state -> 'x * 'state) -> Async<'a>)

val state : 'state

val f : ('x * 'state -> 'x * 'state)

val async : AsyncBuilder

Full name: Microsoft.FSharp.Core.ExtraTopLevelOperators.async

val receive : Methods<'state,'x,'x>

member MailboxProcessor.Receive : ?timeout:int -> Async<'Msg>

val f1 : ('x * 'state -> 'x * 'state)

val f2 : ('state -> 'state)

val state1 : 'state

val from : AsyncReplyChannel<'x>

val x : 'x

val reply : 'x

member AsyncReplyChannel.Reply : value:'Reply -> unit

val initDoit : ('a * 'b -> 'a * 'b)

val x : 'a

val state : 'b

val initState : 'state

module Unchecked

from Microsoft.FSharp.Core.Operators

val defaultof<'T> : 'T

Full name: Microsoft.FSharp.Core.Operators.Unchecked.defaultof

val this : UpgradableAgent<'state,'x>

member UpgradableAgent.DoIt : item:'x -> 'x

Full name: Script.UpgradableAgent`2.DoIt

val item : 'x

member MailboxProcessor.PostAndReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> 'Reply

val rep : AsyncReplyChannel<'x>

member UpgradableAgent.Upgrade : functionality:(('x * 'state -> 'x * 'state) * ('state -> 'state)) -> unit

Full name: Script.UpgradableAgent`2.Upgrade

val functionality : ('x * 'state -> 'x * 'state) * ('state -> 'state)

member MailboxProcessor.Post : message:'Msg -> unit

val test1 : unit

Full name: Script.test1

val server : UpgradableAgent<int,int>

Multiple items
val int : value:'T -> int (requires member op_Explicit)

Full name: Microsoft.FSharp.Core.Operators.int

--------------------
type int = int32

Full name: Microsoft.FSharp.Core.int

--------------------
type int<'Measure> = int

Full name: Microsoft.FSharp.Core.int<_>

val myfunc : (int * int -> int * int)

val x : int

val state : int

val myStateConvert : ('a -> int)

val initstate : 'a

member UpgradableAgent.Upgrade : functionality:(('x * 'state -> 'x * 'state) * ('state -> 'state)) -> unit

type Console =
  static member BackgroundColor : ConsoleColor with get, set
  static member Beep : unit -> unit + 1 overload
  static member BufferHeight : int with get, set
  static member BufferWidth : int with get, set
  static member CapsLock : bool
  static member Clear : unit -> unit
  static member CursorLeft : int with get, set
  static member CursorSize : int with get, set
  static member CursorTop : int with get, set
  static member CursorVisible : bool with get, set
  ...

Full name: System.Console

Console.WriteLine() : unit
   (+0 other overloads)
Console.WriteLine(value: string) : unit
   (+0 other overloads)
Console.WriteLine(value: obj) : unit
   (+0 other overloads)
Console.WriteLine(value: uint64) : unit
   (+0 other overloads)
Console.WriteLine(value: int64) : unit
   (+0 other overloads)
Console.WriteLine(value: uint32) : unit
   (+0 other overloads)
Console.WriteLine(value: int) : unit
   (+0 other overloads)
Console.WriteLine(value: float32) : unit
   (+0 other overloads)
Console.WriteLine(value: float) : unit
   (+0 other overloads)
Console.WriteLine(value: decimal) : unit
   (+0 other overloads)

member UpgradableAgent.DoIt : item:'x -> 'x

val myfunc2 : (int * int -> int * int)

val myStateConvert2 : ('a -> int)

val test3 : unit

Full name: Script.test3

val server3 : UpgradableAgent<obj,string>

type obj = Object

Full name: Microsoft.FSharp.Core.obj

Multiple items
val string : value:'T -> string

Full name: Microsoft.FSharp.Core.Operators.string

--------------------
type string = String

Full name: Microsoft.FSharp.Core.string

val myfunc3 : (string * obj -> string * obj)

val x : string

val state : obj

val unbox : value:obj -> 'T

Full name: Microsoft.FSharp.Core.Operators.unbox

val box : value:'T -> obj

Full name: Microsoft.FSharp.Core.Operators.box

val myStateConvert3 : ('a -> obj)

Raw view Test code New version

More information

Link:	http://fssnip.net/h3
Posted:	11 years ago
Author:	Tuomas Hietanen
Tags:	mailboxprocessor , agent