Creating Message from ConcreteMessageBuilder

[tgaldes]

Is there a way to use the concrete message builder class to build messages for the Producer to use? The only produce() arguments in producer.h that I see are r/l references to either Messages, or MessageBuilders. I also don't see any functions that return an instance of Message except the one that uses an rd_kafka_message_t* (in message.h:85) as an argument, but if I need to create that myself it seems to make the concrete message builder obsolete, so I know that I'm missing something. An help is much appreciated!

[mfontanini]

BasicMessageBuilders can be constructed from BasicMessageBuilders instantiated with different template types. So you should be able to use a, say, ConcreteMessageBuilder<string> and pass it along to the produce(const MessageBuilder&) overload, which will implicitly construct a MessageBuilder from it.

Regarding the Message question, you can't get a rd_kafka_message our of a message builder because that's created by rdkafka under the hood. There's no way for a user to create those handles. What's the reason why you need a message handle when producing?

[tgaldes]

Thanks for getting back to me!

I don't need a message handle when producing, that's just the only way I saw in the api to construct a message, but using anything from rd_kafka seemed like improper use of the cppkafka api.

For ConcreteMessageBuilder, I don't see how it can be implicitly converted to a MessageBuilder when T != string. ConcreteMessageBuilder and MessageBuilder inherit from BasicMessageBuilder, but Concrete doesn't inherit from MessageBuilder. The example you gave works fine since Concrete is the same type as MessageBuilder, but again, I don't see how to use produce() when T != string

[tgaldes]

From what little experience I have with templates it seems like you would need a templated overload of produce():

template
void produce(const ConcreteMessageBuilder& builder);

[mfontanini]

This should work as it is. It works because of this constructor. Did you actually try doing this?

[tgaldes]

My relevant changes to producer_example.cpp

ConcreteMessageBuilder<double> cb(topic_name); cb.partition(partition_value) cb.payload(0.9); producer.produce(cb); <-- This is the offending line for the following compilation error

In file included from ../include/cppkafka/producer.h:39:0,
from producer_example.cpp:4:
../include/cppkafka/message_builder.h: In instantiation of ‘void cppkafka::MessageBuilder::construct_buffer(cppkafka::Buffer&, const T&) [with T = double]’:
../include/cppkafka/message_builder.h:250:5: required from ‘cppkafka::BasicMessageBuilder<BufferType, Concrete>::BasicMessageBuilder(const cppkafka::BasicMessageBuilder<OtherBufferType, OtherConcrete>&) [with OtherBufferType = double; OtherConcrete = cppkafka::ConcreteMessageBuilder; BufferType = cppkafka::Buffer; Concrete = cppkafka::MessageBuilder]’
../include/cppkafka/message_builder.h:413:32: required from here
../include/cppkafka/message_builder.h:425:15: error: no matching function for call to ‘cppkafka::Buffer::Buffer(const double&)’
lhs = Buffer(rhs);
^~~~~~~~~~~
In file included from ../include/cppkafka/message.h:39:0,
from ../include/cppkafka/event.h:37,
from ../include/cppkafka/configuration.h:45,
from ../include/cppkafka/kafka_handle_base.h:46,
from ../include/cppkafka/producer.h:34,
from producer_example.cpp:4:
../include/cppkafka/buffer.h:111:5: note: candidate: constexpr cppkafka::Buffer::Buffer(cppkafka::Buffer&&)
Buffer(Buffer&&) = default;
^~~~~~
../include/cppkafka/buffer.h:111:5: note: no known conversion for argument 1 from ‘const double’ to ‘cppkafka::Buffer&&’
../include/cppkafka/buffer.h:105:5: note: candidate: cppkafka::Buffer::Buffer(const string&)
Buffer(const std::string& data);
^~~~~~
../include/cppkafka/buffer.h:105:5: note: no known conversion for argument 1 from ‘const double’ to ‘const string& {aka const std::basic_string&}’
../include/cppkafka/buffer.h:97:5: note: candidate: template cppkafka::Buffer::Buffer(std::vector&&)
Buffer(std::vector&& data) = delete;
^~~~~~
../include/cppkafka/buffer.h:97:5: note: template argument deduction/substitution failed:
In file included from ../include/cppkafka/producer.h:39:0,
from producer_example.cpp:4:
../include/cppkafka/message_builder.h:425:15: note: mismatched types ‘std::vector’ and ‘const double’
lhs = Buffer(rhs);
^~~~~~~~~~~
In file included from ../include/cppkafka/message.h:39:0,
from ../include/cppkafka/event.h:37,
from ../include/cppkafka/configuration.h:45,
from ../include/cppkafka/kafka_handle_base.h:46,
from ../include/cppkafka/producer.h:34,
from producer_example.cpp:4:
../include/cppkafka/buffer.h:90:5: note: candidate: template cppkafka::Buffer::Buffer(const std::vector&)
Buffer(const std::vector& data)
^~~~~~
../include/cppkafka/buffer.h:90:5: note: template argument deduction/substitution failed:
In file included from ../include/cppkafka/producer.h:39:0,
from producer_example.cpp:4:
../include/cppkafka/message_builder.h:425:15: note: mismatched types ‘const std::vector’ and ‘const double’
lhs = Buffer(rhs);
^~~~~~~~~~~
In file included from ../include/cppkafka/message.h:39:0,
from ../include/cppkafka/event.h:37,
from ../include/cppkafka/configuration.h:45,
from ../include/cppkafka/kafka_handle_base.h:46,
from ../include/cppkafka/producer.h:34,
from producer_example.cpp:4:
../include/cppkafka/buffer.h:76:5: note: candidate: template cppkafka::Buffer::Buffer(const T*, size_t)
Buffer(const T* data, size_t size)
^~~~~~
../include/cppkafka/buffer.h:76:5: note: template argument deduction/substitution failed:
In file included from ../include/cppkafka/producer.h:39:0,
from producer_example.cpp:4:
../include/cppkafka/message_builder.h:425:15: note: mismatched types ‘const T*’ and ‘double’
lhs = Buffer(rhs);
^~~~~~~~~~~
In file included from ../include/cppkafka/message.h:39:0,
from ../include/cppkafka/event.h:37,
from ../include/cppkafka/configuration.h:45,
from ../include/cppkafka/kafka_handle_base.h:46,
from ../include/cppkafka/producer.h:34,
from producer_example.cpp:4:
../include/cppkafka/buffer.h:67:5: note: candidate: cppkafka::Buffer::Buffer()
Buffer();
^~~~~~
../include/cppkafka/buffer.h:67:5: note: candidate expects 0 arguments, 1 provided

[mfontanini]

You can't really use doubles. How do you encode doubles on the wire? The idea (and this should probably be stated in the API) is that you provide some type of buffer that holds the data (e.g. vector, string, etc). cppkafka (and librdkafka) don't know how you want to encode double, your code should handle the conversion and your ConcreteMessageBuilder should hold some type of buffer.

[tgaldes]

Ah gotcha! Thank you for the clarification of sending over the wire- now I understand why there's the limitation. I got it to work with vector

[tgaldes]

Does your api support using non stl containers like char[]/void* and a size to construct payloads?

[mfontanini]

I don't think that will work. There should be another overload like this one for std::array so at least you can construct them from an array which won't incur a memory allocation. Feel free to add a PR there if you have the time to do so.

[tgaldes]

#138

What's your protocol for testing on pull requests?