s2e-systems / dust-dds Goto Github PK

View Code? Open in Web Editor NEW

62.0 3.0 8.0 11.11 MB

Rust implementation of the Data Distribution Service (DDS)

License: Apache License 2.0

Rust 98.15% Dockerfile 0.12% CMake 0.26% C 0.88% C++ 0.59%

data-centric dds middleware rust udp dcps

dust-dds's Issues

Receive messages from other topics when communicate with cyclonedds

Based on dust_subsciber and cyclonedds_publisher

When I create two readers with different topic:

use dust_dds::{
    domain::domain_participant::DomainParticipant,
    domain::domain_participant_factory::DomainParticipantFactory,
    infrastructure::{
        listeners::NoOpListener,
        qos::{DataReaderQos, QosKind},
        qos_policy::{
            DurabilityQosPolicy, DurabilityQosPolicyKind, ReliabilityQosPolicy,
            ReliabilityQosPolicyKind,
        },
        status::{StatusKind, NO_STATUS},
        time::{Duration, DurationKind},
        wait_set::{Condition, WaitSet},
    },
    subscription::sample_info::{ANY_INSTANCE_STATE, ANY_SAMPLE_STATE, ANY_VIEW_STATE},
    subscription::subscriber::Subscriber,
};

mod hello_world {
    include!("target/idl/hello_world.rs");
}

