Integration branch for v2 packages

ably/error_names.go

@@ -31,4 +31,5 @@ const (
 	ErrProtocolError                             ErrorCode = 80013
 	ErrChannelOperationFailed                    ErrorCode = 90000
 	ErrChannelOperationFailedInvalidChannelState ErrorCode = 90001
+	ErrChannelReleaseInvalidState                ErrorCode = 90011
 )

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
 	}
 }
 

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

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
 }
 

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))
 }

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"))
+		})
+	}
+}

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 {

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-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...)
+}

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-pubsub-device": ""}), agent)
+}

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-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...)
+}

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-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-pubsub-server"), "missing side agent: %q", *agent)
+	assert.True(t, strings.Contains(*agent, " foo/1.2.3"), "missing user agent: %q", *agent)
+}