[ECO-5426][LiveObjects] Implement object sync / operation #1137
Merged
[ECO-5426][LiveObjects] Implement object sync / operation #1137
Conversation
…erfaces - Enhanced LiveCounter interface with increment/decrement operations and comprehensive JavaDoc - Extended LiveObjectsPlugin interface for modular plugin architecture - Updated ErrorCodes enum for standardized live objects error handling - Established blocking/non-blocking operation annotations for API consistency
- Created ObjectId class with comprehensive validation and parsing logic - Enhanced Utils helper functions for common object operations - Updated Helpers class for shared live objects functionality - Established consistent logging patterns and error handling across components
- Enhanced DefaultLiveObjectsPlugin with seamless channel integration - Updated DefaultLiveObjects with coroutine-based sequential message processing - Added objects pool initialization and comprehensive lifecycle management - Implemented channel state change handling for live objects synchronization
…sting - Added BaseLiveObject abstract class as foundation for LiveMap/LiveCounter - Implemented site timeserials tracking and tombstone lifecycle management - Created BaseLiveObjectTest with spec-compliant validation and edge cases - Established thread-safe object state management and operation handling
…system - Enhanced ObjectMessage data class with complete serialization field mapping - Created comprehensive ObjectId validation with extensive test coverage - Added message size calculation and validation logic for protocol compliance - Established edge case testing for object ID formats and parsing scenarios
- Enhanced TestUtils with advanced mock object generation capabilities - Updated TestHelpers with channel and adapter mocking for integration tests - Added parameterized test base classes for systematic integration testing - Established consistent test patterns and mock factories for all live object types
…e management - Added ObjectsPool with ConcurrentHashMap for thread-safe object storage - Implemented automatic garbage collection for tombstoned objects - Created pool initialization with root object and comprehensive lifecycle management - Added extensive ObjectsPoolTest with concurrent access validation and edge cases
…ocessing - Implemented ObjectsManager for OBJECT and OBJECT_SYNC message processing - Added sync objects data pool for collecting and managing sync sequences - Created buffered object operations during sync state with proper ordering - Established comprehensive ObjectsManagerTest with sync sequence validation scenarios
…tions - Added ObjectsSyncTracker for managing sync sequences and cursor tracking - Implemented sync state transitions and channel serial management - Created comprehensive sync tracking tests with edge case coverage - Established sync completion detection and buffered operation processing logic
- Created LiveMapEntry with timestamp and value management for conflict resolution - Enhanced ObjectMessageFixtures for consistent test data generation across test suites - Added entry lifecycle management with tombstone support and GC integration - Established fixture patterns for various object message scenarios and edge cases
- Added DefaultLiveMap with ConcurrentHashMap for thread-safe map operations - Implemented map semantics support (LWW) and comprehensive entry management - Created essential LiveMap operations (get, set, remove, size) with validation - Established DefaultLiveMapTest with fundamental operation validation and concurrency tests
…ocessing - Implemented LiveMapManager for handling complex map operations and state management - Added operation validation and conflict resolution logic with LWW semantics - Created comprehensive LiveMapManagerTest with operation scenarios and edge cases - Established map update calculation and change notification system for subscribers
- Added DefaultLiveCounter with AtomicReference for thread-safe counting operations - Implemented counter operations (increment, decrement, value) with proper synchronization - Created DefaultLiveCounterTest with concurrent operation validation and stress testing - Established counter data management and operation result handling with notifications
- Implemented LiveCounterManager for counter-specific operations and state management - Added counter operation validation and comprehensive state management logic - Created extensive LiveCounterManagerTest with edge cases and error scenarios - Established counter update calculation and notification system for real-time updates
…ting - Created comprehensive DefaultLiveObjectsTest with integration scenarios - Added complete live objects lifecycle and state management validation - Implemented channel state integration testing with various state transitions - Established end-to-end workflow testing for live objects synchronization
- Enhanced JsonSerialization with custom type adapters for enum serialization - Improved ObjectMessage JSON serialization and deserialization with proper formatting - Updated enum code type adapters for ObjectOperationAction and MapSemantics - Established JSON object conversion utilities with comprehensive error handling
- Enhanced MsgpackSerialization for efficient binary message encoding - Improved ObjectMessage MessagePack write and read operations with field optimization - Updated DefaultLiveObjectSerializer supporting both JSON and MessagePack formats - Established efficient binary serialization with minimal memory allocation
- Enhanced ObjectMessageSerializationTest with comprehensive format validation - Updated ObjectMessageSizeTest with size calculation validation and limits enforcement - Improved message size calculation methods for all object message components - Established size validation utilities with comprehensive test coverage and edge cases
- Enhanced ChannelBase to support live objects functionality with seamless integration - Updated live objects protocol message handling in channel pipeline - Improved channel state integration for live objects synchronization - Established robust integration between channels and live objects with proper lifecycle
WalkthroughThis pull request introduces a comprehensive implementation of live object synchronization and real-time operation handling for Ably's live objects system. It adds core classes for object pooling, synchronization tracking, object management, and type-specific logic for live maps and counters. Extensive changes include new interfaces, state machines, serialization/deserialization updates, and a large suite of unit tests covering the new functionality. Changes
Sequence Diagram(s)sequenceDiagram
participant ChannelBase
participant LiveObjectsPlugin
participant DefaultLiveObjects
participant ObjectsManager
participant ObjectsPool
ChannelBase->>LiveObjectsPlugin: handleStateChange(channelName, state, hasObjects)
LiveObjectsPlugin->>DefaultLiveObjects: handleStateChange(state, hasObjects)
DefaultLiveObjects->>ObjectsManager: startNewSync / endSync (based on state)
DefaultLiveObjects->>ObjectsManager: handle(protocolMessage)
ObjectsManager->>ObjectsPool: applyObjectMessages / applyObjectSyncMessages
ObjectsPool->>ObjectsManager: get/set objects as needed
Estimated code review effort🎯 5 (Critical) | ⏱️ ~90+ minutes Assessment against linked issues
Assessment against linked issues: Out-of-scope changes
Possibly related PRs
Suggested reviewers
Poem
Note ⚡️ Unit Test Generation is now available in beta!Learn more here, or try it out under "Finishing Touches" below. ✨ Finishing Touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
sacOO7
force-pushed
the
feature/object-sync-refactored
branch
from
b0c8984 to
c1d1d90
Compare
sacOO7
changed the title
sacOO7
force-pushed
the
feature/object-sync-refactored
branch
from
c1d1d90 to
0476f23
Compare
sacOO7
force-pushed
the
feature/object-sync-refactored
branch
from
0476f23 to
34449c7
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🔭 Outside diff range comments (1)
lib/src/main/java/io/ably/lib/objects/LiveCounter.java (1)
57-57: Update outdated Javadoc comment.The comment still references "as a Long" but the return type has been changed to
Double.- * @return the current value of the counter as a Long. + * @return the current value of the counter as a Double.
🧹 Nitpick comments (13)
live-objects/src/main/kotlin/io/ably/lib/objects/DefaultLiveObjectsPlugin.kt (1)
25-25: Fix typo in dispose reason message."ben" should be "been" in the dispose reason string.
- liveObjects[channelName]?.dispose("Channel has ben released using channels.release()") + liveObjects[channelName]?.dispose("Channel has been released using channels.release()")live-objects/src/main/kotlin/io/ably/lib/objects/type/livecounter/DefaultLiveCounter.kt (1)
37-55: Reminder: Implement core counter operations.Several core methods are marked as TODO. These are essential for the LiveCounter functionality.
Would you like me to help implement these counter operations or create an issue to track this work?
live-objects/src/main/kotlin/io/ably/lib/objects/type/livemap/DefaultLiveMap.kt (2)
4-4: Remove redundant import.
MapSemanticsis already imported via the wildcard import on line 3.-import io.ably.lib.objects.MapSemantics
40-74: Reminder: Implement core map operations.Several essential LiveMap methods are marked as TODO and need implementation.
Would you like me to help implement these map operations or create an issue to track this work?
live-objects/src/main/kotlin/io/ably/lib/objects/ObjectMessage.kt (1)
445-447: Consider simplifying the null-safety logic.The function correctly identifies invalid ObjectData, but the logic could be more readable:
internal fun ObjectData?.isInvalid(): Boolean { - return this?.objectId.isNullOrEmpty() && this?.value == null + return this == null || (objectId.isNullOrEmpty() && value == null) }This makes it explicit that null ObjectData is also invalid, and uses smart casting for cleaner property access.
live-objects/src/main/kotlin/io/ably/lib/objects/serialization/MsgpackSerialization.kt (2)
229-231: Redundant Unknown enum lookup.The second search for
code == -1is unnecessary sinceUnknownis already defined with code-1. Simplify to:- action = ObjectOperationAction.entries.firstOrNull { it.code == actionCode } - ?: ObjectOperationAction.entries.firstOrNull { it.code == -1 } - ?: throw objectError("Unknown ObjectOperationAction code: $actionCode and no Unknown fallback found") + action = ObjectOperationAction.entries.firstOrNull { it.code == actionCode } + ?: ObjectOperationAction.Unknown
489-491: Redundant Unknown enum lookup.Similar to the ObjectOperationAction case, simplify the MapSemantics lookup:
- semantics = MapSemantics.entries.firstOrNull { it.code == semanticsCode } - ?: MapSemantics.entries.firstOrNull { it.code == -1 } - ?: throw objectError("Unknown MapSemantics code: $semanticsCode and no UNKNOWN fallback found") + semantics = MapSemantics.entries.firstOrNull { it.code == semanticsCode } + ?: MapSemantics.Unknownlive-objects/src/main/kotlin/io/ably/lib/objects/type/BaseLiveObject.kt (1)
66-91: Consider improving null-safety handling.While the non-null assertions on line 84 are safe (since
canApplyOperationthrows for null values), consider returning the validated values fromcanApplyOperationto avoid assertions:// Return a data class or Pair from canApplyOperation internal fun validateAndCheckOperation(siteCode: String?, timeSerial: String?): Pair<String, String>? { // validation logic... return if (canApply) Pair(siteCode, timeSerial) else null }This would make the null-safety guarantees explicit in the type system.
live-objects/src/test/kotlin/io/ably/lib/objects/unit/type/livecounter/LiveCounterManagerTest.kt (2)
10-10: Test class naming inconsistency.The class is named
DefaultLiveCounterManagerTestbut it testsLiveCounterManager. Consider renaming to match the class under test:-class DefaultLiveCounterManagerTest { +class LiveCounterManagerTest {
217-255: Consider removing duplicate increment tests.Tests at lines 217-235 and 237-255 duplicate the increment functionality already tested at lines 175-192. Consider either removing duplicates or differentiating them with unique scenarios.
live-objects/src/main/kotlin/io/ably/lib/objects/ObjectsPool.kt (1)
50-56: Uselateinitfor deferred initialization of gcJobThe
gcJobvariable is declared without initialization but assigned in the init block. Consider usinglateinitto make the deferred initialization pattern explicit.- private var gcJob: Job // Job for the garbage collection coroutine + private lateinit var gcJob: Job // Job for the garbage collection coroutinelive-objects/src/main/kotlin/io/ably/lib/objects/DefaultLiveObjects.kt (1)
41-47: Consider documenting the buffer overflow strategyThe
MutableSharedFlowusesUNLIMITEDbuffer capacity which could lead to memory issues if messages accumulate faster than processing. While this ensures no messages are dropped, consider documenting the expected message throughput and any backpressure strategies.live-objects/src/main/kotlin/io/ably/lib/objects/type/livemap/LiveMapManager.kt (1)
115-117: Validate data before serial comparisonThe data validation happens after the serial comparison check. Consider moving the validation before the serial check to fail fast on invalid data, avoiding unnecessary serial comparisons.
): Map<String, String> { val existingEntry = liveMap.data[mapOp.key] + if (mapOp.data.isInvalid()) { + throw objectError("Invalid object data for MAP_SET op on objectId=${objectId} on key=${mapOp.key}") + } + // RTLM7a if (existingEntry != null && !canApplyMapOperation(existingEntry.timeserial, timeSerial)) { // RTLM7a1 - the operation's serial <= the entry's serial, ignore the operation Log.v(tag, "Skipping update for key=\"${mapOp.key}\": op serial $timeSerial <= entry serial ${existingEntry.timeserial};" + " objectId=${objectId}" ) return mapOf() } - if (mapOp.data.isInvalid()) { - throw objectError("Invalid object data for MAP_SET op on objectId=${objectId} on key=${mapOp.key}") - }
live-objects/src/main/kotlin/io/ably/lib/objects/type/livecounter/DefaultLiveCounter.kt
Show resolved
Hide resolved
live-objects/src/test/kotlin/io/ably/lib/objects/unit/type/BaseLiveObjectTest.kt
Show resolved
Hide resolved
live-objects/src/test/kotlin/io/ably/lib/objects/unit/type/BaseLiveObjectTest.kt
Show resolved
Hide resolved
live-objects/src/test/kotlin/io/ably/lib/objects/unit/type/BaseLiveObjectTest.kt
Show resolved
Hide resolved
live-objects/src/test/kotlin/io/ably/lib/objects/unit/type/livemap/LiveMapManagerTest.kt
Show resolved
Hide resolved
ttypic
approved these changes
Aug 8, 2025
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
Summary by CodeRabbit
New Features
Bug Fixes
Tests
Documentation
Style