Repository navigation
Add a Cyclone DDS transport - #46
mikelangmayr wants to merge 2 commits into
Conversation
3199136 to
0c2ba31
Compare
| self._rx_thread.start() | ||
| self._settle_discovery() | ||
|
|
||
| def stop(self) -> None: |
There was a problem hiding this comment.
Does this only stop the receive thread? Is the participant and both readers stay alive?
There was a problem hiding this comment.
Yes, it did. stop() now deletes the participant, which removes its readers and writers.
| def stop(self) -> None: | ||
| """Stop receiving and drop later sends.""" | ||
| self._stop.set() | ||
| self._wake.set(True) |
There was a problem hiding this comment.
does _wake need to be set again in start()? in the case of calling stop() then start() after
There was a problem hiding this comment.
stop() is now final; start() after it raises, so _wake is never reused.
| def _rx_loop(self) -> None: | ||
| while not self._stop.is_set(): | ||
| self._waitset.wait(duration(seconds=RX_WAKE_S)) | ||
| for reader in (self._direct_reader, self._broadcast_reader): |
There was a problem hiding this comment.
One exception in the callback kills the receiving thread, might be good to have a similar try/except like you've done in rabbitmq for handling this
There was a problem hiding this comment.
Callback errors are now caught and logged; receiving continues.
CycloneDdsTransport: brokerless, peers found by DDS discovery, no address bookpack_frames = False); direct messages are routed by the destination's DDS partitiontransport: cycloneddsin daemon, client and CLI config,Libby.cyclonedds(),Client.cyclonedds(), or--transport cycloneddswith--dds-domain/--dds-configlibby[cyclonedds], also installed in the tox test and docs envs