anon58192932 anon58192932 - 1 month ago 50
C++ Question

Simple FlatBuffer over ZeroMQ example - Copying struct to flatbuffer over zmq and back to a struct again

I've referenced some examples and I'm modeling my system off of the server and client example and I feel like I'm very close.

StarBuffer.fbs:

table StarBuffer {
radius: double;
mass: double;
volume: double;
}
root_type StarBuffer;


subscriber.cpp:

//
// Durable subscriber
//
// Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>

#include "zhelpers.hpp"
#include <zmq.hpp>
#include "StarBuffer.h" // generated flat file from StarBuffer.fbs

struct Star {
double radius ;
double mass;
double volume;
};

int main (int argc, char *argv[])
{
zmq::context_t context = zmq::context_t(1);

// Connect our subscriber socket
zmq::socket_t subSocket = zmq::socket_t(context, ZMQ_SUB);
subSocket.setsockopt(ZMQ_IDENTITY, "Hello", 5);
subSocket.setsockopt(ZMQ_SUBSCRIBE, "", 0);
subSocket.connect("tcp://localhost:5565");

// Get updates, expect random Ctrl-C death
while (1) {
zmq::message_t receiveMessage;
subSocket.recv(&receiveMessage);

flatbuffers::FlatBufferBuilder fbb;
StarBufferBuilder builder(fbb);

auto star = GetStarBuffer(receiveMessage.data());

std::cout << "Received Star" << std::endl;
std::cout << "radius: " << star->radius() << std::endl;
std::cout << "mass: " << star->mass() << std::endl;
std::cout << "volume: " << star->volume() << std::endl;
}
return 0;
}


publisher.cpp:

//
// Publisher for durable subscriber
//
// Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>

#include "zhelpers.hpp"
#include <zmq.hpp>
#include <random>
#include "StarBuffer.h" // generated flat file from StarBuffer.fbs

struct Star {
double radius ;
double mass;
double volume;
};

template<typename Numeric, typename Generator = std::mt19937>
Numeric random(Numeric from, Numeric to)
{
thread_local static Generator gen(std::random_device{}());

using dist_type = typename std::conditional
<
std::is_integral<Numeric>::value
, std::uniform_int_distribution<Numeric>
, std::uniform_real_distribution<Numeric>
>::type;

thread_local static dist_type dist;

return dist(gen, typename dist_type::param_type{from, to});
}


Star getRandomStar() {
double lower_bound = 0;
double upper_bound = std::numeric_limits<double>::max();
double randomRadius = random(lower_bound, upper_bound);
double randomMass = random(lower_bound, upper_bound);
double randomVolume = random(lower_bound, upper_bound);

Star s = Star();
s.radius = randomRadius;
s.mass = randomMass;
s.volume = randomVolume;

return s;
}

int main () {

zmq::context_t context = zmq::context_t(1);

// We send updates via this socket
zmq::socket_t publishSocket = zmq::socket_t(context, ZMQ_PUB);
publishSocket.bind("tcp://*:5565");

// Now broadcast exactly 10 updates with pause
int update_nbr;
for (update_nbr = 0; update_nbr < 100; update_nbr++) {

Star randomStar = getRandomStar();

flatbuffers::FlatBufferBuilder fbb;
StarBufferBuilder builder(fbb);

builder.add_radius(randomStar.radius);
builder.add_mass(randomStar.mass);
builder.add_volume(randomStar.volume);

auto response = builder.Finish();
fbb.Finish(response);

std::cout << "Sending Star " << update_nbr << "…" << std::endl;
int buffersize = fbb.GetSize();
zmq::message_t request(buffersize);
memcpy((void *)request.data(), fbb.GetBufferPointer(), buffersize);
publishSocket.send(request);
std::cout << "Star sent!" << std::endl;

sleep(1);
}
return 0;
}


EDIT: The code compiles and works! I've updated the code here to reflect my working code on my local machine for people to use going forward as this took me a good amount of time to finish.

Since it took me a while to figure this stuff out too here's the build process:

flatc --cpp StarBuffer.fbs
mv StarBuffer_generated.h StarBuffer.h
g++ -std=c++11 publisher.cpp -lzmq -o ./bin/zmq_pub
g++ -std=c++11 subscriber.cpp -lzmq -o ./bin/zmq_sub


Requires 'zhelpers.cpp', and 'flatbuffers/flatbuffers.h' in your local directory.

Answer

Try auto star = GetRootAsStarBuffer(request.data());. The Create function you used is to create a buffer, not to read an existing one.

Unrelated, but on the writing side, you should see if you can create the message_t without having to use memcpy, which would be more efficient.