Repository: ably/ably-go
Pull request #682 (2025-06-04 15:41:50)

Add Experimental Live Objects Plugin
Add experimental Live Objects implementation. This PR adds the following:

- Define a live objects plugin interface.
- Add the live objects experimental plugin option.
- Add Objects message types for constructing Object operation requests.
- Add an ExperimentalObjects component to the RealtimeChannel to allow for Object message publishing.

<!-- This is an auto-generated comment: release notes by coderabbit.ai -->
## Summary by CodeRabbit

- **New Features**
  - Introduced experimental LiveObjects support for real-time object operations within channels.
  - Added new channel modes for object publishing and subscribing.
  - Enabled publishing of LiveObject messages with plugin-based preparation and acknowledgment handling.
  - Added client option to configure an experimental objects plugin.

- **Enhancements**
  - Extended protocol message structure to support object state messages.
  - Updated channel and protocol handling to recognize and process object-related actions.
  - Introduced core data structures and interfaces for LiveObjects representation and manipulation.

- **Bug Fixes**
  - Added new error code for missing plugin to improve error handling.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

coderabbitai[bot]: <!-- This is an auto-generated comment: summarize by coderabbit.ai -->
<!-- This is an auto-generated comment: review in progress by coderabbit.ai -->

> [!NOTE]
> Currently processing new changes in this PR. This may take a few minutes, please wait...
> 
> <details>
> <summary>📥 Commits</summary>
> 
> Reviewing files that changed from the base of the PR and between 4596ce4cda88298e84684d0bf6ac5b3845bfdab6 and 521a0feee1ae69a1450ebe3f40b5fb5c78d4409e.
> 
> </details>
> 
> <details>
> <summary>📒 Files selected for processing (11)</summary>
> 
> * `ably/objects/liveobjects.go` (1 hunks)
> * `ably/objects/liveobjects_test.go` (1 hunks)
> * `ably/objects/objects.go` (1 hunks)
> * `ably/options.go` (3 hunks)
> * `ably/proto_action.go` (2 hunks)
> * `ably/proto_channel_options.go` (3 hunks)
> * `ably/proto_protocol_message.go` (3 hunks)
> * `ably/realtime_channel.go` (6 hunks)
> * `ably/realtime_experimental_objects.go` (1 hunks)
> * `examples/liveobjects/objects_test.go` (1 hunks)
> * `examples/liveobjects/util.go` (1 hunks)
> 
> </details>
> 
> ```ascii
>  _________________________________________
> < Never ask an AI its Energy Star rating. >
>  -----------------------------------------
>   \
>    \   (\__/)
>        (•ㅅ•)
>        / 　 づ
> ```

<!-- end of auto-generated comment: review in progress by coderabbit.ai -->
<!-- usage_tips_start -->

