You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
- memqueue - in memory queue that can be used for local unit testing.
39
-
- azsqs - Amazon SQS backend.
40
-
- ironmq - IronMQ backend.
41
-
- Manager - provides common interface for creating new queues.
42
-
- Processor - queue processor that works with memqueue, azsqs, and ironmq.
63
+
API that adds tasks:
43
64
44
-
rate limiting is implemented in the processor package using [redis_rate](https://github.com/go-redis/redis_rate). Call once is implemented in clients by checking if message name exists in Redis database.
65
+
```go
66
+
for {
67
+
err:= api_worker.CountTask.Call()
68
+
if err != nil {
69
+
log.Fatal(err)
70
+
}
71
+
api_worker.IncrLocalCounter()
72
+
}
73
+
```
45
74
46
-
## API overview
75
+
Worker that processes tasks:
47
76
48
-
```go
49
-
import"github.com/go-msgqueue/msgqueue"
50
-
import"github.com/go-redis/redis"
51
-
import"golang.org/x/time/rate"
52
-
53
-
// Create in-memory queue that prints greetings.
54
-
q:= memqueue.NewQueue(&msgqueue.Options{
55
-
// Handler is automatically retried on error.
56
-
Handler: func(name string) error {
57
-
fmt.Println("Hello", name)
58
-
returnnil
59
-
},
77
+
```go
78
+
err:= api_worker.QueueFactory.StartConsumers()
79
+
if err != nil {
80
+
log.Fatal(err)
81
+
}
82
+
```
60
83
61
-
RateLimit: rate.Every(time.Second),
84
+
## API overview
62
85
63
-
// Redis is only needed for rate limiting and call once.
64
-
Redis: redis.NewClient(&redis.Options{
65
-
Addr: ":6379",
66
-
}),
86
+
```go
87
+
t:= myQueue.NewTask(&taskq.TaskOptions{
88
+
Name: "greeting",
89
+
Handler: func(name string) error {
90
+
fmt.Println("Hello", name)
91
+
returnnil
92
+
},
67
93
})
68
94
69
-
//Invoke handler with arguments.
70
-
q.Call("World")
95
+
//Say "Hello World".
96
+
t.Call("World")
71
97
72
98
// Same using Message API.
73
-
q.Add(msgqueue.NewMessage("World"))
99
+
t.AddMessage(taskq.NewMessage("World"))
74
100
75
101
// Say "Hello World" with 1 hour delay.
76
-
msg:=msgqueue.NewMessage("World")
102
+
msg:=taskq.NewMessage("World")
77
103
msg.Delay = time.Hour
78
-
q.Add(msg)
104
+
t.AddMessage(msg)
79
105
80
106
// Say "Hello World" once.
81
107
fori:=0; i < 100; i++ {
82
-
msg:=msgqueue.NewMessage("hello")
83
-
msg.Name = "hello-world"
84
-
q.Add(msg)
108
+
msg:=taskq.NewMessage("hello")
109
+
msg.Name = "hello-world"// unique
110
+
t.Add(msg)
85
111
}
86
112
87
113
// Say "Hello World" once with 1 hour delay.
88
114
fori:=0; i < 100; i++ {
89
-
msg:=msgqueue.NewMessage("hello")
115
+
msg:=taskq.NewMessage("hello")
90
116
msg.Name = "hello-world"
91
117
msg.Delay = time.Hour
92
-
q.Add(msg)
118
+
t.Add(msg)
93
119
}
94
120
95
121
// Say "Hello World" once in an hour.
96
122
fori:=0; i < 100; i++ {
97
-
q.CallOnce(time.Hour, "hello")
123
+
t.CallOnce(time.Hour, "hello")
98
124
}
99
125
100
126
// Say "Hello World" for Europe region once in an hour.
101
127
fori:=0; i < 100; i++ {
102
-
msg:=msgqueue.NewMessage("hello")
103
-
msg.SetDelayName(delay, "europe") // set delay and autogenerate message name
104
-
q.Add(msg)
128
+
msg:=taskq.NewMessage("hello")
129
+
msg.OnceWithArgs(time.Hour, "europe") // set delay and autogenerate message name
130
+
t.Add(msg)
105
131
}
106
132
```
107
133
108
-
## SQS, IronMQ, and in-memory queues
109
-
110
-
SQS, IronMQ, and memqueue share the same API and can be used interchangeably.
111
-
112
-
### SQS
113
-
114
-
azsqs package uses Amazon Simple Queue Service as queue backend.
memqueue is in-memory queue backend implementation primarily useful for local development / unit testing. Unlike SQS and IronMQ it has running queue processor by default.
182
-
183
-
```go
184
-
import"github.com/go-msgqueue/msgqueue"
185
-
186
-
// Create queue.
187
-
q:= memqueue.NewQueue(&msgqueue.Options{
188
-
Handler: func(name string) error {
189
-
fmt.Println("Hello", name)
190
-
returnnil
191
-
},
192
-
})
193
-
194
-
// Same using Manager.
195
-
man:= memqueue.NewManager()
196
-
q:= man.NewQueue(&msgqueue.Options{...})
197
-
198
-
// Stop processor if you don't need it.
199
-
p:= q.Processor()
200
-
p.Stop()
201
-
202
-
// Process one message.
203
-
err:= p.ProcessOne()
204
-
205
-
// Process all buffered messages.
206
-
err:= p.ProcessAll()
207
-
```
208
-
209
134
## Custom message delay
210
135
211
-
If error returned by handler implements `Delay() time.Duration` that delay is used to postpone message processing.
136
+
If error returned by handler implements `Delay() time.Duration`interface then that delay is used to postpone message processing.
0 commit comments