Repository navigation
Expand file tree
/
Copy pathgoogle_cloud_push_queue_service.go
More file actions
110 lines (92 loc) · 3.03 KB
/
Copy pathgoogle_cloud_push_queue_service.go
File metadata and controls
110 lines (92 loc) · 3.03 KB
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
package services
import (
"context"
"fmt"
"net/http"
"time"
"github.com/avast/retry-go/v5"
cloudtasks "cloud.google.com/go/cloudtasks/apiv2"
"cloud.google.com/go/cloudtasks/apiv2/cloudtaskspb"
"github.com/NdoleStudio/httpsms/pkg/telemetry"
"github.com/NdoleStudio/stacktrace"
"google.golang.org/protobuf/types/known/timestamppb"
)
type googlePushQueue struct {
queueConfig PushQueueConfig
logger telemetry.Logger
tracer telemetry.Tracer
client *cloudtasks.Client
}
// NewGooglePushQueue creates a new googlePushQueue
func NewGooglePushQueue(
logger telemetry.Logger,
tracer telemetry.Tracer,
client *cloudtasks.Client,
queueConfig PushQueueConfig,
) PushQueue {
return &googlePushQueue{
client: client,
tracer: tracer,
logger: logger,
queueConfig: queueConfig,
}
}
// Enqueue a task to the queue
func (queue *googlePushQueue) Enqueue(ctx context.Context, task *PushQueueTask, timeout time.Duration) (queueID string, err error) {
err = retry.New(retry.Attempts(3)).Do(func() error {
queueID, err = queue.enqueueImpl(ctx, task, timeout)
return err
})
return queueID, err
}
// enqueueImpl a task to the queue
func (queue *googlePushQueue) enqueueImpl(ctx context.Context, task *PushQueueTask, timeout time.Duration) (queueID string, err error) {
ctx, span := queue.tracer.Start(ctx)
defer span.End()
ctxLogger := queue.tracer.CtxLogger(queue.logger, span)
headers := map[string]string{"Content-Type": "application/json"}
for key, value := range task.Headers {
headers[key] = value
}
// Build the Task payload.
req := &cloudtaskspb.CreateTaskRequest{
Parent: queue.queueConfig.Name,
Task: &cloudtaskspb.Task{
MessageType: &cloudtaskspb.Task_HttpRequest{
HttpRequest: &cloudtaskspb.HttpRequest{
Headers: headers,
HttpMethod: queue.httpMethodToProtoHTTPMethod(task.Method),
Url: task.URL,
},
},
ScheduleTime: timestamppb.New(time.Now().UTC().Add(timeout)),
},
}
// Add a payload message if one is present.
req.Task.GetHttpRequest().Body = task.Body
requestCtx, cancel := context.WithTimeout(ctx, time.Second)
defer cancel()
queueTask, err := queue.client.CreateTask(requestCtx, req)
if err != nil {
ctxLogger.Error(stacktrace.Propagatef(err, "cannot schedule task [%s] to URL [%s]", string(task.Body), task.URL))
return queueID, queue.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot schedule task [%s] to URL [%s]", string(task.Body), task.URL))
}
ctxLogger.Info(fmt.Sprintf(
"item added to [%s] queue with id [%s] and schedule [%s]",
queue.queueConfig.Name,
queueTask.Name,
queueTask.GetScheduleTime().AsTime(),
))
return queueTask.Name, nil
}
func (queue *googlePushQueue) httpMethodToProtoHTTPMethod(httpMethod string) cloudtaskspb.HttpMethod {
method, ok := map[string]cloudtaskspb.HttpMethod{
http.MethodGet: cloudtaskspb.HttpMethod_GET,
http.MethodPost: cloudtaskspb.HttpMethod_POST,
http.MethodPut: cloudtaskspb.HttpMethod_PUT,
}[httpMethod]
if !ok {
return cloudtaskspb.HttpMethod_POST
}
return method
}