> [!TIP]
> <details>
> <summary>CodeRabbit can use your project's `golangci-lint` configuration to improve the quality of Go code reviews.</summary>
> 
> Add a [configuration file](https://golangci-lint.run/usage/configuration/) to your project to customize how CodeRabbit runs `golangci-lint`.
> 
> </details>

<!-- usage_tips_end -->
<!-- finishing_touch_checkbox_start -->

<details open="true">
<summary>✨ Finishing Touches</summary>

- [ ] <!-- {"checkboxId": "7962f53c-55bc-4827-bfbf-6a18da830691"} --> 📝 Generate Docstrings

</details>

<!-- finishing_touch_checkbox_end -->
<!-- tips_start -->

---

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.

<details>
<summary>❤️ Share</summary>

- [X](https://twitter.com/intent/tweet?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A&url=https%3A//coderabbit.ai)
- [Mastodon](https://mastodon.social/share?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A%20https%3A%2F%2Fcoderabbit.ai)
- [Reddit](https://www.reddit.com/submit?title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&text=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code.%20Check%20it%20out%3A%20https%3A//coderabbit.ai)
- [LinkedIn](https://www.linkedin.com/sharing/share-offsite/?url=https%3A%2F%2Fcoderabbit.ai&mini=true&title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&summary=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code)

</details>

<details>
<summary>🪧 Tips</summary>

### Chat

There are 3 ways to chat with [CodeRabbit](https://coderabbit.ai?utm_source=oss&utm_medium=github&utm_campaign=ably/ably-go&utm_content=682):

- Review comments: Directly reply to a review comment made by CodeRabbit. Example:
  - `I pushed a fix in commit <commit_id>, please review it.`
  - `Explain this complex logic.`
  - `Open a follow-up GitHub issue for this discussion.`
- Files and specific lines of code (under the "Files changed" tab): Tag `@coderabbitai` in a new review comment at the desired location with your query. Examples:
  - `@coderabbitai explain this code block.`
  -	`@coderabbitai modularize this function.`
- PR comments: Tag `@coderabbitai` in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  - `@coderabbitai gather interesting stats about this repository and render them as a table. Additionally, render a pie chart showing the language distribution in the codebase.`
  - `@coderabbitai read src/utils.ts and explain its main purpose.`
  - `@coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.`
  - `@coderabbitai help me debug CodeRabbit configuration file.`

### Support

Need help? Create a ticket on our [support page](https://www.coderabbit.ai/contact-us/support) for assistance with any issues or questions.

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

### CodeRabbit Commands (Invoked using PR comments)

- `@coderabbitai pause` to pause the reviews on a PR.
- `@coderabbitai resume` to resume the paused reviews.
- `@coderabbitai review` to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
- `@coderabbitai full review` to do a full review from scratch and review all the files again.
- `@coderabbitai summary` to regenerate the summary of the PR.
- `@coderabbitai generate docstrings` to [generate docstrings](https://docs.coderabbit.ai/finishing-touches/docstrings) for this PR.
- `@coderabbitai generate sequence diagram` to generate a sequence diagram of the changes in this PR.
- `@coderabbitai resolve` resolve all the CodeRabbit review comments.
- `@coderabbitai configuration` to show the current CodeRabbit configuration for the repository.
- `@coderabbitai help` to get help.

### Other keywords and placeholders

- Add `@coderabbitai ignore` anywhere in the PR description to prevent this PR from being reviewed.
- Add `@coderabbitai summary` to generate the high-level summary at a specific location in the PR description.
- Add `@coderabbitai` anywhere in the PR title to generate the title automatically.

### CodeRabbit Configuration File (`.coderabbit.yaml`)

- You can programmatically configure CodeRabbit by adding a `.coderabbit.yaml` file to the root of your repository.
- Please see the [configuration documentation](https://docs.coderabbit.ai/guides/configure-coderabbit) for more information.
- If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: `# yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json`

### Documentation and Community

- Visit our [Documentation](https://docs.coderabbit.ai) for detailed information on how to use CodeRabbit.
- Join our [Discord Community](http://discord.gg/coderabbit) to get help, request features, and share feedback.
- Follow us on [X/Twitter](https://twitter.com/coderabbitai) for updates and announcements.

</details>

<!-- tips_end -->
kaschula: @lmars I have fixed the suggestions you made. I also realised should probably add a unit test for the `RealtimeExperimentalObjects` type which I played around with claude to write. I have added it in a seperate commit in case you think its not worth adding. I will squash it into the main commit if we are keeping it.

Patch:
diff --git a/.gitignore b/.gitignore
index 1e102f1..8b0ef0c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -12,3 +12,4 @@
 vendor/
 .idea/
 *.env
+.claude
\ No newline at end of file
diff --git a/ably/error_names.go b/ably/error_names.go
index 708477e..e7607e2 100644
--- a/ably/error_names.go
+++ b/ably/error_names.go
@@ -9,6 +9,7 @@ const (
 	ErrBadRequest                                ErrorCode = 40000
 	ErrInvalidCredential                         ErrorCode = 40005
 	ErrInvalidClientID                           ErrorCode = 40012
+	ErrMissingPlugin                             ErrorCode = 40019
 	ErrUnauthorized                              ErrorCode = 40100
 	ErrInvalidCredentials                        ErrorCode = 40101
 	ErrIncompatibleCredentials                   ErrorCode = 40102
diff --git a/ably/objects/liveobjects.go b/ably/objects/liveobjects.go
new file mode 100644
index 0000000..afe44c5
--- /dev/null
+++ b/ably/objects/liveobjects.go
@@ -0,0 +1,134 @@
+package objects
+
+// Plugin is used to add LiveObject functionality to a realtime client:
+//
+//	client, err := ably.NewRealtime(
+//		ably.WithKey(...),
+//		ably.WithExperimentalObjectsPlugin(myObjectsPlugin),
+//	)
+//
+// The plugin is used to prepare object messages to be published in the
+// realtime channel's publishObjects method, and is used to handle incoming
+// object and object sync messages.
+type Plugin interface {
+	// PrepareObject prepares an object to be published (e.g. setting the
+	// objectId and intialValue for a COUNTER_CREATE op).
+	PrepareObject(msg *Message) error
+
+	// HandleObjectMessages is called to handle incoming object messages.
+	HandleObjectMessages(msgs []*Message)
+
+	// HandleObjectSyncMessages is called to handle incoming object sync
+	// messages.
+	HandleObjectSyncMessages(msgs []*Message, channelSerial string)
+}
+
+type Message struct {
+	ID           string                 `json:"id,omitempty" codec:"id,omitempty"`
+	Timestamp    int64                  `json:"timestamp,omitempty" codec:"timestamp,omitempty"`
+	ClientID     string                 `json:"clientId,omitempty" codec:"clientId,omitempty"`
+	ConnectionID string                 `json:"connectionId,omitempty" codec:"connectionId,omitempty"`
+	Operation    *Operation             `json:"operation,omitempty" codec:"operation,omitempty"`
+	Object       *State                 `json:"object,omitempty" codec:"object,omitempty"`
+	Extras       map[string]interface{} `json:"extras,omitempty" codec:"extras,omitempty"`
+	Serial       string                 `json:"serial,omitempty" codec:"serial,omitempty"`
+	SiteCode     string                 `json:"siteCode,omitempty" codec:"siteCode,omitempty"`
+}
+
+type State struct {
+	ObjectId        string            `json:"objectId,omitempty" codec:"objectId,omitempty"`
+	SiteTimeserials map[string]string `json:"siteTimeserials,omitempty" codec:"siteTimeserials,omitempty"`
+	Tombstone       bool              `json:"tombstone,omitempty" codec:"tombstone,omitempty"`
+	CreateOp        *Operation        `json:"createOp,omitempty" codec:"createOp,omitempty"`
+	Map             *Map              `json:"map,omitempty" codec:"map,omitempty"`
+	Counter         *Counter          `json:"counter,omitempty" codec:"counter,omitempty"`
+}
+
+type OperationAction int
+
+const (
+	Operation_MapCreate     OperationAction = 0
+	Operation_MapSet        OperationAction = 1
+	Operation_MapRemove     OperationAction = 2
+	Operation_CounterCreate OperationAction = 3
+	Operation_CounterInc    OperationAction = 4
+	Operation_Delete        OperationAction = 5
+)
+
+func (o OperationAction) String() string {
+	switch o {
+	case Operation_MapCreate:
+		return "MAP_CREATE"
+	case Operation_MapSet:
+		return "MAP_SET"
+	case Operation_MapRemove:
+		return "MAP_REMOVE"
+	case Operation_CounterCreate:
+		return "COUNTER_CREATE"
+	case Operation_CounterInc:
+		return "COUNTER_INC"
+	case Operation_Delete:
+		return "OBJECT_DELETE"
+	default:
+		return ""
+	}
+}
+
+type Operation struct {
+	Action       OperationAction `json:"action,omitempty" codec:"action,omitempty"`
+	ObjectID     string          `json:"objectId,omitempty" codec:"objectId,omitempty"`
+	MapOp        *MapOp          `json:"mapOp,omitempty" codec:"mapOp,omitempty"`
+	CounterOp    *CounterOp      `json:"counterOp,omitempty" codec:"counterOp,omitempty"`
+	Map          *Map            `json:"map,omitempty" codec:"map,omitempty"`
+	Counter      *Counter        `json:"counter,omitempty" codec:"counter,omitempty"`
+	Nonce        *string         `json:"nonce,omitempty" codec:"nonce,omitempty"`
+	InitialValue string          `json:"initialValue,omitempty" codec:"initialValue,omitempty"`
+}
+
+type CounterOp struct {
+	Amount float64 `json:"amount,omitempty" codec:"amount,omitempty"`
+}
+
+type Counter struct {
+	Count float64 `json:"count,omitempty" codec:"count,omitempty"`
+}
+
+type MapOp struct {
+	Key  string `json:"key,omitempty" codec:"key,omitempty"`
+	Data *Data  `json:"data,omitempty" codec:"data,omitempty"`
+}
+
+type Data struct {
+	ObjectId *string  `json:"objectId,omitempty" codec:"objectId,omitempty"`
+	Encoding *string  `json:"encoding,omitempty" codec:"encoding,omitempty"`
+	Boolean  *bool    `json:"boolean,omitempty" codec:"boolean,omitempty"`
+	Bytes    []byte   `json:"bytes,omitempty" codec:"bytes,omitempty"`
+	Number   *float64 `json:"number,omitempty" codec:"number,omitempty"`
+	String   *string  `json:"string,omitempty" codec:"string,omitempty"`
+}
+
+type Map struct {
+	Semantics MapSemantics         `json:"semantics,omitempty" codec:"semantics,omitempty"`
+	Entries   map[string]*MapEntry `json:"entries,omitempty" codec:"entries,omitempty"`
+}
+
+type MapSemantics int
+
+const (
+	Map_LWW MapSemantics = 0
+)
+
+func (m MapSemantics) String() string {
+	switch m {
+	case Map_LWW:
+		return "LWW"
+	default:
+		return ""
+	}
+}
+
+type MapEntry struct {
+	Tombstone  bool    `json:"tombstone,omitempty" codec:"tombstone,omitempty"`
+	Timeserial *string `json:"timeserial,omitempty" codec:"timeserial,omitempty"`
+	Data       *Data   `json:"data,omitempty" codec:"data,omitempty"`
+}
diff --git a/ably/options.go b/ably/options.go
index 8bfcae1..707a06a 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/ably/ably-go/ably/internal/ablyutil"
+	"github.com/ably/ably-go/ably/objects"
 )
 
 const (
@@ -426,6 +427,15 @@ type clientOptions struct {
 	// InsecureAllowBasicAuthWithoutTLS permits an API key to be used even if the connection
 	// will not use TLS, something which would otherwise not be permitted for security reasons.
 	InsecureAllowBasicAuthWithoutTLS bool
+
+	// ExperimentalObjectsPlugin is the plugin to use to implement [LiveObjects] functionality.
+	//
+	// NOTE: this option is experimental, the LiveObjects plugin API may change in a
+	// backwards incompatible way between minor/patch versions. Once the API has been finalised,
+	// a new non-experimental option will be added, and this one will be removed.
+	//
+	// [LiveObjects]: https://ably.com/docs/liveobjects
+	ExperimentalObjectsPlugin objects.Plugin
 }
 
 func (opts *clientOptions) validate() error {
@@ -1404,6 +1414,20 @@ func WithInsecureAllowBasicAuthWithoutTLS() ClientOption {
 	}
 }
 
+// WithExperimentalObjectsPlugin configures the client to use the given plugin to implement
+// [LiveObjects] functionality.
+//
+// NOTE: this option is experimental, the LiveObjects plugin API may change in a
+// backwards incompatible way between minor/patch versions. Once the API has been finalised,
+// a new non-experimental option will be added, and this one will be removed.
+//
+// [LiveObjects]: https://ably.com/docs/liveobjects
+func WithExperimentalObjectsPlugin(plugin objects.Plugin) ClientOption {
+	return func(opts *clientOptions) {
+		opts.ExperimentalObjectsPlugin = plugin
+	}
+}
+
 func applyOptionsWithDefaults(opts ...ClientOption) *clientOptions {
 	to := defaultOptions
 	// No need to set hosts by default
diff --git a/ably/proto_action.go b/ably/proto_action.go
index c1a37a1..a80a66e 100644
--- a/ably/proto_action.go
+++ b/ably/proto_action.go
@@ -21,6 +21,9 @@ const (
 	actionMessage
 	actionSync
 	actionAuth
+	actionActivate
+	actionObject
+	actionObjectSync
 )
 
 var actions = map[protoAction]string{
@@ -42,6 +45,9 @@ var actions = map[protoAction]string{
 	actionMessage:      "message",
 	actionSync:         "sync",
 	actionAuth:         "auth",
+	actionActivate:     "activate",
+	actionObject:       "object",
+	actionObjectSync:   "object_sync",
 }
 
 func (a protoAction) String() string {
diff --git a/ably/proto_channel_options.go b/ably/proto_channel_options.go
index 186b343..98893bb 100644
--- a/ably/proto_channel_options.go
+++ b/ably/proto_channel_options.go
@@ -14,6 +14,10 @@ const (
 	ChannelModeSubscribe
 	// ChannelModePresenceSubscribe allows the attached channel to subscribe to Presence updates.
 	ChannelModePresenceSubscribe
+	// ChannelModeObjectSubscribe allows the attached channel to subscribe to Objects updates.
+	ChannelModeObjectSubscribe
+	// ChannelModeObjectPublish allows for objects messages to be published on the attached channel.
+	ChannelModeObjectPublish
 )
 
 func (mode ChannelMode) toFlag() protoFlag {
@@ -26,6 +30,10 @@ func (mode ChannelMode) toFlag() protoFlag {
 		return flagSubscribe
 	case ChannelModePresenceSubscribe:
 		return flagPresenceSubscribe
+	case ChannelModeObjectSubscribe:
+		return flagObjectSubscribe
+	case ChannelModeObjectPublish:
+		return flagObjectPublish
 	default:
 		return 0
 	}
@@ -45,6 +53,12 @@ func channelModeFromFlag(flags protoFlag) []ChannelMode {
 	if flags.Has(flagPresenceSubscribe) {
 		modes = append(modes, ChannelModePresenceSubscribe)
 	}
+	if flags.Has(flagObjectSubscribe) {
+		modes = append(modes, ChannelModeObjectSubscribe)
+	}
+	if flags.Has(flagObjectPublish) {
+		modes = append(modes, ChannelModeObjectPublish)
+	}
 	return modes
 }
 
diff --git a/ably/proto_protocol_message.go b/ably/proto_protocol_message.go
index 2fc460b..5cce4e0 100644
--- a/ably/proto_protocol_message.go
+++ b/ably/proto_protocol_message.go
@@ -3,6 +3,8 @@ package ably
 import (
 	"fmt"
 	"time"
+
+	"github.com/ably/ably-go/ably/objects"
 )
 
 // TR3
@@ -16,6 +18,8 @@ const (
 	flagPublish           protoFlag = 1 << 17
 	flagSubscribe         protoFlag = 1 << 18
 	flagPresenceSubscribe protoFlag = 1 << 19
+	flagObjectSubscribe   protoFlag = 1 << 24
+	flagObjectPublish     protoFlag = 1 << 25
 )
 
 type protoFlag int64
@@ -116,6 +120,7 @@ func coerceInt64(v interface{}) int64 {
 type protocolMessage struct {
 	Messages          []*Message         `json:"messages,omitempty" codec:"messages,omitempty"`
 	Presence          []*PresenceMessage `json:"presence,omitempty" codec:"presence,omitempty"`
+	State             []*objects.Message `json:"state,omitempty" codec:"state,omitempty"`
 	ID                string             `json:"id,omitempty" codec:"id,omitempty"`
 	ApplicationID     string             `json:"applicationId,omitempty" codec:"applicationId,omitempty"`
 	ConnectionID      string             `json:"connectionId,omitempty" codec:"connectionId,omitempty"`
diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go
index 8446984..9a51d12 100644
--- a/ably/realtime_channel.go
+++ b/ably/realtime_channel.go
@@ -224,6 +224,16 @@ type RealtimeChannel struct {
 	// Presence is a [ably.RealtimePresence] object, provides for entering and leaving client presence (RTL9).
 	Presence *RealtimePresence
 
+	// ExperimentalObjects returns an experimental implementation of [LiveObjects] functionality
+	// capable of publishing and receiving object changes.
+	//
+	// NOTE: this method is experimental, the LiveObjects plugin API may change in a
+	// backwards incompatible way between minor/patch versions. Once the API has been finalised,
+	// a new non-experimental method will be added, and this one will be removed.
+	//
+	// [LiveObjects]: https://ably.com/docs/liveobjects
+	ExperimentalObjects *RealtimeExperimentalObjects
+
 	// state is the current [ably.ChannelState] of the channel (RTL2b).
 	state ChannelState
 
@@ -266,6 +276,7 @@ func newRealtimeChannel(name string, client *Realtime, chOptions *channelOptions
 	}
 	c.Presence = newRealtimePresence(c)
 	c.queue = newMsgQueue(client.Connection)
+	c.ExperimentalObjects = newRealtimeExperimentalObjects(c)
 	return c
 }
 
@@ -481,6 +492,18 @@ func (c *RealtimeChannel) detachUnsafe() (result, error) {
 	return c.sendDetachMsg()
 }
 
+func (c *RealtimeChannel) getClientOptions() *clientOptions {
+	if c.client == nil {
+		return nil
+	}
+
+	return c.client.opts()
+}
+
+func (c *RealtimeChannel) getName() string {
+	return c.Name
+}
+
 type subscriptionName string
 
 func (subscriptionName) isEmitterEvent() {}
@@ -861,6 +884,14 @@ func (c *RealtimeChannel) notify(msg *protocolMessage) {
 				c.messageEmitter.Emit(subscriptionName(msg.Name), (*subscriptionMessage)(msg))
 			}
 		}
+	case actionObject:
+		if plugin := c.client.opts().ExperimentalObjectsPlugin; plugin != nil {
+			plugin.HandleObjectMessages(msg.State)
+		}
+	case actionObjectSync:
+		if plugin := c.client.opts().ExperimentalObjectsPlugin; plugin != nil {
+			plugin.HandleObjectSyncMessages(msg.State, msg.ChannelSerial)
+		}
 	default:
 	}
 }
diff --git a/ably/realtime_experimental_objects.go b/ably/realtime_experimental_objects.go
new file mode 100644
index 0000000..b40ff5a
--- /dev/null
+++ b/ably/realtime_experimental_objects.go
@@ -0,0 +1,69 @@
+package ably
+
+import (
+	"context"
+	"errors"
+
+	"github.com/ably/ably-go/ably/objects"
+)
+
+// RealtimeExperimentalObjects exposes a method to publish LiveObject messages on a Realtime channel.
+// It requires the LiveObjects plugin to be configured see ably-go/ably/objects/liveobjects.go.
+//
+// NOTE: this option is experimental, the LiveObjects plugin API may change in a
+// backwards incompatible way between minor/patch versions. Once the API has been finalised,
+// a new non-experimental option will be added, and this one will be removed.
+//
+// [LiveObjects]: https://ably.com/docs/liveobjects
+type RealtimeExperimentalObjects struct {
+	channel channel
+}
+
+func newRealtimeExperimentalObjects(channel channel) *RealtimeExperimentalObjects {
+	return &RealtimeExperimentalObjects{channel: channel}
+}
+
+// PublishObjects publishes the given object messages to the channel.
+//
+// An objects plugin must be configured on the realtime client, which is used
+// to prepare the message to be published (e.g. setting the objectId and
+// initialValue for a COUNTER_CREATE op).
+func (o *RealtimeExperimentalObjects) PublishObjects(ctx context.Context, msgs ...*objects.Message) error {
+	plugin := o.channel.getClientOptions().ExperimentalObjectsPlugin
+	if plugin == nil {
+		return newError(ErrMissingPlugin, errors.New("missing objects plugin"))
+	}
+
+	for _, msg := range msgs {
+		if err := plugin.PrepareObject(msg); err != nil {
+			return err
+		}
+	}
+
+	listen := make(chan error, 1)
+	onAck := func(err error) {
+		listen <- err
+	}
+
+	msg := &protocolMessage{
+		Action:  actionObject,
+		Channel: o.channel.getName(),
+		State:   msgs,
+	}
+	if err := o.channel.send(msg, onAck); err != nil {
+		return err
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case err := <-listen:
+		return err
+	}
+}
+
+type channel interface {
+	send(msg *protocolMessage, onAck func(error)) error
+	getClientOptions() *clientOptions
+	getName() string
+}


Test patch:
diff --git a/ably/realtime_experimental_objects_test.go b/ably/realtime_experimental_objects_test.go
new file mode 100644
index 0000000..21d1eaa
--- /dev/null
+++ b/ably/realtime_experimental_objects_test.go
@@ -0,0 +1,310 @@
+package ably
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/ably/ably-go/ably/objects"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRealtimeExperimentalObjects_PublishObjects(t *testing.T) {
+	channelName := "test-channel"
+
+	tests := []struct {
+		name                 string
+		messages             []*objects.Message
+		plugin               objects.Plugin
+		sendError            error
+		ackError             error
+		expectedError        string
+		expectedProtocolMsg  *protocolMessage
+		expectedPrepareCount int
+	}{
+		{
+			name: "successful publish with single message",
+			messages: []*objects.Message{
+				{
+					ID:       "msg1",
+					ClientID: "client1",
+					Operation: &objects.Operation{
+						Action:   objects.Operation_CounterCreate,
+						ObjectID: "counter1",
+					},
+				},
+			},
+			plugin: &pluginMock{
+				PrepareObjectFunc: func(msg *objects.Message) error {
+					return nil
+				},
+			},
+			sendError:            nil,
+			ackError:             nil,
+			expectedError:        "",
+			expectedPrepareCount: 1,
+		},
+		{
+			name: "successful publish with multiple messages",
+			messages: []*objects.Message{
+				{
+					ID:       "msg1",
+					ClientID: "client1",
+					Operation: &objects.Operation{
+						Action:   objects.Operation_CounterCreate,
+						ObjectID: "counter1",
+					},
+				},
+				{
+					ID:       "msg2",
+					ClientID: "client1",
+					Operation: &objects.Operation{
+						Action:   objects.Operation_MapCreate,
+						ObjectID: "map1",
+					},
+				},
+			},
+			plugin: &pluginMock{
+				PrepareObjectFunc: func(msg *objects.Message) error {
+					return nil
+				},
+			},
+			sendError:            nil,
+			ackError:             nil,
+			expectedError:        "",
+			expectedPrepareCount: 2,
+		},
+		{
+			name: "error when no plugin configured",
+			messages: []*objects.Message{
+				{
+					ID:       "msg1",
+					ClientID: "client1",
+					Operation: &objects.Operation{
+						Action:   objects.Operation_CounterCreate,
+						ObjectID: "counter1",
+					},
+				},
+			},
+			plugin:               nil,
+			sendError:            nil,
+			ackError:             nil,
+			expectedError:        "missing objects plugin",
+			expectedPrepareCount: 0,
+		},
+		{
+			name: "error during prepare object",
+			messages: []*objects.Message{
+				{
+					ID:       "msg1",
+					ClientID: "client1",
+					Operation: &objects.Operation{
+						Action:   objects.Operation_CounterCreate,
+						ObjectID: "counter1",
+					},
+				},
+			},
+			plugin: &pluginMock{
+				PrepareObjectFunc: func(msg *objects.Message) error {
+					return errors.New("prepare failed")
+				},
+			},
+			sendError:            nil,
+			ackError:             nil,
+			expectedError:        "prepare failed",
+			expectedPrepareCount: 1,
+		},
+		{
+			name: "error during send",
+			messages: []*objects.Message{
+				{
+					ID:       "msg1",
+					ClientID: "client1",
+					Operation: &objects.Operation{
+						Action:   objects.Operation_CounterCreate,
+						ObjectID: "counter1",
+					},
+				},
+			},
+			plugin: &pluginMock{
+				PrepareObjectFunc: func(msg *objects.Message) error {
+					return nil
+				},
+			},
+			sendError:            errors.New("send failed"),
+			ackError:             nil,
+			expectedError:        "send failed",
+			expectedPrepareCount: 1,
+		},
+		{
+			name: "error during ack",
+			messages: []*objects.Message{
+				{
+					ID:       "msg1",
+					ClientID: "client1",
+					Operation: &objects.Operation{
+						Action:   objects.Operation_CounterCreate,
+						ObjectID: "counter1",
+					},
+				},
+			},
+			plugin: &pluginMock{
+				PrepareObjectFunc: func(msg *objects.Message) error {
+					return nil
+				},
+			},
+			sendError:            nil,
+			ackError:             errors.New("ack failed"),
+			expectedError:        "ack failed",
+			expectedPrepareCount: 1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var prepareObjectFuncCalled int
+			var capturedProtocolMsg *protocolMessage
+
+			// Create channel mock
+			channelMock := &channelMock{
+				SendFunc: func(msg *protocolMessage, onAck func(error)) error {
+					if tt.sendError != nil {
+						return tt.sendError
+					}
+					capturedProtocolMsg = msg
+					// Simulate async ack
+					go func() {
+						onAck(tt.ackError)
+					}()
+					return nil
+				},
+				GetClientOptionsFunc: func() *clientOptions {
+					return &clientOptions{
+						ExperimentalObjectsPlugin: tt.plugin,
+					}
+				},
+				GetNameFunc: func() string {
+					return channelName
+				},
+			}
+
+			// Wrap plugin to count prepare calls
+			var wrappedPlugin objects.Plugin
+			if tt.plugin != nil {
+				wrappedPlugin = &pluginMock{
+					PrepareObjectFunc: func(msg *objects.Message) error {
+						prepareObjectFuncCalled++
+						return tt.plugin.PrepareObject(msg)
+					},
+				}
+			}
+
+			// Update the channel's client options to use the wrapped plugin
+			channelMock.GetClientOptionsFunc = func() *clientOptions {
+				return &clientOptions{
+					ExperimentalObjectsPlugin: wrappedPlugin,
+				}
+			}
+
+			exp := newRealtimeExperimentalObjects(channelMock)
+
+			ctx := context.Background()
+			err := exp.PublishObjects(ctx, tt.messages...)
+
+			if tt.expectedError != "" {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), tt.expectedError)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			assert.Equal(t, tt.expectedPrepareCount, prepareObjectFuncCalled)
+
+			// Verify protocol message if no error expected
+			if tt.expectedError == "" && tt.sendError == nil {
+				assert.NotNil(t, capturedProtocolMsg)
+				assert.Equal(t, actionObject, capturedProtocolMsg.Action)
+				assert.Equal(t, channelName, capturedProtocolMsg.Channel)
+				assert.Equal(t, tt.messages, capturedProtocolMsg.State)
+			}
+		})
+	}
+}
+
+func TestRealtimeExperimentalObjects_PublishObjectsContextCancellation(t *testing.T) {
+	// Test context cancellation during publish
+	channelMock := &channelMock{
+		SendFunc: func(msg *protocolMessage, onAck func(error)) error {
+			// Don't call onAck to simulate hanging
+			return nil
+		},
+		GetClientOptionsFunc: func() *clientOptions {
+			return &clientOptions{
+				ExperimentalObjectsPlugin: &pluginMock{
+					PrepareObjectFunc: func(msg *objects.Message) error {
+						return nil
+					},
+				},
+			}
+		},
+		GetNameFunc: func() string {
+			return "test-channel"
+		},
+	}
+
+	exp := newRealtimeExperimentalObjects(channelMock)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // Cancel immediately
+
+	messages := []*objects.Message{
+		{
+			ID:       "msg1",
+			ClientID: "client1",
+			Operation: &objects.Operation{
+				Action:   objects.Operation_CounterCreate,
+				ObjectID: "counter1",
+			},
+		},
+	}
+
+	err := exp.PublishObjects(ctx, messages...)
+	assert.Error(t, err)
+	assert.Equal(t, context.Canceled, err)
+}
+
+// channelMock implements the channel interface
+type channelMock struct {
+	SendFunc             func(msg *protocolMessage, onAck func(error)) error
+	GetClientOptionsFunc func() *clientOptions
+	GetNameFunc          func() string
+}
+
+func (c channelMock) send(msg *protocolMessage, onAck func(error)) error {
+	return c.SendFunc(msg, onAck)
+}
+
+func (c channelMock) getClientOptions() *clientOptions {
+	return c.GetClientOptionsFunc()
+}
+
+func (c channelMock) getName() string {
+	return c.GetNameFunc()
+}
+
+// pluginMock implements the objects.Plugin interface
+type pluginMock struct {
+	PrepareObjectFunc func(msg *objects.Message) error
+}
+
+func (p pluginMock) PrepareObject(msg *objects.Message) error {
+	return p.PrepareObjectFunc(msg)
+}
+
+func (p pluginMock) HandleObjectMessages(msgs []*objects.Message) {
+	// implementation not required for test
+}
+
+func (p pluginMock) HandleObjectSyncMessages(msgs []*objects.Message, channelSerial string) {
+	// implementation not required for test
+}