Subscriptions for PathObject and Instance

Author: VeskeR

textile/features.textile

@@ -2126,6 +2126,12 @@ h4. ReferenceExtras
 *** @(REX2b1)@ Should be written in reverse domain name notation
 *** @(REX2b2)@ Types beginning with @com.ably.@ are reserved
 
+h4. Subscription
+
+* @(SUB1)@ A @Subscription@ represents a registration for receiving events from a subscribe operation
+* @(SUB2)@ The @Subscription@ object has the following method:
+** @(SUB2a)@ @unsubscribe@ - deregisters the listener that was registered by the corresponding @subscribe@ call. Once @unsubscribe@ called, the listener must not be called for any subsequent events
+
 h3(#options). Option types
 
 h4. ClientOptions
@@ -3162,6 +3168,9 @@ class MessageVersion:
   clientId: string? // TM2s3
   description: string? // TM2s4
   metadata: Dict<string, string>? //TM2s5
+
+interface Subscription: // SUB*
+  unsubscribe() // SUB2a
 </pre>
 
 h2(#old-specs). Old specs

textile/objects-features.textile

@@ -290,6 +290,14 @@ h3(#realtime-objects). RealtimeObject
 * @(RTO22)@ @ObjectsOperationSource@ is an internal enum describing the source of an operation being applied:
 ** @(RTO22a)@ @LOCAL@ - an operation that originated locally, being applied upon receipt of the @ACK@ from Realtime
 ** @(RTO22b)@ @CHANNEL@ - an operation received over a Realtime channel
+* @(RTO24)@ Internal @PathObjectSubscriptionRegister@ - manages path-based subscriptions for @PathObject#subscribe@ ("RTPO19":#RTPO19)
+** @(RTO24a)@ The @RealtimeObject@ instance maintains a single @PathObjectSubscriptionRegister@ that manages all path-based subscriptions for the channel
+** @(RTO24b)@ When a @LiveObject@ in the @ObjectsPool@ emits a @LiveObjectUpdate@ (per "RTLO4b4":#RTLO4b4), the @PathObjectSubscriptionRegister@ must determine which subscriptions should be notified:
+*** @(RTO24b1)@ Determine the paths in the LiveObjects tree at which the updated @LiveObject@ is located
+*** @(RTO24b2)@ For each registered subscription, check whether the event path starts with (or equals) the subscription's path
+*** @(RTO24b3)@ If the event path matches, apply depth filtering: the event is dispatched to the subscription if the number of path segments from the subscription path to the event path plus 1 does not exceed the subscription's @depth@ option (or if @depth@ is undefined). Formally, the event is dispatched if @eventPath.length - subscriptionPath.length + 1 <= depth@
+*** @(RTO24b4)@ Create a @PathObjectSubscriptionEvent@ with a @PathObject@ pointing to the event path and the @ObjectMessage@ that caused the change, and call the subscription's listener
+*** @(RTO24b5)@ If a listener throws an error, the error must be caught and logged without affecting the dispatch to other subscriptions
 
 h3(#liveobject). LiveObject
 
@@ -317,9 +325,10 @@ h3(#liveobject). LiveObject
 **** @(RTLO4b4c)@ When a @LiveObjectUpdate@ is emitted:
 ***** @(RTLO4b4c1)@ If @LiveObjectUpdate@ is indicated to be a no-op, do nothing
 ***** @(RTLO4b4c2)@ Otherwise, the registered listener is called with the @LiveObjectUpdate@ object
-*** @(RTLO4b5)@ The client library may return a subscription object (or the idiomatic equivalent for the language) as a result of this operation:
-**** @(RTLO4b5a)@ The subscription object includes an @unsubscribe@ function
-**** @(RTLO4b5b)@ Calling @unsubscribe@ deregisters the listener previously registered by the user via the corresponding @subscribe@ call
+*** @(RTLO4b5)@ This clause has been replaced by "RTLO4b7":#RTLO4b7
+**** @(RTLO4b5a)@ This clause has been replaced by "RTLO4b7":#RTLO4b7
+**** @(RTLO4b5b)@ This clause has been replaced by "RTLO4b7":#RTLO4b7
+*** @(RTLO4b7)@ Returns a "@Subscription@":../features#SUB1 object
 *** @(RTLO4b6)@ This operation must not have any side effects on @RealtimeObject@, the underlying channel, or their status
 ** @(RTLO4c)@ public @unsubscribe@ - unsubscribes a previously registered listener
 *** @(RTLO4c1)@ This operation does not require any specific channel modes to be granted, nor does it require the channel to be in a specific state
@@ -857,6 +866,31 @@ A @PathObject@ is obtained from @RealtimeObject#get@ ("RTO23":#RTO23), which ret
 ** @(RTPO18b)@ Resolves the path using the path resolution procedure ("RTPO3":#RTPO3). On failure, throws per "RTPO3c2":#RTPO3c2
 ** @(RTPO18c)@ If the resolved value is a @LiveCounter@, delegates to @LiveCounter#decrement@ ("RTLC13":#RTLC13) with the provided @amount@
 ** @(RTPO18d)@ If the resolved value is not a @LiveCounter@, the library must throw an @ErrorInfo@ error with @statusCode@ 400 and @code@ 92007
+* @(RTPO19)@ @PathObject#subscribe@ function:
+** @(RTPO19a)@ Expects the following arguments:
+*** @(RTPO19a1)@ @listener@ - a callback function that receives a @PathObjectSubscriptionEvent@ ("RTPO19d":#RTPO19d) when a change occurs at or below this path
+*** @(RTPO19a2)@ @options@ @PathObjectSubscriptionOptions@ (optional) - subscription options
+** @(RTPO19b)@ @PathObjectSubscriptionOptions@ has the following properties:
+*** @(RTPO19b1)@ @depth@ @Number@ (optional) - controls how many levels deep in the subtree changes trigger the listener:
+**** @(RTPO19b1a)@ If undefined (default), the subscription receives events for changes at any depth below the subscribed path
+**** @(RTPO19b1b)@ If @depth@ is 1, only changes to the object at the exact subscribed path trigger the listener
+**** @(RTPO19b1c)@ If @depth@ is @n@, changes up to @n - 1@ levels of children below the subscribed path trigger the listener
+**** @(RTPO19b1d)@ If @depth@ is provided and is not a positive integer, the library must throw an @ErrorInfo@ error with @statusCode@ 400 and @code@ 40003
+** @(RTPO19c)@ Returns a "@Subscription@":../features#SUB1 object
+** @(RTPO19d)@ The listener receives a @PathObjectSubscriptionEvent@ object with:
+*** @(RTPO19d1)@ @object@ - a @PathObject@ pointing to the path where the change occurred
+*** @(RTPO19d2)@ @message@ @ObjectMessage@ (optional) - the @ObjectMessage@ that caused the change
+** @(RTPO19e)@ The subscription is path-based: it follows the path, not a specific object. If the object at the path changes identity (e.g. via a @MAP_SET@ operation replacing it), the subscription continues to deliver events for the new object at that path
+** @(RTPO19f)@ Events at child paths bubble up to the subscription, subject to depth filtering. For example, a subscription at path @a.b@ receives events for changes at @a.b@, @a.b.c@, @a.b.c.d@, etc., depending on the configured depth. The dispatch rules are described in "RTO24b":#RTO24b
+** @(RTPO19g)@ This operation must not have any side effects on @RealtimeObject@, the underlying channel, or their status
+* @(RTPO20)@ @PathObject#unsubscribe@ function:
+** @(RTPO20a)@ Accepts a @listener@ argument and deregisters it from receiving further events for this @PathObject@'s path
+** @(RTPO20b)@ This operation must not have any side effects on @RealtimeObject@, the underlying channel, or their status
+* @(RTPO21)@ The client library should provide a method that allows consuming subscription events as a stream or iterable, rather than via a callback. A suggested name for this method is @subscribeIterator@:
+** @(RTPO21a)@ Expects the following arguments:
+*** @(RTPO21a1)@ @options@ @PathObjectSubscriptionOptions@ (optional) - same options as @PathObject#subscribe@ ("RTPO19b":#RTPO19b)
+** @(RTPO21b)@ Returns a stream or iterable that yields @PathObjectSubscriptionEvent@ objects, using the idiomatic construct for the language (e.g. async iterators, channels, flows, or async sequences)
+** @(RTPO21c)@ Internally wraps @PathObject#subscribe@ ("RTPO19":#RTPO19), converting the callback-based subscription into the appropriate streaming or iterable pattern
 
 h3(#instance). Instance
 
@@ -913,6 +947,24 @@ An @Instance@ holds a direct reference to a specific resolved @LiveObject@ or pr
 *** @(RTINS15a1)@ @amount@ @Number@ (optional) - the amount by which to decrement the counter value. Defaults to 1
 ** @(RTINS15b)@ If the wrapped value is a @LiveCounter@, delegates to @LiveCounter#decrement@ ("RTLC13":#RTLC13) with the provided @amount@
 ** @(RTINS15c)@ If the wrapped value is not a @LiveCounter@, the library must throw an @ErrorInfo@ error with @statusCode@ 400 and @code@ 92007
+* @(RTINS16)@ @Instance#subscribe@ function:
+** @(RTINS16a)@ Expects the following arguments:
+*** @(RTINS16a1)@ @listener@ - a callback function that receives an @InstanceSubscriptionEvent@ ("RTINS16d":#RTINS16d) when the wrapped object is updated
+** @(RTINS16b)@ If the wrapped value is not a @LiveObject@ (i.e. it is a primitive), the library must throw an @ErrorInfo@ error with @statusCode@ 400 and @code@ 92007, indicating that subscribe is not supported for primitive values
+** @(RTINS16c)@ Subscribes to data updates on the underlying @LiveObject@ using @LiveObject#subscribe@ ("RTLO4b":#RTLO4b)
+** @(RTINS16d)@ The listener receives an @InstanceSubscriptionEvent@ object with:
+*** @(RTINS16d1)@ @object@ - the @Instance@ representing the updated object
+*** @(RTINS16d2)@ @message@ @ObjectMessage@ (optional) - the @ObjectMessage@ that caused the change
+** @(RTINS16e)@ Returns a "@Subscription@":../features#SUB1 object
+** @(RTINS16f)@ The subscription is identity-based: it follows the specific @LiveObject@ instance, regardless of where it sits in the tree
+** @(RTINS16g)@ This operation must not have any side effects on @RealtimeObject@, the underlying channel, or their status
+* @(RTINS17)@ @Instance#unsubscribe@ function:
+** @(RTINS17a)@ Accepts a @listener@ argument and deregisters it from receiving further events using @LiveObject#unsubscribe@ ("RTLO4c":#RTLO4c)
+** @(RTINS17b)@ This operation must not have any side effects on @RealtimeObject@, the underlying channel, or their status
+* @(RTINS18)@ The client library should provide a method that allows consuming subscription events as a stream or iterable, rather than via a callback. A suggested name for this method is @subscribeIterator@:
+** @(RTINS18a)@ If the wrapped value is not a @LiveObject@, the library must throw an @ErrorInfo@ error with @statusCode@ 400 and @code@ 92007
+** @(RTINS18b)@ Returns a stream or iterable that yields @InstanceSubscriptionEvent@ objects, using the idiomatic construct for the language (e.g. async iterators, channels, flows, or async sequences)
+** @(RTINS18c)@ Internally wraps @Instance#subscribe@ ("RTINS16":#RTINS16), converting the callback-based subscription into the appropriate streaming or iterable pattern
 
 h2(#idl). Interface Definition
 
@@ -951,13 +1003,10 @@ class LiveObject: // RTLO*, internal
   tombstonedAt: Time? // RTLO3e
   canApplyOperation(ObjectMessage) -> Boolean // RTLO4a
   tombstone(ObjectMessage) // RTLO4e
-  subscribe((LiveObjectUpdate) ->) -> LiveObjectSubscription // RTLO4b
+  subscribe((LiveObjectUpdate) ->) -> Subscription // RTLO4b
   unsubscribe((LiveObjectUpdate) ->) // RTLO4c
 
-interface LiveObjectSubscription: // RTLO4b5
-  unsubscribe() // RTLO4b5a
-
-interface LiveObjectUpdate: // RTLO4b4, internal
+interface LiveObjectUpdate: // RTLO4b4
   update: Object // RTLO4b4a
   noop: Boolean // RTLO4b4b
 
@@ -989,6 +1038,17 @@ class LiveCounterValueType: // RTLCV*
 class LiveMapValueType: // RTLMV*
   // created via LiveMap.create(), RTLMV3
 
+interface PathObjectSubscriptionEvent: // RTPO19d
+  object: PathObject // RTPO19d1
+  message: ObjectMessage? // RTPO19d2
+
+interface PathObjectSubscriptionOptions: // RTPO19b
+  depth: Number? // RTPO19b1
+
+interface InstanceSubscriptionEvent: // RTINS16d
+  object: Instance // RTINS16d1
+  message: ObjectMessage? // RTINS16d2
+
 class PathObject: // RTPO*
   path() -> String // RTPO4
   get(String key) -> PathObject // RTPO5
@@ -1005,6 +1065,9 @@ class PathObject: // RTPO*
   remove(String key) => io // RTPO16
   increment(Number amount?) => io // RTPO17
   decrement(Number amount?) => io // RTPO18
+  subscribe((PathObjectSubscriptionEvent) -> listener, PathObjectSubscriptionOptions? options) -> Subscription // RTPO19
+  unsubscribe((PathObjectSubscriptionEvent) -> listener) // RTPO20
+  subscribeIterator(PathObjectSubscriptionOptions? options) -> Stream<PathObjectSubscriptionEvent> // RTPO21
 
 class Instance: // RTINS*
   id: String? // RTINS3
@@ -1020,4 +1083,7 @@ class Instance: // RTINS*
   remove(String key) => io // RTINS13
   increment(Number amount?) => io // RTINS14
   decrement(Number amount?) => io // RTINS15
+  subscribe((InstanceSubscriptionEvent) -> listener) -> Subscription // RTINS16
+  unsubscribe((InstanceSubscriptionEvent) -> listener) // RTINS17
+  subscribeIterator() -> Stream<InstanceSubscriptionEvent> // RTINS18
 </pre>