rti.connext 7.7.0

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 Developer Guide helps you install the software and run your first application.

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

• Additional Connext documentation, including the User’s Manual and Release Notes, is available on 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.

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())