From 07bb6d5bc601e6297edb71d26d04b4d0f8d9d7de Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Tue, 28 Jul 2026 23:23:10 +0100 Subject: [PATCH 1/3] pubsub: add device and server packages Introduce per-side pubsub packages, so applications declare whether they are running on an end-user device or a server by their choice of package, shippable in a minor release: - pubsub/device: NewClient, returning a Client (= ably.Realtime) - pubsub/server: NewHTTPClient and NewRealtimeClient, returning an HTTPClient (= ably.REST) and a RealtimeClient (= ably.Realtime) The constructors return type aliases of today's clients rather than wrapper types: the side is the package and constructor choice plus the agent declaration, not a type distinction. Each constructor appends a side-declaring agent (ably-go-pubsub-device / ably-go-pubsub-server) to the Ably-Agent header, which is what monthly-active-user classification keys on. WithAgents now merges its entries into any agents already configured instead of replacing them, so the side agent appended by the constructors and a user-supplied WithAgents coexist regardless of option order. Co-Authored-By: Claude Opus 5 (1M context) --- ably/options.go | 11 ++++++- pubsub/device/device.go | 21 ++++++++++++++ pubsub/device/device_test.go | 38 ++++++++++++++++++++++++ pubsub/server/server.go | 32 +++++++++++++++++++++ pubsub/server/server_test.go | 56 ++++++++++++++++++++++++++++++++++++ 5 files changed, 157 insertions(+), 1 deletion(-) create mode 100644 pubsub/device/device.go create mode 100644 pubsub/device/device_test.go create mode 100644 pubsub/server/server.go create mode 100644 pubsub/server/server_test.go diff --git a/ably/options.go b/ably/options.go index 70af0547..33d2eec7 100644 --- a/ably/options.go +++ b/ably/options.go @@ -1194,9 +1194,18 @@ func WithLogLevel(level LogLevel) ClientOption { // WithAgents is used to add product/version key-value pairs to include in the // agent library identifiers. This must only be used by Ably-authored SDKs. +// Successive calls merge their entries into any agents already configured +// rather than replacing them. func WithAgents(agents map[string]string) ClientOption { return func(os *clientOptions) { - os.Agents = agents + merged := make(map[string]string, len(os.Agents)+len(agents)) + for product, version := range os.Agents { + merged[product] = version + } + for product, version := range agents { + merged[product] = version + } + os.Agents = merged } } diff --git a/pubsub/device/device.go b/pubsub/device/device.go new file mode 100644 index 00000000..4e3c0228 --- /dev/null +++ b/pubsub/device/device.go @@ -0,0 +1,21 @@ +// Package device provides the Ably Pub/Sub client for devices: applications +// running on end-user devices, whose connections are identified by a +// clientId and counted on accounts with monthly-active-user billing. +package device + +import "github.com/ably/ably-go/ably" + +// agentName declares the side in the Ably-Agent header (RSC7d) so that +// traffic from clients constructed by this package is classified as +// device-side. +const agentName = "ably-go-pubsub-device" + +// Client is a device Pub/Sub client. +type Client = ably.Realtime + +// NewClient constructs a device Pub/Sub client: a realtime connection to +// Ably with channels, presence and history. +func NewClient(opts ...ably.ClientOption) (*Client, error) { + opts = append(opts[:len(opts):len(opts)], ably.WithAgents(map[string]string{agentName: ""})) + return ably.NewRealtime(opts...) +} diff --git a/pubsub/device/device_test.go b/pubsub/device/device_test.go new file mode 100644 index 00000000..c93ee3ef --- /dev/null +++ b/pubsub/device/device_test.go @@ -0,0 +1,38 @@ +package device_test + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "testing" + + "github.com/ably/ably-go/ably" + "github.com/ably/ably-go/pubsub/device" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewClient_DeclaresDeviceAgent(t *testing.T) { + var agent string + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + agent = r.Header.Get(ably.AblyAgentHeaderName) + w.WriteHeader(http.StatusInternalServerError) + })) + defer ts.Close() + u, err := url.Parse(ts.URL) + require.NoError(t, err) + + // The realtime client's HTTP requests carry the same agents as its + // connection, so assert the header via Time without connecting. + client, err := device.NewClient( + ably.WithEndpoint(u.Host), + ably.WithTLS(false), + ably.WithUseTokenAuth(true), + ably.WithAutoConnect(false), + ) + require.NoError(t, err) + + client.Time(context.Background()) + assert.Equal(t, ably.AgentIdentifier(map[string]string{"ably-go-pubsub-device": ""}), agent) +} diff --git a/pubsub/server/server.go b/pubsub/server/server.go new file mode 100644 index 00000000..cc223bad --- /dev/null +++ b/pubsub/server/server.go @@ -0,0 +1,32 @@ +// Package server provides the Ably Pub/Sub clients for servers: trusted +// environments which typically authenticate with an API key and whose +// connections are exempt from monthly-active-user counting. +package server + +import "github.com/ably/ably-go/ably" + +// agentName declares the side in the Ably-Agent header (RSC7d) so that +// traffic from clients constructed by this package is classified as +// server-side. +const agentName = "ably-go-pubsub-server" + +// HTTPClient is a server Pub/Sub client that operates entirely over HTTP. +type HTTPClient = ably.REST + +// RealtimeClient is a server Pub/Sub client with a realtime connection. +type RealtimeClient = ably.Realtime + +// NewHTTPClient constructs a server Pub/Sub client that operates entirely +// over HTTP: publish, history, presence reads, stats, token issuing. +func NewHTTPClient(opts ...ably.ClientOption) (*HTTPClient, error) { + opts = append(opts[:len(opts):len(opts)], ably.WithAgents(map[string]string{agentName: ""})) + return ably.NewREST(opts...) +} + +// NewRealtimeClient constructs a server Pub/Sub client with a persistent +// realtime connection: everything the HTTP client does, plus subscribing +// to channels and entering presence. +func NewRealtimeClient(opts ...ably.ClientOption) (*RealtimeClient, error) { + opts = append(opts[:len(opts):len(opts)], ably.WithAgents(map[string]string{agentName: ""})) + return ably.NewRealtime(opts...) +} diff --git a/pubsub/server/server_test.go b/pubsub/server/server_test.go new file mode 100644 index 00000000..aad70aa9 --- /dev/null +++ b/pubsub/server/server_test.go @@ -0,0 +1,56 @@ +package server_test + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + "github.com/ably/ably-go/ably" + "github.com/ably/ably-go/pubsub/server" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newAgentRecorder returns a test server which records the Ably-Agent header +// of each request it receives, along with options pointing a client at it. +func newAgentRecorder(t *testing.T) (agent *string, opts []ably.ClientOption) { + t.Helper() + agent = new(string) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + *agent = r.Header.Get(ably.AblyAgentHeaderName) + w.WriteHeader(http.StatusInternalServerError) + })) + t.Cleanup(ts.Close) + u, err := url.Parse(ts.URL) + require.NoError(t, err) + return agent, []ably.ClientOption{ + ably.WithEndpoint(u.Host), + ably.WithTLS(false), + ably.WithUseTokenAuth(true), + } +} + +func TestNewHTTPClient_DeclaresServerAgent(t *testing.T) { + agent, opts := newAgentRecorder(t) + + client, err := server.NewHTTPClient(opts...) + require.NoError(t, err) + + client.Time(context.Background()) + assert.Equal(t, ably.AgentIdentifier(map[string]string{"ably-go-pubsub-server": ""}), *agent) +} + +func TestNewHTTPClient_MergesUserAgents(t *testing.T) { + agent, opts := newAgentRecorder(t) + + client, err := server.NewHTTPClient(append(opts, ably.WithAgents(map[string]string{"foo": "1.2.3"}))...) + require.NoError(t, err) + + client.Time(context.Background()) + // Agent map iteration order is unspecified, so assert each entry. + assert.True(t, strings.Contains(*agent, " ably-go-pubsub-server"), "missing side agent: %q", *agent) + assert.True(t, strings.Contains(*agent, " foo/1.2.3"), "missing user agent: %q", *agent) +} From 24bd01ac841bd0f217271e86d799df6d8cdbd965 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 21 Sep 2026 21:02:26 +0100 Subject: [PATCH 2/3] Update Ably-Agent identifiers to 'ably-pubsub' standard --- ably/proto_http.go | 4 ++-- pubsub/device/device.go | 2 +- pubsub/device/device_test.go | 2 +- pubsub/server/server.go | 2 +- pubsub/server/server_test.go | 4 ++-- 5 files changed, 7 insertions(+), 7 deletions(-) diff --git a/ably/proto_http.go b/ably/proto_http.go index d6d66781..c9729abf 100644 --- a/ably/proto_http.go +++ b/ably/proto_http.go @@ -23,8 +23,8 @@ const ( ablyProtocolVersion = "5" // CSV2 ablyClientIDHeader = "X-Ably-ClientId" hostHeader = "Host" - ablyAgentHeader = "Ably-Agent" // RSC7d - ablySDKIdentifier = "ably-go/" + clientLibraryVersion // RSC7d1 + ablyAgentHeader = "Ably-Agent" // RSC7d + ablySDKIdentifier = "ably-pubsub-go/" + clientLibraryVersion // RSC7d1 ) // AblyAgentHeaderName is the name of the Ably-Agent header (RSC7d), exported for diff --git a/pubsub/device/device.go b/pubsub/device/device.go index 4e3c0228..81ea873d 100644 --- a/pubsub/device/device.go +++ b/pubsub/device/device.go @@ -8,7 +8,7 @@ import "github.com/ably/ably-go/ably" // agentName declares the side in the Ably-Agent header (RSC7d) so that // traffic from clients constructed by this package is classified as // device-side. -const agentName = "ably-go-pubsub-device" +const agentName = "ably-pubsub-device" // Client is a device Pub/Sub client. type Client = ably.Realtime diff --git a/pubsub/device/device_test.go b/pubsub/device/device_test.go index c93ee3ef..552ffec1 100644 --- a/pubsub/device/device_test.go +++ b/pubsub/device/device_test.go @@ -34,5 +34,5 @@ func TestNewClient_DeclaresDeviceAgent(t *testing.T) { require.NoError(t, err) client.Time(context.Background()) - assert.Equal(t, ably.AgentIdentifier(map[string]string{"ably-go-pubsub-device": ""}), agent) + assert.Equal(t, ably.AgentIdentifier(map[string]string{"ably-pubsub-device": ""}), agent) } diff --git a/pubsub/server/server.go b/pubsub/server/server.go index cc223bad..a7dcb657 100644 --- a/pubsub/server/server.go +++ b/pubsub/server/server.go @@ -8,7 +8,7 @@ import "github.com/ably/ably-go/ably" // agentName declares the side in the Ably-Agent header (RSC7d) so that // traffic from clients constructed by this package is classified as // server-side. -const agentName = "ably-go-pubsub-server" +const agentName = "ably-pubsub-server" // HTTPClient is a server Pub/Sub client that operates entirely over HTTP. type HTTPClient = ably.REST diff --git a/pubsub/server/server_test.go b/pubsub/server/server_test.go index aad70aa9..59ced619 100644 --- a/pubsub/server/server_test.go +++ b/pubsub/server/server_test.go @@ -40,7 +40,7 @@ func TestNewHTTPClient_DeclaresServerAgent(t *testing.T) { require.NoError(t, err) client.Time(context.Background()) - assert.Equal(t, ably.AgentIdentifier(map[string]string{"ably-go-pubsub-server": ""}), *agent) + assert.Equal(t, ably.AgentIdentifier(map[string]string{"ably-pubsub-server": ""}), *agent) } func TestNewHTTPClient_MergesUserAgents(t *testing.T) { @@ -51,6 +51,6 @@ func TestNewHTTPClient_MergesUserAgents(t *testing.T) { client.Time(context.Background()) // Agent map iteration order is unspecified, so assert each entry. - assert.True(t, strings.Contains(*agent, " ably-go-pubsub-server"), "missing side agent: %q", *agent) + assert.True(t, strings.Contains(*agent, " ably-pubsub-server"), "missing side agent: %q", *agent) assert.True(t, strings.Contains(*agent, " foo/1.2.3"), "missing user agent: %q", *agent) } From 1c4eaf0186e521fa5af274ca7988097d191068a9 Mon Sep 17 00:00:00 2001 From: Simon Woolf Date: Fri, 9 Oct 2026 04:11:29 +0100 Subject: [PATCH 3/3] ably: make Channels.Release return an error unless the channel is detached (RTS4c-e) Spec 6.3.0 replaces RTS4a with RTS4c-e. Release previously detached the channel and removed it once the detach completed, so a Get for the same name in that window returned the channel that was about to be removed, which then received no further messages. It also failed to release a FAILED channel, since detaching one is an error (RTL5b). Release now removes a channel in the INITIALIZED, DETACHED or FAILED state immediately, and for any other state returns an ErrorInfo with code 90011 and statusCode 400 without changing the channel or the collection. As it no longer blocks, it no longer takes a context. This is a breaking change: callers must call RealtimeChannel.Detach and wait for it to return before calling Channels.Release(name). ErrChannelReleaseInvalidState is added by hand to error_names.go. 90011 is registered in ably-common, but the common pin is not bumped: the errors.go generator cannot parse the current errors.json format. Co-Authored-By: Claude Opus 5.5 (1M context) --- ably/error_names.go | 1 + ably/realtime_channel.go | 33 ++++++++++------- ably/realtime_channel_integration_test.go | 26 +++++++++----- ably/realtime_channel_internal_test.go | 43 +++++++++++++++++++++++ ably/state.go | 10 ++++++ 5 files changed, 92 insertions(+), 21 deletions(-) diff --git a/ably/error_names.go b/ably/error_names.go index ffa11b14..db1aece3 100644 --- a/ably/error_names.go +++ b/ably/error_names.go @@ -31,4 +31,5 @@ const ( ErrProtocolError ErrorCode = 80013 ErrChannelOperationFailed ErrorCode = 90000 ErrChannelOperationFailedInvalidChannelState ErrorCode = 90001 + ErrChannelReleaseInvalidState ErrorCode = 90011 ) diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go index 1d609839..a598358a 100644 --- a/ably/realtime_channel.go +++ b/ably/realtime_channel.go @@ -203,23 +203,32 @@ func (c *RealtimeChannels) Exists(name string) bool { return ok } -// Release releases a [ably.RealtimeChannel] object with given channel name (detaching it first), frees all -// resources associated, e.g. Removes any listeners associated with the channel. -// To release a channel, the [ably.ChannelState] must be INITIALIZED, DETACHED, or FAILED (RSN4, RTS4). -func (ch *RealtimeChannels) Release(ctx context.Context, name string) error { +// Release releases a [ably.RealtimeChannel] object with given channel name, frees all +// resources associated, e.g. Removes any listeners associated with the channel (RSN4, RTS4). +// +// A realtime channel can only be released when it is in the INITIALIZED, DETACHED, or FAILED state. +// For a channel in any other state, Release returns an [ably.ErrorInfo] with code 90011 and leaves the +// channel as it is. Call [ably.RealtimeChannel.Detach] and wait for it to return before releasing it. +func (ch *RealtimeChannels) Release(name string) error { ch.mtx.Lock() + defer ch.mtx.Unlock() c, ok := ch.chans[name] - ch.mtx.Unlock() if !ok { - return nil + return nil // RTS4c } - err := c.Detach(ctx) - if err != nil { - return err + // Hold the channel lock until the channel is removed, so that a concurrent Attach can't move it + // out of a releasable state in between. + c.mtx.Lock() + defer c.mtx.Unlock() + if state := c.state; !state.canRelease() { + // RTS4e + return &ErrorInfo{ + Code: ErrChannelReleaseInvalidState, + StatusCode: 400, + err: fmt.Errorf("can only release a channel in a state where there is no possibility of further updates from the server being received (INITIALIZED, DETACHED, or FAILED); the current state is %s", state), + } } - ch.mtx.Lock() - delete(ch.chans, name) - ch.mtx.Unlock() + delete(ch.chans, name) // RTS4d return nil } diff --git a/ably/realtime_channel_integration_test.go b/ably/realtime_channel_integration_test.go index 9e96b8e8..48457356 100644 --- a/ably/realtime_channel_integration_test.go +++ b/ably/realtime_channel_integration_test.go @@ -346,7 +346,7 @@ func TestRealtimeChannel_ShouldReturnErrorIfReadLimitExceeded(t *testing.T) { assert.Equal(t, "failed to read: read limited at 1025 bytes", errorInfo.Unwrap().Error()) } -// Test that RealtimeChannels.Release detaches a channel and releases it. +// Test that RealtimeChannels.Release refuses to release an attached channel, and releases it once detached. func TestRealtimeChannels_Release(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -376,17 +376,25 @@ func TestRealtimeChannels_Release(t *testing.T) { ablytest.Soon.Recv(t, &channelStatechange, channelStateChanges, t.Fatalf) assert.Equal(t, ably.ChannelStateAttached, channelStatechange.Current) - t.Log("releasing test channel") - err = client.Channels.Release(ctx, ablytest.UniqueChannelName(t, "test")) - assert.NoError(t, err) - - t.Log("checking test channel is detached") + t.Log("releasing attached test channel") + err = client.Channels.Release(channel.Name) + var errInfo *ably.ErrorInfo + require.ErrorAs(t, err, &errInfo) + assert.Equal(t, ably.ErrChannelReleaseInvalidState, errInfo.Code) + assert.Equal(t, 400, errInfo.StatusCode) + assert.Equal(t, ably.ChannelStateAttached, channel.State()) + assert.True(t, client.Channels.Exists(channel.Name)) + + t.Log("detaching test channel") + err = channel.Detach(ctx) + require.NoError(t, err) ablytest.Soon.Recv(t, &channelStatechange, channelStateChanges, t.Fatalf) assert.Equal(t, ably.ChannelStateDetaching, channelStatechange.Current) ablytest.Soon.Recv(t, &channelStatechange, channelStateChanges, t.Fatalf) assert.Equal(t, ably.ChannelStateDetached, channelStatechange.Current) - t.Log("checking test channel is released") - exists := client.Channels.Exists(ablytest.UniqueChannelName(t, "test")) - assert.False(t, exists) + t.Log("releasing detached test channel") + err = client.Channels.Release(channel.Name) + assert.NoError(t, err) + assert.False(t, client.Channels.Exists(channel.Name)) } diff --git a/ably/realtime_channel_internal_test.go b/ably/realtime_channel_internal_test.go index bc15128b..112c16ca 100644 --- a/ably/realtime_channel_internal_test.go +++ b/ably/realtime_channel_internal_test.go @@ -4,9 +4,11 @@ package ably import ( + "fmt" "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestChannelOptionChannelWithCipherKey(t *testing.T) { @@ -163,3 +165,44 @@ func TestChannelGet(t *testing.T) { }) } } + +func TestChannelsRelease(t *testing.T) { + newClient := func(t *testing.T) *Realtime { + client, err := NewRealtime(WithKey("abc:def"), WithAutoConnect(false)) + require.NoError(t, err) + return client + } + + t.Run("RTS4c: releasing a non-existent channel is a no-op", func(t *testing.T) { + client := newClient(t) + assert.NoError(t, client.Channels.Release("nonexistent")) + }) + + for _, state := range []ChannelState{ChannelStateInitialized, ChannelStateDetached, ChannelStateFailed} { + t.Run(fmt.Sprintf("RTS4d: releasing a %s channel removes it", state), func(t *testing.T) { + client := newClient(t) + channel := client.Channels.Get("test") + channel.state = state + assert.NoError(t, client.Channels.Release("test")) + assert.False(t, client.Channels.Exists("test")) + assert.Equal(t, state, channel.State()) + assert.NotSame(t, channel, client.Channels.Get("test")) + }) + } + + for _, state := range []ChannelState{ChannelStateAttaching, ChannelStateAttached, ChannelStateDetaching, ChannelStateSuspended} { + t.Run(fmt.Sprintf("RTS4e: releasing a %s channel fails and leaves it in place", state), func(t *testing.T) { + client := newClient(t) + channel := client.Channels.Get("test") + channel.state = state + err := client.Channels.Release("test") + var errInfo *ErrorInfo + require.ErrorAs(t, err, &errInfo) + assert.Equal(t, ErrChannelReleaseInvalidState, errInfo.Code) + assert.Equal(t, 400, errInfo.StatusCode) + assert.Contains(t, errInfo.Message(), "the current state is "+state.String()) + assert.Equal(t, state, channel.State()) + assert.Same(t, channel, client.Channels.Get("test")) + }) + } +} diff --git a/ably/state.go b/ably/state.go index c3c5bf91..8f007421 100644 --- a/ably/state.go +++ b/ably/state.go @@ -580,6 +580,16 @@ func (e ChannelState) String() string { return e.name } +// canRelease reports whether a channel in this state can no longer receive updates from +// the server, so can be released (RTS4d). +func (e ChannelState) canRelease() bool { + switch e { + case ChannelStateInitialized, ChannelStateDetached, ChannelStateFailed: + return true + } + return false +} + // ChannelEvent describes the events emitted by a [ably.RESTChannel] or [ably.RealtimeChannel] object. // An event is either an UPDATE or a [ably.ChannelState]. type ChannelEvent struct {