Skip to content

Commit 835e27d

Browse files
appleboyappleboy
committed
chore(queue): update queue message package
Signed-off-by: Bo-Yi Wu <[email protected]>
1 parent 3f9a6f4 commit 835e27d

File tree

5 files changed

+27
-24
lines changed

5 files changed

+27
-24
lines changed

‎go.mod‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,9 @@ module github.com/golang-queue/nsq
33
go 1.18
44

55
require (
6-
github.com/golang-queue/queue v0.0.13-0.20220403053215-9d5c5466dc9a
6+
github.com/golang-queue/queue v0.0.13-0.20220408035349-ed24fa14aa00
77
github.com/nsqio/go-nsq v1.1.0
8-
github.com/stretchr/testify v1.7.0
8+
github.com/stretchr/testify v1.7.1
99
go.uber.org/goleak v1.1.12
1010
)
1111

‎go.sum‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,8 @@
11
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
22
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
3-
github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747 h1:uNTbCoWORAcna89KcKgP22WFGv5fsij05e70DCnLrUU=
4-
github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747/go.mod h1:ku8iyjYffqYY6Duts+xl+QYfN3/KDK4MEvXMZUkHyio=
5-
github.com/golang-queue/queue v0.0.13-0.20220403053215-9d5c5466dc9a h1:XeBmQF8PCff5DwphR2eZPGKzVnT8/34/JK1IUcSoSpc=
6-
github.com/golang-queue/queue v0.0.13-0.20220403053215-9d5c5466dc9a/go.mod h1:ku8iyjYffqYY6Duts+xl+QYfN3/KDK4MEvXMZUkHyio=
3+
github.com/golang-queue/queue v0.0.13-0.20220408035349-ed24fa14aa00 h1:EFiINOvAuGgxiE3MNu7PKY0O2Pvvc7r6YApdqAnhWnQ=
4+
github.com/golang-queue/queue v0.0.13-0.20220408035349-ed24fa14aa00/go.mod h1:g1yxxDl8JMo4gUfxt11fjjU3SXU1ah61EvwshmDoSIs=
5+
github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
76
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
87
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
98
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
@@ -16,8 +15,9 @@ github.com/nsqio/go-nsq v1.1.0/go.mod h1:vKq36oyeVXgsS5Q8YEO7WghqidAVXQlcFxzQbQT
1615
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
1716
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
1817
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
19-
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
2018
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
19+
github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
20+
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
2121
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
2222
go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA=
2323
go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=

‎nsq.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,11 +8,12 @@ import (
88
"time"
99

1010
"github.com/golang-queue/queue"
11+
"github.com/golang-queue/queue/core"
1112

1213
"github.com/nsqio/go-nsq"
1314
)
1415

15-
var _ queue.Worker = (*Worker)(nil)
16+
var _ core.Worker = (*Worker)(nil)
1617

