Skip to content

Commit 5ff7349

Browse files
lmarslmars
committed
fixup! feat: Implement Message Updates, Deletes, and Appends
1 parent 6040888 commit 5ff7349

7 files changed

Lines changed: 191 additions & 148 deletions

‎ably/example_message_updates_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ func ExampleRESTChannel_PublishWithResult() {
2222
panic(err)
2323
}
2424

25-
fmt.Printf("Message published with serial: %s\n", result.Serial)
25+
fmt.Printf("Message published with serial: %s\n", *result.Serial)
2626
}
2727

2828
// Example demonstrating how to update a message
@@ -42,7 +42,7 @@ func ExampleRESTChannel_UpdateMessage() {
4242

4343
// Update the message
4444
msg := &ably.Message{
45-
Serial: result.Serial,
45+
Serial: *result.Serial,
4646
Data: "updated data",
4747
}
4848

@@ -56,7 +56,7 @@ func ExampleRESTChannel_UpdateMessage() {
5656
panic(err)
5757
}
5858

59-
fmt.Printf("Message updated with version serial: %s\n", updateResult.VersionSerial)
59+
fmt.Printf("Message updated with version serial: %s\n", *updateResult.VersionSerial)
6060
}
6161

6262
// Example demonstrating async message append for AI streaming
@@ -78,11 +78,11 @@ func ExampleRealtimeChannel_AppendMessageAsync() {
7878
tokens := []string{" 42", ".", " This", " is", " the", " answer."}
7979
for _, token := range tokens {
8080
msg := &ably.Message{
81-
Serial: result.Serial,
81+
Serial: *result.Serial,
8282
Data: token,
8383
}
8484
// Non-blocking append - critical for AI streaming
85-
err := channel.AppendMessageAsync(msg, func(r *ably.UpdateResult, err error) {
85+
err := channel.AppendMessageAsync(msg, func(r *ably.UpdateDeleteResult, err error) {
8686
if err != nil {
8787
fmt.Printf("Append failed: %v\n", err)
8888
}

‎ably/message_updates_integration_test.go‎

Lines changed: 42 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,8 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
3232
t.Run("returns serial for published message", func(t *testing.T) {
3333
result, err := channel.PublishWithResult(ctx, "event1", "test data")
3434
require.NoError(t, err)
35-
assert.NotEmpty(t, result.Serial, "Expected non-empty serial")
35+
require.NotNil(t, result.Serial, "Expected non-nil serial")
36+
assert.NotEmpty(t, *result.Serial, "Expected non-empty serial")
3637
})
3738

3839
t.Run("PublishMultipleWithResult returns serials for all messages", func(t *testing.T) {
@@ -47,7 +48,8 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
4748
assert.Len(t, results, 3)
4849

4950
for i, result := range results {
50-
assert.NotEmpty(t, result.Serial, "Expected non-empty serial for message %d", i)
51+
require.NotNil(t, result.Serial, "Expected non-nil serial for message %d", i)
52+
assert.NotEmpty(t, *result.Serial, "Expected non-empty serial for message %d", i)
5153
}
5254
})
5355
})
@@ -59,23 +61,25 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
5961
// Publish initial message
6062
publishResult, err := channel.PublishWithResult(ctx, "test-event", "initial data")
6163
require.NoError(t, err)
64+
require.NotNil(t, publishResult.Serial)
6265

6366
// Update the message
6467
msg := &ably.Message{
65-
Serial: publishResult.Serial,
68+
Serial: *publishResult.Serial,
6669
Data: "updated data",
6770
}
6871
updateResult, err := channel.UpdateMessage(ctx, msg,
6972
ably.UpdateWithDescription("Fixed typo"),
7073
ably.UpdateWithMetadata(map[string]string{"editor": "test"}),
7174
)
7275
require.NoError(t, err)
73-
assert.NotEmpty(t, updateResult.VersionSerial, "Expected version serial")
74-
assert.NotEqual(t, publishResult.Serial, updateResult.VersionSerial, "VersionSerial should differ from original Serial")
76+
require.NotNil(t, updateResult.VersionSerial, "Expected non-nil version serial")
77+
assert.NotEmpty(t, *updateResult.VersionSerial, "Expected version serial")
78+
assert.NotEqual(t, *publishResult.Serial, *updateResult.VersionSerial, "VersionSerial should differ from original Serial")
7579

7680
// Verify the update by fetching the message (eventually consistent)
7781
require.Eventually(t, func() bool {
78-
retrieved, err := channel.GetMessage(ctx, publishResult.Serial)
82+
retrieved, err := channel.GetMessage(ctx, *publishResult.Serial)
7983
if err != nil {
8084
return false
8185
}
@@ -102,16 +106,18 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
102106
// Publish initial message
103107
publishResult, err := channel.PublishWithResult(ctx, "test-event", "data to delete")
104108
require.NoError(t, err)
109+
require.NotNil(t, publishResult.Serial)
105110

106111
// Delete the message
107112
msg := &ably.Message{
108-
Serial: publishResult.Serial,
113+
Serial: *publishResult.Serial,
109114
}
110115
deleteResult, err := channel.DeleteMessage(ctx, msg,
111116
ably.UpdateWithDescription("Deleted by test"),
112117
)
113118
require.NoError(t, err)
114-
assert.NotEmpty(t, deleteResult.VersionSerial, "Expected version serial")
119+
require.NotNil(t, deleteResult.VersionSerial, "Expected non-nil version serial")
120+
assert.NotEmpty(t, *deleteResult.VersionSerial, "Expected version serial")
115121
})
116122
})
117123

@@ -122,19 +128,21 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
122128
// Publish initial message
123129
publishResult, err := channel.PublishWithResult(ctx, "test-event", "Hello")
124130
require.NoError(t, err)
131+
require.NotNil(t, publishResult.Serial)
125132

126133
// Append to the message
127134
msg := &ably.Message{
128-
Serial: publishResult.Serial,
135+
Serial: *publishResult.Serial,
129136
Data: " World",
130137
}
131138
appendResult, err := channel.AppendMessage(ctx, msg)
132139
require.NoError(t, err)
133-
assert.NotEmpty(t, appendResult.VersionSerial, "Expected version serial")
140+
require.NotNil(t, appendResult.VersionSerial, "Expected non-nil version serial")
141+
assert.NotEmpty(t, *appendResult.VersionSerial, "Expected version serial")
134142

135143
// Verify by fetching the message - data should be appended (eventually consistent)
136144
require.Eventually(t, func() bool {
137-
retrieved, err := channel.GetMessage(ctx, publishResult.Serial)
145+
retrieved, err := channel.GetMessage(ctx, *publishResult.Serial)
138146
if err != nil {
139147
return false
140148
}
@@ -151,14 +159,15 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
151159
// Publish a message
152160
publishResult, err := channel.PublishWithResult(ctx, "test-event", "test data")
153161
require.NoError(t, err)
162+
require.NotNil(t, publishResult.Serial)
154163

155164
// GetMessage is eventually consistent - retry until message is available
156165
require.Eventually(t, func() bool {
157-
msg, err := channel.GetMessage(ctx, publishResult.Serial)
166+
msg, err := channel.GetMessage(ctx, *publishResult.Serial)
158167
if err != nil {
159168
return false
160169
}
161-
return msg.Data == "test data" && msg.Serial == publishResult.Serial
170+
return msg.Data == "test data" && msg.Serial == *publishResult.Serial
162171
}, 5*time.Second, 100*time.Millisecond, "Message should be retrievable")
163172
})
164173
})
@@ -170,10 +179,11 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
170179
// Publish initial message
171180
publishResult, err := channel.PublishWithResult(ctx, "test-event", "version 1")
172181
require.NoError(t, err)
182+
require.NotNil(t, publishResult.Serial)
173183

