Runtime "EndOfStreamException" error in MS Bond serialization over ZeroMQ

Question

To start off, it is worth mentioning that within a single F# solution, serializing and deserializing of Bond messages works fine. However, I am having trouble properly handling the sending and/or receiving of the message over ZeroMQ.

There is a runtime error on the subscriber side of the following program. A .bond file is defined and compiled with the bond compiler. Then a dll is created from C# to be called from F# . I then have two F# programs. One that publishes serialized data over a tcp socket and another that is a subscriber. When the message is received on the sub, the line which tries to Unmarshal the raw data is the one that causes the runtime error. Can anyone see the reason for this?

[EDIT] Per Fyodor's comment, I made a change on the publisher side which changes the error on the subscriber side. So the error likely has something to do with how i am packing and unpacking the information.

This is the .bond file

namespace Examples

struct Record
{
    0: map<string, double> payload;
}

Here is the publisher:

// publisher

open System
open Bond
open Bond.Protocols
open Bond.IO.Safe
open ZeroMQ

let ctx = new ZContext()
let publisher = new ZSocket(ctx, ZSocketType.PUB)
publisher.Bind("tcp://*:5556")

let src = new Examples.Record()
src.payload.Add("a", 1.)
src.payload.Add("b", 2.)

let output = new OutputBuffer()
let writer = new CompactBinaryWriter<OutputBuffer>(output)

while true do
    Marshal.To(writer, src)
    //let input = new InputBuffer(output.Data)
    //let byteArr = input.ReadBytes(int(input.Length - 1L))
    let updateFrame = new ZFrame(System.Text.Encoding.ASCII.GetString output.Data.Array)
    publisher.Send(updateFrame)

Here is the subscriber:

// subscriber

open Bond
open Bond.Protocols
open Bond.IO.Safe
open System
open System.Text
open ZeroMQ

let ctx = new ZContext()
let subscriber = new ZSocket(ctx, ZSocketType.SUB)
subscriber.Connect("tcp://127.0.0.1:5556")
subscriber.SubscribeAll()

let output = new OutputBuffer()    
while true do    
    let received = subscriber.ReceiveFrame()
    let byteArr = Encoding.ASCII.GetBytes (received.ReadString())
    let arrSeg = ArraySegment<byte>(byteArr)
    let input = new InputBuffer(arrSeg)
    let dst = Unmarshal<Examples.Record>.From(input)
    for KeyValue(k, v) in dst.payload do
        printfn "%A %A" k v

Answer 1

On the receiving side, when you attempt to decode the marshaled Bond Compact Binary as an ASCII string, you're losing some of the payload. When marshaling a struct like Record to Compact Binary, the first four bytes of the payload are 0x43 0x42 0x10 0x00. When reading a string from a ZFrame, the first embedded NUL (0x00) that is encountered signals the end of the string, regardless of the size of the frame. So, the reading side sees only 0x43 0x42 0x10 instead of the whole payload (29 bytes when I tested).

Since Compact Binary is a binary protocol, you'll want to use the ZFrame constructor that takes a buffer on the publisher side:

let updateFrame = new ZFrame(output.Data.Array, output.Data.Offset, output.Data.Count)

On the subscriber side, you'll want to just read the buffer:

let byteArr = received.Read()

Also, on the publisher side, you're constantly accumulating data into the same OutputBuffer. You'll want to reset output.Position to 0 before you marshall your next record to re-use the buffer instead of growing it:

while true do  
    Marshal.To(writer, src)
    let updateFrame = new ZFrame(output.Data.Array, output.Data.Offset, output.Data.Count)output.Data.Array)
    publisher.Send(updateFrame)
    output.Position <- 0

Another thing to note: the default buffer allocated for an OutputBuffer is 65KiB. Consider making this smaller, once you know about how large your payloads are going to be.

NB: I debugged this in a C# application that had similar semantics. Here's what I used:

namespace so_q_zmq
{
    using System;
    using System.Collections.Generic;
    using System.Text;
    using System.Threading.Tasks;
    using Bond;
    using Bond.IO.Safe;
    using Bond.Protocols;
    using ZeroMQ;

    [Schema]
    class Record
    {
        [Id(0)]
        public Dictionary<string, double> payload = new Dictionary<string, double>();
    }

    class Program
    {
        static void Main(string[] args)
        {
            var pTask = Task.Run(() =>
            {
                try
                {
                    Publisher();
                }
                catch (Exception ex)
                {
                    Console.WriteLine("Publisher failed: {0}", ex);
                }
            });

            var sTask = Task.Run(() =>
            {
                try
                {
                    Subscriber();
                }
                catch (Exception ex)
                {
                    Console.WriteLine("Subscriber failed: {0}", ex);
                }
            });

            Task.WaitAll(pTask, sTask);
            Console.WriteLine("Done");
            Console.ReadLine();
        }

        static void Publisher()
        {
            var ctx = new ZContext();
            var publisher = new ZSocket(ctx, ZSocketType.PUB);
            publisher.Bind("tcp://127.0.0.1:12345");

            var src = new Record();
            src.payload.Add("a", 1.0);
            src.payload.Add("b", 2.0);

            var output = new OutputBuffer();
            var writer = new CompactBinaryWriter<OutputBuffer>(output);

            for (;;)
            {
                Marshal.To(writer, src);
                // INCORRECT:
                // var str = Encoding.ASCII.GetString(output.Data.Array);
                // var updateFrame = new ZFrame(str);
                var updateFrame = new ZFrame(output.Data.Array, output.Data.Offset, output.Data.Count);
                publisher.Send(updateFrame);
                output.Position = 0;
            }
        }

        static void Subscriber()
        {
            var ctx = new ZContext();
            var subscriber = new ZSocket(ctx, ZSocketType.SUB);
            subscriber.Connect("tcp://127.0.0.1:12345");
            subscriber.SubscribeAll();

            for (;;)
            {
                var received = subscriber.ReceiveFrame();
                // INCORRECT
                // var str = received.ReadString();
                // var byteArr = Encoding.ASCII.GetBytes(str);
                var byteArr = received.Read();
                var arrSeg = new ArraySegment<byte>(byteArr); // There's an InputBuffer ctor that takes a byte[] directly
                var input = new InputBuffer(arrSeg);
                var dst = Unmarshal<Record>.From(input);
                foreach (var kvp in dst.payload)
                {
                    Console.WriteLine("{0} {1}", kvp.Key, kvp.Value);
                }
            }
        }
    }
}