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 {