174184
// Update the message twice
175185
msg := &ably.Message{
176-
Serial: publishResult.Serial,
186+
Serial: *publishResult.Serial,
177187
Data: "version 2",
178188
}
179189
_, err = channel.UpdateMessage(ctx, msg, ably.UpdateWithDescription("First update"))
@@ -186,7 +196,7 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
186196
// GetMessageVersions is eventually consistent - retry until all versions are available
187197
var versions []*ably.Message
188198
require.Eventually(t, func() bool {
189-
page, err := channel.GetMessageVersions(publishResult.Serial, nil).Pages(ctx)
199+
page, err := channel.GetMessageVersions(*publishResult.Serial, nil).Pages(ctx)
190200
if err != nil {
191201
return false
192202
}
@@ -204,9 +214,9 @@ func TestRESTChannel_MessageUpdates(t *testing.T) {
204214

205215
// Verify we have exactly 3 versions in the correct order
206216
require.Equal(t, 3, len(versions))
207-
assert.Equal(t, ably.MessageActionCreate, versions[0].Action, "First version should be message.create")
208-
assert.Equal(t, ably.MessageActionUpdate, versions[1].Action, "Second version should be message.update")
209-
assert.Equal(t, ably.MessageActionUpdate, versions[2].Action, "Third version should be message.update")
217+
assert.Equal(t, ably.MessageActionCreate, versions[0].Action, "First version should be MESSAGE_CREATE")
218+
assert.Equal(t, ably.MessageActionUpdate, versions[1].Action, "Second version should be MESSAGE_UPDATE")
219+
assert.Equal(t, ably.MessageActionUpdate, versions[2].Action, "Third version should be MESSAGE_UPDATE")
210220
})
211221
})
212222
}
@@ -236,7 +246,8 @@ func TestRealtimeChannel_MessageUpdates(t *testing.T) {
236246
t.Run("returns serial for published message", func(t *testing.T) {
237247
result, err := channel.PublishWithResult(ctx, "event1", "realtime data")
238248
require.NoError(t, err)
239-
assert.NotEmpty(t, result.Serial, "Expected non-empty serial")
249+
require.NotNil(t, result.Serial, "Expected non-nil serial")
250+
assert.NotEmpty(t, *result.Serial, "Expected non-empty serial")
240251
})
241252

242253
t.Run("PublishMultipleWithResult returns serials", func(t *testing.T) {
@@ -250,7 +261,8 @@ func TestRealtimeChannel_MessageUpdates(t *testing.T) {
250261
assert.Len(t, results, 2)
251262

252263
for i, result := range results {
253-
assert.NotEmpty(t, result.Serial, "Expected serial for message %d", i)
264+
require.NotNil(t, result.Serial, "Expected non-nil serial for message %d", i)
265+
assert.NotEmpty(t, *result.Serial, "Expected non-empty serial for message %d", i)
254266
}
255267
})
256268
})
@@ -266,16 +278,17 @@ func TestRealtimeChannel_MessageUpdates(t *testing.T) {
266278
// Publish initial message
267279
publishResult, err := channel.PublishWithResult(ctx, "event", "initial")
268280
require.NoError(t, err)
281+
require.NotNil(t, publishResult.Serial)
269282

270283
// Update asynchronously
271-
done := make(chan *ably.UpdateResult, 1)
284+
done := make(chan *ably.UpdateDeleteResult, 1)
272285
errChan := make(chan error, 1)
273286

274287
msg := &ably.Message{
275-
Serial: publishResult.Serial,
288+
Serial: *publishResult.Serial,
276289
Data: "updated async",
277290
}
278-
err = channel.UpdateMessageAsync(msg, func(result *ably.UpdateResult, err error) {
291+
err = channel.UpdateMessageAsync(msg, func(result *ably.UpdateDeleteResult, err error) {
279292
if err != nil {
280293
errChan <- err
281294
} else {
@@ -287,7 +300,8 @@ func TestRealtimeChannel_MessageUpdates(t *testing.T) {
287300
// Wait for callback
288301
select {
289302
case result := <-done:
290-
assert.NotEmpty(t, result.VersionSerial)
303+
require.NotNil(t, result.VersionSerial)
304+
assert.NotEmpty(t, *result.VersionSerial)
291305
case err := <-errChan:
292306
t.Fatalf("Update failed: %v", err)
293307
case <-time.After(5 * time.Second):
@@ -307,22 +321,23 @@ func TestRealtimeChannel_MessageUpdates(t *testing.T) {
307321
// Publish initial message
308322
publishResult, err := channel.PublishWithResult(ctx, "ai-response", "The answer is")
309323
require.NoError(t, err)
324+
require.NotNil(t, publishResult.Serial)
310325

311326
// Simulate rapid token streaming
312327
tokens := []string{" 42", ".", " This", " is", " correct", "."}
313328
type ack struct {
314-
result *ably.UpdateResult
329+
result *ably.UpdateDeleteResult
315330
err error
316331
}
317332
acks := make(chan ack, len(tokens))
318333

319334
for _, token := range tokens {
320335
msg := &ably.Message{
321-
Serial: publishResult.Serial,
336+
Serial: *publishResult.Serial,
322337
Data: token,
323338
}
324339

325-
err := channel.AppendMessageAsync(msg, func(result *ably.UpdateResult, err error) {
340+
err := channel.AppendMessageAsync(msg, func(result *ably.UpdateDeleteResult, err error) {
326341
acks <- ack{result, err}
327342
})
328343
require.NoError(t, err, "Failed to queue append %q", token)

0 commit comments

Comments
 (0)