1718
// Worker for NSQ
1819
type Worker struct {
@@ -145,7 +146,7 @@ func (w *Worker) handle(job queue.Job) error {
145146
}
146147

147148
// Run start the worker
148-
func (w *Worker) Run(task queue.QueuedMessage) error {
149+
func (w *Worker) Run(task core.QueuedMessage) error {
149150
data, _ := task.(queue.Job)
150151

151152
if err := w.handle(data); err != nil {
@@ -178,7 +179,7 @@ func (w *Worker) Shutdown() error {
178179
}
179180

180181
// Queue send notification to queue
181-
func (w *Worker) Queue(job queue.QueuedMessage) error {
182+
func (w *Worker) Queue(job core.QueuedMessage) error {
182183
if atomic.LoadInt32(&w.stopFlag) == 1 {
183184
return queue.ErrQueueShutdown
184185
}
@@ -187,7 +188,7 @@ func (w *Worker) Queue(job queue.QueuedMessage) error {
187188
}
188189

189190
// Request fetch new task from queue
190-
func (w *Worker) Request() (queue.QueuedMessage, error) {
191+
func (w *Worker) Request() (core.QueuedMessage, error) {
191192
clock := 0
192193
loop:
193194
for {

‎nsq_test.go‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
"time"
1111

1212
"github.com/golang-queue/queue"
13+
"github.com/golang-queue/queue/core"
1314

1415
"github.com/stretchr/testify/assert"
1516
"go.uber.org/goleak"
@@ -78,7 +79,7 @@ func TestNSQCustomFuncAndWait(t *testing.T) {
7879
WithAddr(host+":4150"),
7980
WithTopic("test3"),
8081
WithMaxInFlight(10),
81-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
82+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
8283
time.Sleep(500 * time.Millisecond)
8384
return nil
8485
}),
@@ -129,7 +130,7 @@ func TestJobReachTimeout(t *testing.T) {
129130
WithAddr(host+":4150"),
130131
WithTopic("timeout"),
131132
WithMaxInFlight(2),
132-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
133+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
133134
for {
134135
select {
135136
case <-ctx.Done():
@@ -166,7 +167,7 @@ func TestCancelJobAfterShutdown(t *testing.T) {
166167
WithAddr(host+":4150"),
167168
WithTopic("cancel"),
168169
WithLogger(queue.NewLogger()),
169-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
170+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
170171
for {
171172
select {
172173
case <-ctx.Done():
@@ -203,7 +204,7 @@ func TestGoroutineLeak(t *testing.T) {
203204
WithAddr(host+":4150"),
204205
WithTopic("GoroutineLeak"),
205206
WithLogger(queue.NewEmptyLogger()),
206-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
207+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
207208
for {
208209
select {
209210
case <-ctx.Done():
@@ -247,7 +248,7 @@ func TestGoroutinePanic(t *testing.T) {
247248
w := NewWorker(
248249
WithAddr(host+":4150"),
249250
WithTopic("GoroutinePanic"),
250-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
251+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
251252
panic("missing something")
252253
}),
253254
)
@@ -272,7 +273,7 @@ func TestHandleTimeout(t *testing.T) {
272273
Payload: []byte("foo"),
273274
}
274275
w := NewWorker(
275-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
276+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
276277
time.Sleep(200 * time.Millisecond)
277278
return nil
278279
}),
@@ -289,7 +290,7 @@ func TestHandleTimeout(t *testing.T) {
289290
}
290291

291292
w = NewWorker(
292-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
293+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
293294
time.Sleep(200 * time.Millisecond)
294295
return nil
295296
}),
@@ -314,7 +315,7 @@ func TestJobComplete(t *testing.T) {
314315
Payload: []byte("foo"),
315316
}
316317
w := NewWorker(
317-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
318+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
318319
return errors.New("job completed")
319320
}),
320321
WithDisableConsumer(),
@@ -330,7 +331,7 @@ func TestJobComplete(t *testing.T) {
330331
}
331332

332333
w = NewWorker(
333-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
334+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
334335
time.Sleep(200 * time.Millisecond)
335336
return errors.New("job completed")
336337
}),
@@ -355,7 +356,7 @@ func TestNSQStatsinQueue(t *testing.T) {
355356
w := NewWorker(
356357
WithAddr(host+":4150"),
357358
WithTopic("nsq_stats"),
358-
WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
359+
WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
359360
log.Println("get message")
360361
return nil
361362
}),

‎options.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"runtime"
66

77
"github.com/golang-queue/queue"
8+
"github.com/golang-queue/queue/core"
89
)
910

1011
// Option for queue system
@@ -15,7 +16,7 @@ type options struct {
1516
addr string
1617
topic string
1718
channel string
18-
runFunc func(context.Context, queue.QueuedMessage) error
19+
runFunc func(context.Context, core.QueuedMessage) error
1920
logger queue.Logger
2021
disableConsumer bool
2122
}
@@ -42,7 +43,7 @@ func WithChannel(channel string) Option {
4243
}
4344

4445
// WithRunFunc setup the run func of queue
45-
func WithRunFunc(fn func(context.Context, queue.QueuedMessage) error) Option {
46+
func WithRunFunc(fn func(context.Context, core.QueuedMessage) error) Option {
4647
return func(w *options) {
4748
w.runFunc = fn
4849
}
@@ -77,7 +78,7 @@ func newOptions(opts ...Option) options {
7778
maxInFlight: runtime.NumCPU(),
7879

7980
logger: queue.NewLogger(),
80-
runFunc: func(context.Context, queue.QueuedMessage) error {
81+
runFunc: func(context.Context, core.QueuedMessage) error {
8182
return nil
8283
},
8384
}

0 commit comments

Comments
 (0)