fn create_reader_task(
    topic_name: String,
    participant: &DomainParticipant,
    subscriber: &Subscriber,
) {
    println!("Begin to create reader task, topic[{topic_name}]");
    let topic = participant
        .find_topic::<hello_world::HelloWorldType>(&topic_name, Duration::new(120, 0))
        .unwrap();

    let reader_qos = DataReaderQos {
        reliability: ReliabilityQosPolicy {
            kind: ReliabilityQosPolicyKind::Reliable,
            max_blocking_time: DurationKind::Finite(Duration::new(1, 0)),
        },
        durability: DurabilityQosPolicy {
            kind: DurabilityQosPolicyKind::TransientLocal,
        },
        ..Default::default()
    };
    let reader = subscriber
        .create_datareader::<hello_world::HelloWorldType>(
            &topic,
            QosKind::Specific(reader_qos),
            NoOpListener::new(),
            NO_STATUS,
        )
        .unwrap();
    let topic_name_clone = topic_name.to_string();
    std::thread::spawn(move || {
        loop {
            let sample_result = reader.take_next_sample();
            if let Err(error) = sample_result {
                std::thread::sleep(std::time::Duration::from_millis(100));
                continue;
            }
            let message = sample_result.unwrap().data().unwrap();
            println!(
                "[{}] receive message, message[{:?}]",
                topic_name_clone, message
            );
            if topic_name_clone != message.msg {
                println!(
                    "========Topic confustion, reader_topic[{}] message_topic[{}]",
                    topic_name_clone, message.msg
                );
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        } // loop
    });
    println!("Succ to create reader task, topic[{topic_name}]");
}

fn main() {
    let domain_id = 0;
    let participant_factory = DomainParticipantFactory::get_instance();

    let participant = participant_factory
        .create_participant(domain_id, QosKind::Default, NoOpListener::new(), NO_STATUS)
        .unwrap();

    let subscriber = participant
        .create_subscriber(QosKind::Default, NoOpListener::new(), NO_STATUS)
        .unwrap();
    let topics = vec!["t1", "t2"];
    for topic_name in topics {
        create_reader_task(topic_name.to_string(), &participant, &subscriber);
    }

    // Sleep to allow sending acknowledgements
    loop {
        std::thread::sleep(std::time::Duration::from_secs(2));
    }
}

And add two writers:

#include "ddsc/dds.h"
#include "HelloWorld.h"

int main(int argc, char *argv[])
{
	printf("Begin to create participant\n");
	const dds_entity_t participant = dds_create_participant(0, NULL /*qos*/, NULL /*listener*/);
	if (participant < 0)
	{
		DDS_FATAL("dds_create_participant: %s\n", dds_strretcode(-participant));
	}

	// topic : t1
	printf("Begin to create writer t1\n");
	const char *topic_name = "t1";
	const dds_entity_t topic = dds_create_topic(participant, &HelloWorldType_desc, topic_name, NULL /*qos*/, NULL /*listener*/);
	if (topic < 0)
	{
		DDS_FATAL("dds_create_topic: %s\n", dds_strretcode(-topic));
	}
	dds_qos_t *qos = dds_create_qos();
	dds_qset_reliability(qos, DDS_RELIABILITY_RELIABLE, DDS_SECS(1));
	dds_qset_durability(qos, DDS_DURABILITY_TRANSIENT_LOCAL);

	const dds_entity_t data_writer = dds_create_writer(participant, topic, qos, NULL /*listener*/);
	if (data_writer < 0)
	{
		DDS_FATAL("dds_create_writer: %s\n", dds_strretcode(-data_writer));
	}
	// topic : t2
	printf("Begin to create writer t2\n");
	const char *topic_name_2 = "t2";
	const dds_entity_t topic_2 = dds_create_topic(participant, &HelloWorldType_desc, topic_name_2, NULL /*qos*/, NULL /*listener*/);
	if (topic_2 < 0)
	{
		DDS_FATAL("dds_create_topic: %s\n", dds_strretcode(-topic));
	}
	dds_qos_t *qos_2 = dds_create_qos();
	dds_qset_reliability(qos_2, DDS_RELIABILITY_RELIABLE, DDS_SECS(1));
	dds_qset_durability(qos_2, DDS_DURABILITY_TRANSIENT_LOCAL);

	const dds_entity_t data_writer_2 = dds_create_writer(participant, topic_2, qos_2, NULL /*listener*/);
	if (data_writer_2 < 0)
	{
		DDS_FATAL("dds_create_writer: %s\n", dds_strretcode(-data_writer_2));
	}
	uint32_t id = 0;
	printf("Begin to write msg\n");
	while (true)
	{
		HelloWorldType msg_1 = {id, "t1"};
		printf("Succ to write msg, id[%d], msg[%s]\n", msg_1.id, msg_1.msg);
		dds_write(data_writer, &msg_1);
		HelloWorldType msg_2 = {id, "t2"};
		printf("Succ to write msg, id[%d], msg[%s]\n", msg_2.id, msg_2.msg);
		dds_write(data_writer_2, &msg_2);
		id++;
		sleep(1);
	}
}

HelloWorld.idl:

struct HelloWorldType
{
	unsigned long id;
	string msg;
};

The message received by the reader will be confused:

dust_dds_reader:

    Finished dev [unoptimized + debuginfo] target(s) in 2.59s
     Running `target/debug/dust_dds_subscriber`
Begin to create reader task, topic[t1]
Succ to create reader task, topic[t1]
Begin to create reader task, topic[t2]
Succ to create reader task, topic[t2]
[t2] receive message, message[HelloWorldType { id: 28, msg: "t2" }]
[t1] receive message, message[HelloWorldType { id: 29, msg: "t2" }]
========Topic confustion, reader_topic[t1] message_topic[t2]
[t2] receive message, message[HelloWorldType { id: 29, msg: "t2" }]
[t1] receive message, message[HelloWorldType { id: 30, msg: "t2" }]
========Topic confustion, reader_topic[t1] message_topic[t2]
[t2] receive message, message[HelloWorldType { id: 30, msg: "t2" }]
[t1] receive message, message[HelloWorldType { id: 31, msg: "t2" }]
========Topic confustion, reader_topic[t1] message_topic[t2]
[t2] receive message, message[HelloWorldType { id: 31, msg: "t2" }]
[t1] receive message, message[HelloWorldType { id: 32, msg: "t2" }]
========Topic confustion, reader_topic[t1] message_topic[t2]
[t2] receive message, message[HelloWorldType { id: 32, msg: "t2" }]
[t1] receive message, message[HelloWorldType { id: 33, msg: "t2" }]
========Topic confustion, reader_topic[t1] message_topic[t2]
[t2] receive message, message[HelloWorldType { id: 33, msg: "t2" }]
[t1] receive message, message[HelloWorldType { id: 34, msg: "t2" }]
========Topic confustion, reader_topic[t1] message_topic[t2]
[t2] receive message, message[HelloWorldType { id: 34, msg: "t2" }]
[t1] receive message, message[HelloWorldType { id: 35, msg: "t2" }]
========Topic confustion, reader_topic[t1] message_topic[t2]
[t2] receive message, message[HelloWorldType { id: 35, msg: "t2" }]

cyclonedds_writer:

Begin to create participant
Begin to create writer t1
Begin to create writer t2
Begin to write msg
Succ to write msg, id[0], msg[t1]
Succ to write msg, id[0], msg[t2]
Succ to write msg, id[1], msg[t1]
Succ to write msg, id[1], msg[t2]
Succ to write msg, id[2], msg[t1]
Succ to write msg, id[2], msg[t2]
Succ to write msg, id[3], msg[t1]
Succ to write msg, id[3], msg[t2]
Succ to write msg, id[4], msg[t1]
Succ to write msg, id[4], msg[t2]
Succ to write msg, id[5], msg[t1]
Succ to write msg, id[5], msg[t2]
Succ to write msg, id[6], msg[t1]
Succ to write msg, id[6], msg[t2]
Succ to write msg, id[7], msg[t1]
Succ to write msg, id[7], msg[t2]
Succ to write msg, id[8], msg[t1]
....

If I only use cyclonedds or rust_dds for writers and readers, everything is ok.
The same works if I create two participants.

Scope of types supported for dds_gen

Hello, I am a newbie and recently I am trying to test interoperability functions with xtypes, and I am wondering what features of OMG IDL are implemented currently. Basically I tried to compile i11eperf IDL (https://github.com/eboasson/i11eperf/blob/master/idl/i11eperf.idl) and got errors as follows:

I tried to remove those annotators on structs, and got another error:

I am wondering if these are expected or there are potential problems. Thanks in advance!

Cannot Run DDS Subscriber Within Tokio Context

Since #169, when attempting to run a subscriber within a Tokio context, an error occurs with the message "Cannot block the current thread from within a runtime." This issue seems to be triggered by this blocking:

dust-dds/dds/src/implementation/utils/actor.rs

Line 68 in 9eb89c7

.blocking_recv()

To replicate the error, annotate one of the example subscribers with the #[tokio::main] attribute.

Option Serialization Support

I'm inquiring about the possibility of adding serialization support for Option types?
Are there any existing mechanisms for working with Option types in DDS messages as of now?

DurationKind incompatible with json, ron deserialization

I'm trying to save/load some QoS parameters in a text file for configuration purposes. I noticed that DurationKind has a custom Serialize/Deserialize impl that is causing issues when trying to deserialize it from a JSON or RON strings. Here's what I mean:

use dust_dds::infrastructure::time::{Duration, DurationKind};

#[derive(serde::Serialize, serde::Deserialize, Debug)]
enum AltDurationKind {
    Finite(Duration),
    Infinite,
}

#[test]
fn duration_kind_serde() -> anyhow::Result<()> {
    let duration_kind = DurationKind::Finite(Duration::new(0, 0));
    let alt_duration_kind = AltDurationKind::Finite(Duration::new(0, 0));

    let serialized = bincode::serialize(&duration_kind)?;
    println!("bincode serialized normal: {serialized:?}");
    let result = bincode::deserialize::<DurationKind>(&serialized);
    println!("deserialized: {result:?}");

    let serialized = bincode::serialize(&alt_duration_kind)?;
    println!("bincode serialized alt: {serialized:?}");
    let result = bincode::deserialize::<AltDurationKind>(&serialized);
    println!("deserialized: {result:?}");

    let serialized = serde_json::to_string(&duration_kind)?;
    println!("json serialized normal: {serialized:?}");
    let result = serde_json::from_str::<DurationKind>(&serialized);
    println!("deserialized: {result:?}");

    let serialized = serde_json::to_string(&alt_duration_kind)?;
    println!("json serialized alt: {serialized:?}");
    let result = serde_json::from_str::<AltDurationKind>(&serialized);
    println!("deserialized: {result:?}");

    let serialized = ron::to_string(&duration_kind)?;
    println!("ron serialized normal: {serialized:?}");
    let result = ron::from_str::<DurationKind>(&serialized);
    println!("deserialized: {result:?}");

    let serialized = ron::to_string(&alt_duration_kind)?;
    println!("ron serialized alt: {serialized:?}");
    let result = ron::from_str::<AltDurationKind>(&serialized);
    println!("deserialized: {result:?}");

    Ok(())
}

Output:

bincode serialized normal: [0, 0, 0, 0, 0, 0, 0, 0]
deserialized: Ok(Finite(Duration { sec: 0, nanosec: 0 }))
bincode serialized alt: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
deserialized: Ok(Finite(Duration { sec: 0, nanosec: 0 }))
json serialized normal: "{\"sec\":0,\"nanosec\":0}"
deserialized: Err(Error("invalid type: map, expected DurationKind", line: 1, column: 0))
json serialized alt: "{\"Finite\":{\"sec\":0,\"nanosec\":0}}"
deserialized: Ok(Finite(Duration { sec: 0, nanosec: 0 }))
ron serialized normal: "(sec:0,nanosec:0)"
deserialized: Err(SpannedError { code: ExpectedInteger, position: Position { line: 1, col: 2 } })
ron serialized alt: "Finite((sec:0,nanosec:0))"
deserialized: Ok(Finite(Duration { sec: 0, nanosec: 0 }))

I don't understand serde well enough to say how you could make DurationKind serialize/deserialize properly to/from in all cases. I think it has something to do with how serde handles enums.

The only suggestion I have is to replace DurationKind with just Duration and use a constant for infinite duration instead, like RustDDS does, but I realize this might seem less idiomatic than an enum.

EDIT: I've just noticed that ReliabilityQosPolicyKind (and possibly others?) has the same problem when deserializing with serde_json, but ron seems to work OK.

Remove Sync trait bound for create_datareader, AnyDataReaderListener impl

I needed to create a DataReader with a listener that holds a Box<dyn SerialPort> as returned from this function (SerialPort implies Send but not Sync). At first I couldn't because the create_datareader function requires the listener trait object to be Sync. Then I tried deleting the Sync bound in my local patch of dust_dds, which required also removing it from this AnyDataReaderListener impl. Then dust_dds and my project compiled and ran fine with the SerialPort listener.

Is there a good reason for this bound or could it be removed?

If it could be removed in this case, there may be other similar cases in the code base that it could be removed as well.

[ENTITYID_BUILTIN_PUBLICATIONS_WRITER] Should we choose to continue processing sub-messages even if the sequence numbers are not consecutive?

Our project has a scenario of communicating with cyclonedds.

In some cases, the data sent by cyclonedds has sequence_number discontinuities:

Real-Time Publish-Subscribe Wire Protocol
    Magic: RTPS
    Protocol version: 2.1
    vendorId: 01.16 (Unknown)
    guidPrefix: 01103de4972c41380c170d32
    Default port mapping: domainId=200, participantIdx=103, nature=UNICAST_METATRAFFIC
    submessageId: INFO_DST (0x0e)
        Flags: 0x01, Endianness bit
        octetsToNextHeader: 12
        guidPrefix: c0a87a01c8641c0000000000
    submessageId: INFO_TS (0x09)
        Flags: 0x01, Endianness bit
        octetsToNextHeader: 8
        Timestamp: Apr 28, 2024 10:02:01.270296175 UTC
    submessageId: DATA (0x15)
        Flags: 0x05, Data present, Endianness bit
        octetsToNextHeader: 308
        0000 0000 0000 0000 = Extra flags: 0x0000
        Octets to inline QoS: 16
        readerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_READER (0x000003c7)
        writerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_WRITER (0x000003c2)
        writerSeqNumber: 1
        serializedData
            encapsulation kind: PL_CDR_LE (0x0003)
            encapsulation options: 0x0000
            serializedData:
    submessageId: INFO_TS (0x09)
        Flags: 0x01, Endianness bit
        octetsToNextHeader: 8
        Timestamp: Apr 28, 2024 10:02:01.374387094 UTC
    submessageId: DATA (0x15)
        Flags: 0x05, Data present, Endianness bit
        octetsToNextHeader: 316
        0000 0000 0000 0000 = Extra flags: 0x0000
        Octets to inline QoS: 16
        readerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_READER (0x000003c7)
        writerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_WRITER (0x000003c2)
        writerSeqNumber: 4
        serializedData
            encapsulation kind: PL_CDR_LE (0x0003)
            encapsulation options: 0x0000
            serializedData:
                PID_TOPIC_NAME
                    parameterId: PID_TOPIC_NAME (0x0005)
                    parameterLength: 36
                    topic: /aos/component/latency_report
                PID_TYPE_NAME
                    parameterId: PID_TYPE_NAME (0x0007)
                    parameterLength: 40
                    typeName: aos::transport::cyclone::DDSMessage
                PID_LIVELINESS
                    parameterId: PID_LIVELINESS (0x001b)
                    parameterLength: 12
                    PID_LIVELINESS
                PID_RELIABILITY
                    parameterId: PID_RELIABILITY (0x001a)
                    parameterLength: 12
                    Kind: RELIABLE_RELIABILITY_QOS (0x00000002)
                PID_HISTORY
                    parameterId: PID_HISTORY (0x0040)
                    parameterLength: 8
                    Kind: KEEP_LAST_HISTORY_QOS (0x00000000)
                    Depth: 128
                PID_DATA_REPRESENTATION
                    parameterId: PID_DATA_REPRESENTATION (0x0073)
                    parameterLength: 8
                    parameterData: 0200000000000200
                Unknown (0x0075)
                    parameterId: Unknown (0x0075)
                    parameterLength: 100
                    parameterData: 6000000001100040280000002400000014000000f1f66653…
                PID_PROTOCOL_VERSION
                    parameterId: PID_PROTOCOL_VERSION (0x0015)
                    parameterLength: 4
                    Protocol version: 2.1
                PID_VENDOR_ID
                    parameterId: PID_VENDOR_ID (0x0016)
                    parameterLength: 4
                    vendorId: 01.16 (Unknown)
                PID_ENDPOINT_GUID
                    parameterId: PID_ENDPOINT_GUID (0x005a)
                    parameterLength: 16
                    Endpoint GUID: 01103de4 972c4138 0c170d32 00000803
                Unknown (0x800c)
                    parameterId: Unknown (0x800c)
                    parameterLength: 4
                    parameterData: 01000000
                PID_SENTINEL
                    parameterId: PID_SENTINEL (0x0001)
    submessageId: INFO_TS (0x09)
        Flags: 0x01, Endianness bit
        octetsToNextHeader: 8
        Timestamp: Apr 28, 2024 10:02:01.375171175 UTC
    submessageId: DATA (0x15)
        Flags: 0x05, Data present, Endianness bit
        octetsToNextHeader: 304
        0000 0000 0000 0000 = Extra flags: 0x0000
        Octets to inline QoS: 16
        readerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_READER (0x000003c7)
        writerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_WRITER (0x000003c2)
        writerSeqNumber: 5
        serializedData
            encapsulation kind: PL_CDR_LE (0x0003)
            encapsulation options: 0x0000
            serializedData:
                PID_TOPIC_NAME
                    parameterId: PID_TOPIC_NAME (0x0005)
                    parameterLength: 24
                    topic: aos_topic_change
                PID_TYPE_NAME
                    parameterId: PID_TYPE_NAME (0x0007)
                    parameterLength: 40
                    typeName: aos::transport::cyclone::DDSMessage
                PID_LIVELINESS
                    parameterId: PID_LIVELINESS (0x001b)
                    parameterLength: 12
                    PID_LIVELINESS
                PID_RELIABILITY
                    parameterId: PID_RELIABILITY (0x001a)
                    parameterLength: 12
                    Kind: RELIABLE_RELIABILITY_QOS (0x00000002)
                PID_HISTORY
                    parameterId: PID_HISTORY (0x0040)
                    parameterLength: 8
                    Kind: KEEP_LAST_HISTORY_QOS (0x00000000)
                    Depth: 128
                PID_DATA_REPRESENTATION
                    parameterId: PID_DATA_REPRESENTATION (0x0073)
                    parameterLength: 8
                    parameterData: 0200000000000200
                Unknown (0x0075)
                    parameterId: Unknown (0x0075)
                    parameterLength: 100
                    parameterData: 6000000001100040280000002400000014000000f1f66653…
                PID_PROTOCOL_VERSION
                    parameterId: PID_PROTOCOL_VERSION (0x0015)
                    parameterLength: 4
                    Protocol version: 2.1
                PID_VENDOR_ID
                    parameterId: PID_VENDOR_ID (0x0016)
                    parameterLength: 4
                    vendorId: 01.16 (Unknown)
                PID_ENDPOINT_GUID
                    parameterId: PID_ENDPOINT_GUID (0x005a)
                    parameterLength: 16
                    Endpoint GUID: 01103de4 972c4138 0c170d32 00000b03
                Unknown (0x800c)
                    parameterId: Unknown (0x800c)
                    parameterLength: 4
                    parameterData: 01000000
                PID_SENTINEL
                    parameterId: PID_SENTINEL (0x0001)
    submessageId: INFO_TS (0x09)
        Flags: 0x01, Endianness bit
        octetsToNextHeader: 8
        Timestamp: Apr 28, 2024 10:02:01.476425548 UTC
    submessageId: DATA (0x15)
        Flags: 0x05, Data present, Endianness bit
        octetsToNextHeader: 316
        0000 0000 0000 0000 = Extra flags: 0x0000
        Octets to inline QoS: 16
        readerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_READER (0x000003c7)
        writerEntityId: ENTITYID_BUILTIN_PUBLICATIONS_WRITER (0x000003c2)
        writerSeqNumber: 6
        serializedData
            encapsulation kind: PL_CDR_LE (0x0003)
            encapsulation options: 0x0000
            serializedData:
                PID_TOPIC_NAME
                    parameterId: PID_TOPIC_NAME (0x0005)
                    parameterLength: 36
                    topic: /aos/component/heartbeat_report
                PID_TYPE_NAME
                    parameterId: PID_TYPE_NAME (0x0007)
                    parameterLength: 40
                    typeName: aos::transport::cyclone::DDSMessage
                PID_LIVELINESS
                    parameterId: PID_LIVELINESS (0x001b)
                    parameterLength: 12
                    PID_LIVELINESS
                PID_RELIABILITY
                    parameterId: PID_RELIABILITY (0x001a)
                    parameterLength: 12
                    Kind: RELIABLE_RELIABILITY_QOS (0x00000002)
                PID_HISTORY
                    parameterId: PID_HISTORY (0x0040)
                    parameterLength: 8
                    Kind: KEEP_LAST_HISTORY_QOS (0x00000000)
                    Depth: 128
                PID_DATA_REPRESENTATION
                    parameterId: PID_DATA_REPRESENTATION (0x0073)
                    parameterLength: 8
                    parameterData: 0200000000000200
                Unknown (0x0075)
                    parameterId: Unknown (0x0075)
                    parameterLength: 100
                    parameterData: 6000000001100040280000002400000014000000f1f66653…
                PID_PROTOCOL_VERSION
                    parameterId: PID_PROTOCOL_VERSION (0x0015)
                    parameterLength: 4
                    Protocol version: 2.1
                PID_VENDOR_ID
                    parameterId: PID_VENDOR_ID (0x0016)
                    parameterLength: 4
                    vendorId: 01.16 (Unknown)
                PID_ENDPOINT_GUID
                    parameterId: PID_ENDPOINT_GUID (0x005a)
                    parameterLength: 16
                    Endpoint GUID: 01103de4 972c4138 0c170d32 00000d03
                Unknown (0x800c)
                    parameterId: Unknown (0x800c)
                    parameterLength: 4
                    parameterData: 01000000
                PID_SENTINEL
                    parameterId: PID_SENTINEL (0x0001)

So when encountering this situation (serial number changes from 1 to 4), Can we choose to raise an alarm and continue processing these sub-messages?

Comparison to RustDDS

Hi! This looks like a really cool project! Thanks for sharing it!

I did notice that there's another Rust implementation of DDS, RustDDS and I'm trying to decide which one to use, so I'd like to suggest that you clarify the important technical and non-technical differences between these projects so that developers like me know how to choose. I've seen some projects post an FAQ right on their main README for visibility to newcomers.

Specifically, my questions right now are:

What is motivating this development effort separate from RustDDS? Why not just contribute to a that project, or a fork?
What are S2E Systems long term support intentions for this project?
What is the currently implemented set of DDS features, what's not implemented, and what's planned to be implemented.

I also noticed that y'all haven't had any issues yet, so I thought I'd kick things off for you 😄

Interoperability with FastDDS

I'm trying to test interoperability with FastDDS on Windows 11. You already test interop with CycloneDDS, right?

My test so far goes like this:

Download the FastDDS Shapes Demo app.
- This requires filling out a form, but is pretty easy.
Run the contained bin/ShapesDemo.exe.
In the GUI, click "Publish" and accept the default settings to create a red square publisher.
Run the test programs below.

My Dust DDS test program is as follows, using version 0.4.0:

use dust_dds::{
    domain::domain_participant_factory::DomainParticipantFactory,
    infrastructure::{error::DdsResult, qos::QosKind},
};
use std::{thread, time::Duration};

fn main() -> DdsResult<()> {
    let participant = DomainParticipantFactory::get_instance().create_participant(
        0,
        QosKind::Default,
        None,
        &[],
    )?;
    loop {
        println!(
            "{:?}",
            participant
                .get_discovered_topics()?
                .into_iter()
                .map(|topic_handle| participant.get_discovered_topic_data(topic_handle).unwrap())
                .collect::<Vec<_>>()
        );
        thread::sleep(Duration::from_secs(1));
    }
}

The result is empty:

[]
[]
...

For comparison, I tried the same thing using RustDDS:

use std::{thread, time::Duration};
use anyhow::Result;
use rustdds::DomainParticipant;

fn main() -> Result<()> {
    let participant = DomainParticipant::new(0)?;
    loop {
        println!("{:?}", participant.discovered_topics());
        thread::sleep(Duration::from_secs(1));
    }
}

and I get the following output:

[]
[]
[]
[DiscoveredTopicData { updated_time: 2023-09-19T20:28:12.684058600Z, topic_data: TopicBuiltinTopicData { key: None, name: "Square", type_name: "ShapeType", durability: Some(Volatile), deadline: Some(Deadline(Duration { seconds: 2147483647, fraction: 4294967295 })), latency_budget: Some(LatencyBudget { duration: Duration { seconds: 0, fraction: 0 } }), liveliness: Some(Automatic { lease_duration: Duration { seconds: 2147483647, fraction: 4294967295 } }), reliability: Some(Reliable { max_blocking_time: Duration { seconds: 0, fraction: 429496730 } }), lifespan: Some(Lifespan { duration: Duration { seconds: 2147483647, fraction: 4294967295 } }), destination_order: Some(ByReceptionTimestamp), presentation: Some(Presentation { access_scope: Instance, coherent_access: false, ordered_access: false }), history: None, resource_limits: None, ownership: Some(Shared) } }]
...

So I'm wondering if there's an incompatibility or if I'm just using the Dust DDS API wrong?

Better configuration

In general, it is the configuration of DustDDS does not scale and is hard to work with. I'm opening this issue to discuss possible alternatives.

Inconsistent documentation on interfaces:
The docs mention that you can pass an array of interface names as configuration option: [string, null].
However, the implementation only accepts one String value xor null.

Env variables does not scale well
Environment variables are only ok for small number of config options and get cumbersome to use with great number of values.

Possible solutions:

RTI Connext uses XML
Eclipse CycloneDDS also uses XML

I think a TOML based configuration file would fit well into the rustacean philosophy. Then the configuration file path could be provided like DUST_DDS_CONFIGURATION=/etc/dds/config.toml.
I'd be glad to hear the opinion of the maintainers! 🙂

Why can't I receive it correctly using the example in the warehouse?

I am a newbie. First I run hello_world_subscriber.rs in the example, and then I run hello_world_publisher.rs, but I cannot receive the response correctly.

failed to resolve: could not find `implementation` in the crate root When use actor_interface

dust_dds_derive v0.8.2.
https://docs.rs/dust_dds_derive/0.8.2/dust_dds_derive/attr.actor_interface.html

use dust_dds::{
    dds_async::{topic::TopicAsync, topic_listener::TopicListenerAsync},
    infrastructure::status::InconsistentTopicStatus,
};
use dust_dds_derive::actor_interface;
pub struct TopicListenerActor {
    listener: Option<Box<dyn TopicListenerAsync + Send>>,
}

#[actor_interface]
impl TopicListenerActor {
    async fn on_inconsistent_topic(
        &mut self,
        the_topic: TopicAsync,
        status: InconsistentTopicStatus,
    ) {
        if let Some(l) = &mut self.listener {
            l.on_inconsistent_topic(the_topic, status).await
        }
    }
}

fn main() {
    println!("Hello, world!");
}

$ cargo run
   Compiling dust_dds_examples v0.1.0 
error[E0433]: failed to resolve: could not find `implementation` in the crate root
  --> src/main.rs:10:1
   |
10 | #[actor_interface]
   | ^^^^^^^^^^^^^^^^^^ could not find `implementation` in the crate root
   |
   = note: this error originates in the attribute macro `actor_interface` (in Nightly builds, run with -Z macro-backtrace for more info)

For more information about this error, try `rustc --explain E0433`.
error: could not compile `dust_dds_examples` (bin "dust_dds_examples") due to previous error

Cannot start a runtime from within a runtime

dustdds not working in tokio main:

use common::dds_message::aos::transport::cyclone::DDSMessage;
use dust_dds::{
    domain::domain_participant_factory::DomainParticipantFactory,
    infrastructure::{listeners::NoOpListener, qos::QosKind, status::NO_STATUS},
};
#[tokio::main]
async fn main() {
    let domain_id = 0;
    let participant_factory = DomainParticipantFactory::get_instance();

    let participant = participant_factory
        .create_participant(domain_id, QosKind::Default, NoOpListener::new(), NO_STATUS)
        .unwrap();
}

Inconsistent Behavior with DataWriterQos "HistoryQosPolicyKind::KeepLast" in a Publisher-Subscriber Scenario

I have encountered a puzzling issue while working with a data writer configured with a specific Quality of Service (QoS) policy.
Here's my QoS configuration:

let writer_qos = DataWriterQos {
    reliability: ReliabilityQosPolicy {
        kind: ReliabilityQosPolicyKind::Reliable,
        max_blocking_time: DurationKind::Finite(Duration::new(1, 0)),
    },
    durability: DurabilityQosPolicy {
        kind: DurabilityQosPolicyKind::TransientLocal,
    },
    history: HistoryQosPolicy {
        kind: HistoryQosPolicyKind::KeepLast(1),
    },
    ..Default::default()
};

My setup involves a data writer that writes data 10 times, and then a data reader is initiated with the same QoS settings. Surprisingly, I observe the following behavior:

When the HistoryQosPolicyKind is set to KeepLast(1), I receive the first and the last message.
If I modify the HistoryQosPolicy to KeepLast(2), I receive the first, the second, and the last messages.

I expected to only receive the last message when the HistoryQosPolicyKind is set to KeepLast(1). Why am I receiving both the first and last messages in this case?
Furthermore, when I set the HistoryQosPolicyKind to KeepLast(2), I anticipate receiving only the last two messages, not the first, second, and last messages. Is this behavior in line with the expected outcome?

Additional Context:
I'm using the hello_world_publisher with the same message ID for all the messages, only changing the message content during each write operation.

DdsType derivation macro support for nested types

Hello, I was wondering if there is any way/any plans to make it possible to apply the DdsType derivation macro to structs with nested types?

The purpose of this is to create topic listeners and publishers that can handle XML messages with nested fields. Having used the IDL compiler on the IDLs for this specific XML standard, the generated structs are then non compileable as the DdsType derivation macro panics.

Apologies for the lack of screenshots, but it would be against company policy

s2e-systems / dust-dds Goto Github PK

dust-dds's Issues

Receive messages from other topics when communicate with cyclonedds

Scope of types supported for dds_gen

Cannot Run DDS Subscriber Within Tokio Context

Option Serialization Support

DurationKind incompatible with json, ron deserialization

Remove Sync trait bound for create_datareader, AnyDataReaderListener impl

[ENTITYID_BUILTIN_PUBLICATIONS_WRITER] Should we choose to continue processing sub-messages even if the sequence numbers are not consecutive?

Comparison to RustDDS

Interoperability with FastDDS

Better configuration

Why can't I receive it correctly using the example in the warehouse?

failed to resolve: could not find `implementation` in the crate root When use actor_interface

Cannot start a runtime from within a runtime

Inconsistent Behavior with DataWriterQos "HistoryQosPolicyKind::KeepLast" in a Publisher-Subscriber Scenario

DdsType derivation macro support for nested types

Recommend Projects

React

Vue.js

Typescript

TensorFlow

Django

Laravel

D3

Recommend Topics

javascript

web

server

Machine learning

Visualization

Game

Recommend Org

Facebook

Microsoft

Google

Alibaba

D3

Tencent