FSharpWebSocketHelpers

WebSocketHelpers.fs

module WebsocketHelpers =
    open System
    open System.Net.WebSockets
    open System.Threading
    open Hopac
    let readMessage messageType (stream : #IO.Stream) (socket : WebSocket) =  job {
        let buffer = new ArraySegment<Byte>( Array.create (1500) Byte.MinValue)

        let rec readTillEnd' () = job {
            let! (result : WebSocketReceiveResult) = socket.ReceiveAsync(buffer,CancellationToken.None)
            if result.MessageType <> messageType then return ()
            stream.Write(buffer.Array,buffer.Offset, result.Count )
            if result.EndOfMessage then return ()
            else return! readTillEnd' ()
        }
        do! readTillEnd' ()

        return stream
    }

    let readText socket =
        readMessage WebSocketMessageType.Text (new IO.MemoryStream()) socket
        |> Job.map(
            fun stream ->
                stream.Seek(0L,IO.SeekOrigin.Begin) |> ignore
                let chars =
                    stream.ToArray()
                    //Remove null terminators
                    |> Array.filter(fun x -> x <> '\000'B)
                Text.Encoding.UTF8.GetString(chars)
            )

    let readBinary stream socket =
        readMessage WebSocketMessageType.Binary stream socket    

    let readAsStream (socketJ : unit -> Job<WebSocket>) =
        let generatedStream (websocket : WebSocket) =
            Stream.unfoldJob
                (fun (websocket : WebSocket) ->
                    if websocket.State <> WebSocketState.Closed then
                        readText websocket
                        |> Job.map(fun text -> Some((websocket,text),websocket))
                    else
                        Job.result None
                ) websocket
        socketJ ()
        |> Job.map(generatedStream)


    let sendMessage messageType (message : #IO.Stream)  (socket : WebSocket)= job {
        let buffer = Array.create (1500) Byte.MinValue
        let rec sendMessage' () = job {
            let! read = message.ReadAsync(buffer,0,buffer.Length)
            if read > 0 then
                do! socket.SendAsync(ArraySegment(buffer), messageType, false, CancellationToken.None) |> Job.awaitUnitTask
                return! sendMessage'()
            else
                do! socket.SendAsync(ArraySegment(Array.empty), messageType, true, CancellationToken.None) |> Job.awaitUnitTask
        }

       return! sendMessage'()
    }

    let sendText (text : string) (socket : WebSocket) =
        Job.using
            (new IO.MemoryStream(Text.Encoding.UTF8.GetBytes text))
            (fun memStream -> sendMessage WebSocketMessageType.Text memStream socket)


    let sendBinary stream socket =
        sendMessage WebSocketMessageType.Binary stream socket