UdpMb

UdpMb.fs

open System.Net
open System

type System.Net.Sockets.UdpClient with
      member x.AsyncSend (bytes: byte[]) =
            Async.FromBeginEnd((fun (ar, s) -> x.BeginSend(bytes, bytes.Length, ar, s)), x.EndSend)
      member x.AsyncReceive (endPoint: IPEndPoint ref) =
            Async.FromBeginEnd(x.BeginReceive, fun ar -> x.EndReceive(ar, endPoint)) 

module Udp =
    open System.Net.Sockets
    open System.Net
    open Nessos.FsPickler
    open System.IO

    let binary = FsPickler.CreateBinary()
    type Sender<'msg> = Sender of IPEndPoint
    type MessageWrapper<'msg> = {sender : IPEndPoint; msg : 'msg}
    type Mailbox<'msg>(port: int) =
        let event = new Event<MessageWrapper<'msg>>()
        let udp = new UdpClient(port)
            
        member x.Listen() = 
            let rec loop () = async{
                let ip = new IPEndPoint(IPAddress.Any, port);
                let! bytes = udp.AsyncReceive(ref ip);
                let stream = new MemoryStream(bytes)
                let msg : 'msg = binary.Deserialize(stream)
                event.Trigger({msg = msg; sender = ip})
                return! loop ()
            }
            loop() |> Async.Start
            event.Publish

        member x.Send msg = async{
           let client = new UdpClient()
           client.EnableBroadcast <- true
           let ip =  new IPEndPoint(IPAddress.Broadcast, port)
           client.Connect ip
           let stream = new MemoryStream()
           binary.Serialize(stream,msg)
           let bytes = stream.ToArray()
           let! _ = client.AsyncSend (bytes) 
           return ()
           }

module Actor = 
    type Message(id, contents) =
        static let mutable count = -1
        member this.ID = id
        member this.Contents = contents
        static member CreateMessage(contents) =
            count <- count + 1
            Message(count, contents)
    let create() = new MailboxProcessor<Udp.MessageWrapper<Message>>(fun inbox ->
        let rec loop count =
            async { printfn "Message count = %d. Waiting for next message." count
                    let! msg = inbox.Receive()
                    printfn "Message received. from %A ID: %d Contents: %s"  msg.ip msg.msg.ID msg.msg.Contents
                    return! loop( count + 1) }
        loop -1)

[<EntryPoint>]
let main argv = 
    let mb = Udp.Mailbox<Actor.Message>(11000)
    let ev = mb.Listen()
    let actor = Actor.create()
    actor.Start()
    ev.Add(actor.Post)
    async{
    for i in 0..10 do
        do! mb.Send(Actor.Message.CreateMessage(sprintf "%d" i))
    } |> Async.RunSynchronously
    Console.ReadKey() |> ignore
    0 // return an integer exit code