core¶

class cyclonedds.core.Qos(*policies, base=None)[source]¶

Bases: object

This class represents a collections of policies. It allows for easy inspection of this set. When you retrieve a Qos object from an entity modifying that object would actually does not change the Qos of the entity. To reflect this Qos objects are immutable.

x == y: Checks if two Qos objects contain the same policies. This is a full comparison, not a match.

x != y: Checks if two Qos objects do not contain the same policies.

p in qos: Check if a Policy p is contained in Qos object qos. You can use all levels of generalization, for example: Policy.History in qos, Policy.History.KeepLast in qos and Policy.History.KeepLast(1) in qos.

qos[p]: Obtain the Policy matched with p from the Qos object, for example: qos[Policy.History] -> Policy.History.KeepAll

iter(x): The Qos object supports iteration over it’s contents.

len(x): Return the number of Policies in the Qos object.

str(x): Human-readable description of the contained Qos policies.

Parameters: base (Optional[Qos]) –

policies¶

A sorted tuple of the Policies contained in this Qos object

Type: Tuple[BasePolicy]

__init__(*policies, base=None)[source]¶

Initialize a Qos object

Parameters

*policies (BasePolicy) – Pass in any number of constructed Policies.
base (Qos, optional) – Optionally inherit policies from another Qos object. Inherited policies are overwritten by those newly set.

Raises

TypeError – If you pass something that is not a Policy or use a base that is not a Qos object this will be treated as a TypeError.
ValueError – If you pass two overlapping Policies, for example Policy.History.KeepLast(10) and Policy.History.KeepAll this will be treated as a ValueError.

asdict()[source]¶

Convert a Qos object to a python dictionary.

Returns: Fully describe the Qos object using a python dictionary with only built-in types (dict, list, string, int, boolean). This format is not guaranteed to stay consistent between cyclonedds versions but can be useful for debugging or use within an application.
Return type: dict

classmethod fromdict(data)[source]¶

Convert a python dictionary as generated by asdict() to a Qos object.

Returns: Note that the format of the python dictionary is not guaranteed between cyclonedds versions thus storing these dictionaries to disk and loading them again is not recommended.
Return type: Qos
Parameters: data (dict) –

class cyclonedds.core.Policy(**kwargs)[source]¶

The Policy class is fully static and should never need to be instantiated.

See also

qoshowto: How to work with Qos and Policy, TODO.

class Reliability(**kwargs)[source]¶

The Reliability Qos Policy

Examples

>>> Policy.Reliability.BestEffort
>>> Policy.Reliability.Reliable(max_blocking_time=duration(seconds=1))

Reliable[source]¶: alias of cyclonedds.qos.Policy.Reliability.Reliable

class Durability(**kwargs)[source]¶

The Durability Qos Policy

Examples

>>> Policy.Durability.Volatile
>>> Policy.Durability.TransientLocal
>>> Policy.Durability.Transient
>>> Policy.Durability.Persistent

class History(**kwargs)[source]¶

The History Qos Policy

Examples

>>> Policy.History.KeepAll
>>> Policy.History.KeepLast(depth=10)

KeepAll¶

The type of this entity is not publicly specified.

Type: Tuple[PolicyType, Any]

KeepLast[source]¶: alias of cyclonedds.qos.Policy.History.KeepLast

class ResourceLimits(max_samples=- 1, max_instances=- 1, max_samples_per_instance=- 1)[source]¶

The ResourceLimits Qos Policy

Examples

>>> Policy.ResourceLimits(
>>>     max_samples=10,
>>>     max_instances=10,
>>>     max_samples_per_instance=2
>>> )

Parameters

max_samples (int) –
max_instances (int) –
max_samples_per_instance (int) –

Return type

None

max_samples¶

Max number of samples total.

Type: int

max_instances¶

Max number of instances total.

Type: int

max_samples_per_instance¶

Max number of samples per instance.

Type: int

class PresentationAccessScope(**kwargs)[source]¶

The Presentation Access Scope Qos Policy

Examples

>>> Policy.PresentationAccessScope.Instance(coherent_access=True, ordered_access=False)
>>> Policy.PresentationAccessScope.Topic(coherent_access=True, ordered_access=False)
>>> Policy.PresentationAccessScope.Group(coherent_access=True, ordered_access=False)

Instance[source]¶: alias of cyclonedds.qos.Policy.PresentationAccessScope.Instance

Topic[source]¶: alias of cyclonedds.qos.Policy.PresentationAccessScope.Topic

Group[source]¶: alias of cyclonedds.qos.Policy.PresentationAccessScope.Group

class Lifespan(lifespan)[source]¶

The Lifespan Qos Policy

Examples

>>> Policy.Lifespan(duration(seconds=2))

Parameters: lifespan (int) –
Return type: None

lifespan¶

Expiration time relative to the source timestamp of a sample in nanoseconds.

Type: int

class Deadline(deadline)[source]¶

The Deadline Qos Policy

Examples

>>> Policy.Deadline(deadline=duration(seconds=2))

Parameters: deadline (int) –
Return type: None

deadline¶

Deadline of a sample in nanoseconds.

Type: int

class LatencyBudget(budget)[source]¶

The Latency Budget Qos Policy

Examples

>>> Policy.LatencyBudget(duration(seconds=2))

Parameters: budget (int) – Latency budget in nanoseconds.
Return type: None

class Ownership(**kwargs)[source]¶

The Ownership Qos Policy

Examples

>>> Policy.Ownership.Shared
>>> Policy.Ownership.Exclusive

Shared¶

Type: Policy.Ownership.Shared

Exclusive¶

Type: Policy.Ownership.Exclusive

class OwnershipStrength(strength)[source]¶

The Ownership Strength Qos Policy

Examples

