rti.connext 7.2.0.dev0

A full Python binding for RTI Connext

RTI® Connext® is a connectivity framework for building distributed applications with requirements for high performance and scalability.

The RTI Connext Python API provides access to most Connext features from Python.

Documentation

• The Connext Getting Started Guide helps you install the software and run your first RTI Connext Python application while learning general concepts of Connext.

• The Connext Python API Reference an overview of the API, additional examples and the API reference.

• Additional Connext DDS documentation, including the User’s Manual is available at the Connext Community Portal.

Hello World Example

The following is the basic code to publish and subscribe to a topic defined by a simple data type. Similar code can be generated by rtiddsgen from an IDL or XML file defining your data types.

Define your types:

# hello.py

import rti.types as idl

@idl.struct
class HelloWorld:
    message: str = ""

Create a DataWriter to publish the HelloWorld Topic:

# hello_publisher.py

import time
import rti.connextdds as dds
from hello import HelloWorld

participant = dds.DomainParticipant(domain_id=0)
topic = dds.Topic(participant, 'HelloWorld Topic', HelloWorld)
writer = dds.DataWriter(participant.implicit_publisher, topic)

for i in range(10):
    writer.write(HelloWorld(message=f'Hello World! #{i}'))
    time.sleep(1)

Create a DataReader to subscribe to the HelloWorld Topic:

# hello_subscriber.py

import rti.connextdds as dds
import rti.asyncio
from hello import HelloWorld

participant = dds.DomainParticipant(domain_id=0)
topic = dds.Topic(participant, 'HelloWorld Topic', HelloWorld)
reader = dds.DataReader(participant.implicit_subscriber, topic)

async def print_data():
    async for data in reader.take_data_async():
        print(f"Received: {data}")

rti.asyncio.run(print_data())