FSharpWebSocketHelpers

WebSocketHelpers.fs

module WebsocketHelpers =
    open System
    open System.Net.WebSockets
    open System.Threading
    open Hopac
    let readMessage buffer messageType (stream : #IO.Stream) (socket : WebSocket) =  job {
        let buffer = new ArraySegment<Byte>( Array.create (buffer) 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 buffer (decoder : _ -> string) socket =
        Job.using (new IO.MemoryStream()) <|
            fun stream ->
                readMessage buffer WebSocketMessageType.Text (stream) socket
                |> Job.map(
                    fun stream ->
                        stream.Seek(0L,IO.SeekOrigin.Begin) |> ignore
                        stream.ToArray()
                        |> Array.filter(fun x -> x <> '\000'B) //Remove null terminators
                        |> decoder
                    )

    let readTextUtf8 buffer = readText buffer Text.Encoding.UTF8.GetString

    let readBinary stream socket =
        readMessage 1500 WebSocketMessageType.Binary stream socket

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


    let readTextAsStream = readAsStream (readTextUtf8 1500)

    let sendMessage buffer messageType (message : #IO.Stream)  (socket : WebSocket)= job {
        let buffer = Array.create (buffer) Byte.MinValue
        let rec sendMessage' () = job {
            let! read = message.ReadAsync(buffer,0,buffer.Length)
            if read > 0 then
                do! socket.SendAsync(ArraySegment(buffer |> Array.take read), 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 buffer (encoder : string -> byte array) (text : string) (socket : WebSocket) =
        Job.using (new IO.MemoryStream( encoder text)) <|
            fun memStream -> sendMessage buffer WebSocketMessageType.Text memStream socket


    let sendTextUtf8 buffer = sendText buffer Text.Encoding.UTF8.GetBytes

    let sendBinary stream socket =
        sendMessage 1500 WebSocketMessageType.Binary stream socket