-
Notifications
You must be signed in to change notification settings - Fork 237
/
messaging_batch.go
372 lines (319 loc) · 10.7 KB
/
messaging_batch.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
// Copyright 2019 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package messaging
import (
"bufio"
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
"mime"
"mime/multipart"
"net/http"
"net/textproto"
"firebase.google.com/go/internal"
)
const maxMessages = 500
const multipartBoundary = "__END_OF_PART__"
// MulticastMessage represents a message that can be sent to multiple devices via Firebase Cloud
// Messaging (FCM).
//
// It contains payload information as well as the list of device registration tokens to which the
// message should be sent. A single MulticastMessage may contain up to 500 registration tokens.
type MulticastMessage struct {
Tokens []string
Data map[string]string
Notification *Notification
Android *AndroidConfig
Webpush *WebpushConfig
APNS *APNSConfig
}
func (mm *MulticastMessage) toMessages() ([]*Message, error) {
if len(mm.Tokens) == 0 {
return nil, errors.New("tokens must not be nil or empty")
}
if len(mm.Tokens) > maxMessages {
return nil, fmt.Errorf("tokens must not contain more than %d elements", maxMessages)
}
var messages []*Message
for _, token := range mm.Tokens {
temp := &Message{
Token: token,
Data: mm.Data,
Notification: mm.Notification,
Android: mm.Android,
Webpush: mm.Webpush,
APNS: mm.APNS,
}
messages = append(messages, temp)
}
return messages, nil
}
// SendResponse represents the status of an individual message that was sent as part of a batch
// request.
type SendResponse struct {
Success bool
MessageID string
Error error
}
// BatchResponse represents the response from the `SendAll()` and `SendMulticast()` APIs.
type BatchResponse struct {
SuccessCount int
FailureCount int
Responses []*SendResponse
}
// SendAll sends the messages in the given array via Firebase Cloud Messaging.
//
// The messages array may contain up to 500 messages. SendAll employs batching to send the entire
// array of mssages as a single RPC call. Compared to the `Send()` function,
// this is a significantly more efficient way to send multiple messages. The responses list
// obtained from the return value corresponds to the order of the input messages. An error from
// SendAll indicates a total failure -- i.e. none of the messages in the array could be sent.
// Partial failures are indicated by a `BatchResponse` return value.
func (c *fcmClient) SendAll(ctx context.Context, messages []*Message) (*BatchResponse, error) {
return c.sendBatch(ctx, messages, false)
}
// SendAllDryRun sends the messages in the given array via Firebase Cloud Messaging in the
// dry run (validation only) mode.
//
// This function does not actually deliver any messages to target devices. Instead, it performs all
// the SDK-level and backend validations on the messages, and emulates the send operation.
//
// The messages array may contain up to 500 messages. SendAllDryRun employs batching to send the
// entire array of mssages as a single RPC call. Compared to the `SendDryRun()` function, this
// is a significantly more efficient way to validate sending multiple messages. The responses list
// obtained from the return value corresponds to the order of the input messages. An error from
// SendAllDryRun indicates a total failure -- i.e. none of the messages in the array could be sent
// for validation. Partial failures are indicated by a `BatchResponse` return value.
func (c *fcmClient) SendAllDryRun(ctx context.Context, messages []*Message) (*BatchResponse, error) {
return c.sendBatch(ctx, messages, true)
}
// SendMulticast sends the given multicast message to all the FCM registration tokens specified.
//
// The tokens array in MulticastMessage may contain up to 500 tokens. SendMulticast uses the
// `SendAll()` function to send the given message to all the target recipients. The
// responses list obtained from the return value corresponds to the order of the input tokens. An
// error from SendMulticast indicates a total failure -- i.e. the message could not be sent to any
// of the recipients. Partial failures are indicated by a `BatchResponse` return value.
func (c *fcmClient) SendMulticast(ctx context.Context, message *MulticastMessage) (*BatchResponse, error) {
messages, err := toMessages(message)
if err != nil {
return nil, err
}
return c.SendAll(ctx, messages)
}
// SendMulticastDryRun sends the given multicast message to all the specified FCM registration
// tokens in the dry run (validation only) mode.
//
// This function does not actually deliver any messages to target devices. Instead, it performs all
// the SDK-level and backend validations on the messages, and emulates the send operation.
//
// The tokens array in MulticastMessage may contain up to 500 tokens. SendMulticastDryRun uses the
// `SendAllDryRun()` function to send the given message. The responses list obtained from
// the return value corresponds to the order of the input tokens. An error from SendMulticastDryRun
// indicates a total failure -- i.e. none of the messages were sent to FCM for validation. Partial
// failures are indicated by a `BatchResponse` return value.
func (c *fcmClient) SendMulticastDryRun(ctx context.Context, message *MulticastMessage) (*BatchResponse, error) {
messages, err := toMessages(message)
if err != nil {
return nil, err
}
return c.SendAllDryRun(ctx, messages)
}
func toMessages(message *MulticastMessage) ([]*Message, error) {
if message == nil {
return nil, errors.New("message must not be nil")
}
return message.toMessages()
}
func (c *fcmClient) sendBatch(
ctx context.Context, messages []*Message, dryRun bool) (*BatchResponse, error) {
if len(messages) == 0 {
return nil, errors.New("messages must not be nil or empty")
}
if len(messages) > maxMessages {
return nil, fmt.Errorf("messages must not contain more than %d elements", maxMessages)
}
request, err := c.newBatchRequest(messages, dryRun)
if err != nil {
return nil, err
}
resp, err := c.httpClient.Do(ctx, request)
if err != nil {
return nil, err
}
if resp.Status != http.StatusOK {
return nil, handleFCMError(resp)
}
return newBatchResponse(resp)
}
// part represents a HTTP request that can be sent embedded in a multipart batch request.
//
// See https://cloud.google.com/compute/docs/api/how-tos/batch for details on how GCP APIs support multipart batch
// requests.
type part struct {
method string
url string
headers map[string]string
body interface{}
}
// multipartEntity represents an HTTP entity that consists of multiple HTTP requests (parts).
type multipartEntity struct {
parts []*part
}
func (c *fcmClient) newBatchRequest(messages []*Message, dryRun bool) (*internal.Request, error) {
url := fmt.Sprintf("%s/projects/%s/messages:send", c.fcmEndpoint, c.project)
headers := map[string]string{
apiFormatVersionHeader: apiFormatVersion,
firebaseClientHeader: c.version,
}
var parts []*part
for idx, m := range messages {
if err := validateMessage(m); err != nil {
return nil, fmt.Errorf("invalid message at index %d: %v", idx, err)
}
p := &part{
method: http.MethodPost,
url: url,
body: &fcmRequest{
Message: m,
ValidateOnly: dryRun,
},
headers: headers,
}
parts = append(parts, p)
}
return &internal.Request{
Method: http.MethodPost,
URL: c.batchEndpoint,
Body: &multipartEntity{parts: parts},
Opts: []internal.HTTPOption{
internal.WithHeader(firebaseClientHeader, c.version),
},
}, nil
}
func newBatchResponse(resp *internal.Response) (*BatchResponse, error) {
_, params, err := mime.ParseMediaType(resp.Header.Get("Content-Type"))
if err != nil {
return nil, fmt.Errorf("error parsing content-type header: %v", err)
}
mr := multipart.NewReader(bytes.NewBuffer(resp.Body), params["boundary"])
var responses []*SendResponse
successCount := 0
for {
part, err := mr.NextPart()
if err == io.EOF {
break
} else if err != nil {
return nil, err
}
sr, err := newSendResponse(part)
if err != nil {
return nil, err
}
responses = append(responses, sr)
if sr.Success {
successCount++
}
}
return &BatchResponse{
Responses: responses,
SuccessCount: successCount,
FailureCount: len(responses) - successCount,
}, nil
}
func newSendResponse(part *multipart.Part) (*SendResponse, error) {
hr, err := http.ReadResponse(bufio.NewReader(part), nil)
if err != nil {
return nil, fmt.Errorf("error parsing multipart body: %v", err)
}
b, err := ioutil.ReadAll(hr.Body)
if err != nil {
return nil, err
}
if hr.StatusCode != http.StatusOK {
resp := &internal.Response{
Status: hr.StatusCode,
Header: hr.Header,
Body: b,
}
return &SendResponse{
Success: false,
Error: handleFCMError(resp),
}, nil
}
var result fcmResponse
if err := json.Unmarshal(b, &result); err != nil {
return nil, err
}
return &SendResponse{
Success: true,
MessageID: result.Name,
}, nil
}
func (e *multipartEntity) Mime() string {
return fmt.Sprintf("multipart/mixed; boundary=%s", multipartBoundary)
}
func (e *multipartEntity) Bytes() ([]byte, error) {
var buffer bytes.Buffer
writer := multipart.NewWriter(&buffer)
writer.SetBoundary(multipartBoundary)
for idx, part := range e.parts {
if err := part.writeTo(writer, idx); err != nil {
return nil, err
}
}
writer.Close()
return buffer.Bytes(), nil
}
func (p *part) writeTo(writer *multipart.Writer, idx int) error {
b, err := p.bytes()
if err != nil {
return err
}
header := make(textproto.MIMEHeader)
header.Add("Content-Length", fmt.Sprintf("%d", len(b)))
header.Add("Content-Type", "application/http")
header.Add("Content-Id", fmt.Sprintf("%d", idx+1))
header.Add("Content-Transfer-Encoding", "binary")
part, err := writer.CreatePart(header)
if err != nil {
return err
}
_, err = part.Write(b)
return err
}
func (p *part) bytes() ([]byte, error) {
b, err := json.Marshal(p.body)
if err != nil {
return nil, err
}
req, err := http.NewRequest(p.method, p.url, bytes.NewBuffer(b))
if err != nil {
return nil, err
}
for key, value := range p.headers {
req.Header.Set(key, value)
}
req.Header.Set("Content-Type", "application/json; charset=UTF-8")
req.Header.Set("User-Agent", "")
var buffer bytes.Buffer
if err := req.Write(&buffer); err != nil {
return nil, err
}
return buffer.Bytes(), nil
}