>>> Policy.OwnershipStrength(strength=2)

Parameters: strength (int) – Ownership strength as integer.
Return type: None

class Liveliness(**kwargs)[source]¶

The Liveliness Qos Policy

Examples

>>> Policy.Liveliness.Automatic(lease_duration=duration(seconds=10))
>>> Policy.Liveliness.ManualByParticipant(lease_duration=duration(seconds=10))
>>> Policy.Liveliness.ManualByTopic(lease_duration=duration(seconds=10))

Automatic[source]¶: alias of cyclonedds.qos.Policy.Liveliness.Automatic

ManualByParticipant[source]¶: alias of cyclonedds.qos.Policy.Liveliness.ManualByParticipant

ManualByTopic[source]¶: alias of cyclonedds.qos.Policy.Liveliness.ManualByTopic

class TimeBasedFilter(filter_time)[source]¶

The TimeBasedFilter Qos Policy

Examples

>>> Policy.TimeBasedFilter(filter_fn=duration(seconds=2))

Parameters: filter_time (int) –
Return type: None

filter_time¶

Minimum time between samples in nanoseconds. Use the helper function duration to write the duration in a human readable format.

Type: int

class Partition(partitions)[source]¶

The Partition Qos Policy

Examples

>>> Policy.Partition(partitions=["partition_a", "partition_b", "partition_c"])
>>> Policy.Partition(partitions=[f"partition_{i}" for i in range(100)])

Parameters: partitions (Sequence[str]) –
Return type: None

partitions¶

Type: Sequence[str]

class TransportPriority(priority)[source]¶

The TransportPriority Qos Policy

Examples

>>> Policy.TransportPriority(priority=10)

Parameters: priority (int) –
Return type: None

priority¶

Type: int

class DestinationOrder[source]¶

The DestinationOrder Qos Policy

Examples

>>> Policy.DestinationOrder.ByReceptionTimestamp
>>> Policy.DestinationOrder.BySourceTimestamp

class WriterDataLifecycle(autodispose)[source]¶

The WriterDataLifecycle Qos Policy

Examples

>>> Policy.WriterDataLifecycle(autodispose=False)

Parameters: autodispose (bool) –
Return type: None

autodispose¶

Type: bool

class ReaderDataLifecycle(autopurge_nowriter_samples_delay, autopurge_disposed_samples_delay)[source]¶

The ReaderDataLifecycle Qos Policy

Examples

>>> Policy.ReaderDataLifecycle(
>>>     autopurge_nowriter_samples_delay=duration(minutes=2),
>>>     autopurge_disposed_samples_delay=duration(minutes=5)
>>> )

Parameters

autopurge_nowriter_samples_delay (int) –
autopurge_disposed_samples_delay (int) –

Return type

None

autopurge_nowriter_samples_delay¶

Type: bool

autopurge_disposed_samples_delay¶

Type: bool

class DurabilityService(cleanup_delay, history, max_samples, max_instances, max_samples_per_instance)[source]¶

The DurabilityService Qos Policy

Examples

>>> Policy.DurabilityService(
>>>     cleanup_delay=duration(minutes=2.5),
>>>     history=Policy.History.KeepLast(20),
>>>     max_samples=2000,
>>>     max_instances=200,
>>>     max_samples_per_instance=25
>>> )

Parameters

cleanup_delay (int) –
history (Union[Policy.History.KeepAll, Policy.History.KeepLast]) –
max_samples (int) –
max_instances (int) –
max_samples_per_instance (int) –

Return type

None

cleanup_delay¶

Type: int

history¶

Type: Policy.History.KeepAll, Policy.History.KeepLast

max_samples¶

Type: int

max_instances¶

Type: int

max_samples_per_instance¶

Type: int

class IgnoreLocal(**kwargs)[source]¶

The IgnoreLocal Qos Policy

Examples

>>> Policy.IgnoreLocal.Nothing
>>> Policy.IgnoreLocal.Participant
>>> Policy.IgnoreLocal.Process

class Userdata(data)[source]¶

The Userdata Qos Policy

Examples

>>> Policy.Userdata(data=b"Hello, World!")

Parameters: data (bytes) –
Return type: None

class Topicdata(data)[source]¶

The Topicdata Qos Policy

Examples

>>> Policy.Topicdata(data=b"Hello, World!")

Parameters: data (bytes) –
Return type: None

class Groupdata(data)[source]¶

The Groupdata Qos Policy

Examples

>>> Policy.Groupdata(data=b"Hello, World!")

Parameters: data (bytes) –
Return type: None

class Property(key, value)[source]¶

The Property Qos Policy

Examples

>>> Policy.Property(key="host", value="central")

Parameters

key (str) –
value (str) –

Return type

None

class BinaryProperty(key, value)[source]¶

The BinaryProperty Qos Policy

Examples

>>> Policy.BinaryProperty(key="host", value=b"central")

Parameters

key (str) –
value (bytes) –

Return type

None

class TypeConsistency(**kwargs)[source]¶

The TypeConsistency Qos Policy

Examples

>>> Policy.TypeConsistency.DisallowTypeCoercion
>>> Policy.TypeConsistency.AllowTypeCoercion

DisallowTypeCoercion[source]¶: alias of cyclonedds.qos.Policy.TypeConsistency.DisallowTypeCoercion

AllowTypeCoercion[source]¶: alias of cyclonedds.qos.Policy.TypeConsistency.AllowTypeCoercion

class DataRepresentation(use_cdrv0_representation=False, use_xcdrv2_representation=False)[source]¶

The DataRepresentation Qos Policy

Parameters

use_cdrv0_representation (bool) –
use_xcdrv2_representation (bool) –

Return type

None

class EntityName(name)[source]¶

The EntityName Qos Policy

Parameters: name (str) –
Return type: None