Add the subscribe mechanism to the capnp session #9
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
tobiasah
force-pushed
the
tobiasa/subscribe
branch
5 times, most recently
from
cde0a4a to
b3cd9f1
Compare
markheik
approved these changes
Aug 31, 2023
src/labone/core/subscription.py
Outdated
|
|
||
| Subscriptions are implemented through the capnp stream mechanism. This handles | ||
| all the communication stuff, e.g. back pressure and flow control. The only thing | ||
| the client needs to provide is a session_protocol_capnp.StreamingHandle.Server |
There was a problem hiding this comment.
nitpickly: session_protocol_capnp.StreamingHandle.Server
src/labone/core/session.py
Outdated
| ) | ||
| try: | ||
| subscription.path = path | ||
| except (AttributeError, capnp.KjException) as error: |
There was a problem hiding this comment.
I think capnp also raises TypeError, when type of list is given
There was a problem hiding this comment.
which would be fine to show to the user right? I just wanted to avoid KjExceptions being leaked. And the Attribiute Error a weird thing from capnp because it tires to convert the dict into a capnp object
There was a problem hiding this comment.
The error message is not really helpful from capnp:
r = ses._session.listNodes_request()
r.pathExpression = [123] # Should be string
>>> TypeError: Cannot convert str to capnp.lib.capnp._DynamicListBuilder
This is probably super error rare edge case.
There was a problem hiding this comment.
*unless capnp struct assignment throws a different error.
There was a problem hiding this comment.
All right just tested it:
await ses.subscribe(["foobar"])
>>>
TypeError Traceback (most recent call last)
Cell In[5], line 1
----> 1 await ses.subscribe(["foobar"])
File ~/Documents/zhinst/zpy/labone/src/labone/core/session.py:466, in Session.subscribe(self, path)
461 subscription = session_protocol_capnp.Subscription(
462 streamingHandle=streaming_handle,
463 subscriberId=self._client_id.bytes,
464 )
465 try:
--> 466 subscription.path = path
467 except (AttributeError, capnp.KjException) as error:
468 field_type = _request_field_type_description(subscription, "path")
File ~/.pyenv/versions/3.9.13/envs/pylabone/lib/python3.9/site-packages/capnp/lib/capnp.pyx:1368, in capnp.lib.capnp._DynamicStructBuilder.__setattr__()
File ~/.pyenv/versions/3.9.13/envs/pylabone/lib/python3.9/site-packages/capnp/lib/capnp.pyx:1361, in capnp.lib.capnp._DynamicStructBuilder._set()
File ~/.pyenv/versions/3.9.13/envs/pylabone/lib/python3.9/site-packages/capnp/lib/capnp.pyx:773, in capnp.lib.capnp._setDynamicField()
TypeError: Cannot convert str to capnp.lib.capnp._DynamicListBuilder
a7cd733 to
829f0ed
Compare
The subscribe mechanism is based on the capnp stream model. This means that the client creates for each subscription a server that is passed to the kernel. The kernel than calls a specific function on that value for each value change. In this api the received value updates are stored in a async queue. To disconnect a connection the created server would need to be closed. In python this is not so easy due to the hidden reference counting. Therefore a subscription needs to be canceled manually through a disconnect call in the async queue.
tobiasah
force-pushed
the
tobiasa/subscribe
branch
from
829f0ed to
49d3f63
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
The subscribe mechanism is based on the capnp stream model. This means that the client creates for each subscription a server that is passed to the kernel. The kernel than calls a specific function on that value for each value change.
In this api the received value updates are stored in a async queue. To disconnect a connection the created server would need to be closed. In python this is not so easy due to the hidden reference counting. Therefore a subscription needs to be canceled manually through a disconnect call in the async queue.