#include <dds/dds.hpp>

dds::core::xtypes::StructType dds_type()
{
	using namespace dds::core::xtypes;
	StructType dynamic_dds_type("TestObj");
	dynamic_dds_type.add_member(Member("x", primitive_type<int>()));
    return dynamic_dds_type;
}

using DynamicDataTopic = dds::topic::Topic<dds::core::xtypes::DynamicData>;
using DynamicDataFilteredTopic = dds::topic::ContentFilteredTopic<dds::core::xtypes::DynamicData>;

using DynamicDataWriter = dds::pub::DataWriter<dds::core::xtypes::DynamicData>;
using DynamicDataReader = dds::sub::DataReader<dds::core::xtypes::DynamicData>;

DynamicDataReader create_filtered_reader(
		const dds::domain::DomainParticipant domain_participant,
		const std::string &parent_topic_name,
		const std::string &cft_name, const std::string &filter_expression)
{
	auto dds_parent_topic = dds::topic::find<DynamicDataTopic>(domain_participant, parent_topic_name);

	dds::topic::Filter filter(filter_expression);
	DynamicDataFilteredTopic dds_filtered_topic(dds_parent_topic, cft_name, filter);

	return DynamicDataReader(dds::sub::Subscriber(domain_participant), dds_filtered_topic);
}


int main(int argc, char* argv[])
{
	dds::domain::DomainParticipant domain_participant(0);

	// 1. initiate the main (shared) topic
	auto dds_dynamic_type = dds_type();
	std::string topic_name = "TestTopic";
	DynamicDataTopic dds_topic(domain_participant, topic_name, dds_dynamic_type);

	auto writer_qos = dds::core::QosProvider::Default().datawriter_qos();
	rti::core::policy::DataWriterResourceLimits resource_limits;
	writer_qos << resource_limits.max_remote_reader_filters(0);

	// 2. create one DataWriter with a DynamicData type
	DynamicDataWriter dds_writer(dds::pub::Publisher(domain_participant), dds_topic, writer_qos);

	// 3. create the first DataReader with an own ContentFilteredTopic and a specified filter expression
	std::string cft_name1 = topic_name+"_Filtered1";
	std::string filter_expression1 = "x < 2";
	DynamicDataReader dds_subscription_reader1 = create_filtered_reader(domain_participant, topic_name, cft_name1, filter_expression1);

	// 4. create the second DataReader with an own ContentFilteredTopic and a specified filter expression
	std::string cft_name2 = topic_name+"_Filtered2";
	std::string filter_expression2 = "x > 1";
	DynamicDataReader dds_subscription_reader2 = create_filtered_reader(domain_participant, topic_name, cft_name2, filter_expression2);

	dds::core::xtypes::DynamicData data1(dds_dynamic_type);
	dds::core::xtypes::DynamicData data2(dds_dynamic_type);

	data1.value("x", 1);
	data2.value("x", 2);
	// write the two data samples
	dds_writer.write(data1);
	dds_writer.write(data2);

	auto samples1 = dds_subscription_reader1.take();
	for(const auto& sample: samples1) {
		if(sample.info().valid()) {
			std::cout << "reader1 received sample: " << sample.data() << std::endl;
		}
	}
	samples1.return_loan();

	/** the ERROR happens somewhere here ... */

	auto samples2 = dds_subscription_reader2.take();
	for(const auto& sample: samples2) {
		if(sample.info().valid()) {
			std::cout << "reader2 received sample: " << sample.data() << std::endl;
		}
	}
	samples2.return_loan();

	// clean-up the resources
	dds_subscription_reader2 = nullptr;
	dds_subscription_reader1 = nullptr;

	dds_writer = nullptr;
	dds_topic = nullptr;

	domain_participant.close();
	dds::domain::DomainParticipant::finalize_participant_factory();

	return 0;
}