parsebitmessage.fs

type ParserState = {
    InHeader: bool
    Command: string
    Checksum: int
    PayloadLength: int
    Data: byte[]
}

type BitcoinMessageParser(networkData: IObservable<byte[]>) =
    let headerLen = 24

    let rec parse (state: ParserState) (reader: BinaryReader) (messages: BitcoinMessage list) = 
        let remaining = int(reader.BaseStream.Length - reader.BaseStream.Position)
        if (state.InHeader && remaining < headerLen) || (not state.InHeader && remaining < state.PayloadLength) then
            (messages, state)
        else
            if state.InHeader then
                let messageMagic = reader.ReadInt32()
                if messageMagic <> magic then
                    raise ParseException
                else
                    let command = (new string(reader.ReadChars(12))).TrimEnd([| '\000' |])
                    let payloadLength = reader.ReadInt32()
                    let checksum = reader.ReadInt32()
                    parse 
                        ({ state with InHeader = false; Command = command; Checksum = checksum; PayloadLength = payloadLength })
                        reader
                        messages
            else 
                let payload = reader.ReadBytes(state.PayloadLength)
                let hash256 = dsha payload
                let checksum = BitConverter.ToInt32(hash256, 0)
                if checksum <> state.Checksum then
                    raise ParseException
                parse 
                    ({ state with InHeader = true })
                    reader
                    (new BitcoinMessage(state.Command, payload) :: messages)

    let acc (state: ParserState) (chunk: byte[]): BitcoinMessage list * ParserState =
        let data = Array.concat [state.Data; chunk]
        use ms = new MemoryStream(data)
        use reader = new BinaryReader(ms)
        let (messages, newState) = parse state reader List.empty
        let remainingBytes = data.[int ms.Position..]
        (messages, { newState with Data = remainingBytes })

    let bitcoinMessages = 
        networkData
            .Scan(
                (List.empty<BitcoinMessage>, { InHeader = true; Data = Array.empty; Command = ""; Checksum = 0; PayloadLength = 0}),
                fun (messages, state) (chunk) ->
                    acc state chunk 
            )
            .Select(fst)
            .SelectMany(fun m -> (m |> List.toSeq).ToObservable())
    member x.BitcoinMessages with get() = bitcoinMessages