docs: Add producer and consumer example

Signed-off-by: Bo-Yi Wu <[email protected]>

Author: appleboy

_example/producer-consumer/README.md

@@ -0,0 +1,29 @@
+# Example with server and client
+
+Please refer the following steps to build server and client.
+
+## Build server
+
+```sh
+go build -o app server/main.go
+```
+
+## Build client
+
+```sh
+go build -o agent client/main.go
+```
+
+## Usage
+
+Run the multiple agent. (open two console in the same terminal)
+
+```sh
+./agent
+```
+
+Publish the message.
+
+```sh
+./app
+```

_example/producer-consumer/consumer/main.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/appleboy/graceful"
+	"github.com/golang-queue/queue"
+	"github.com/golang-queue/queue/core"
+	"github.com/golang-queue/rabbitmq"
+)
+
+type job struct {
+	Message string
+}
+
+func (j *job) Bytes() []byte {
+	b, err := json.Marshal(j)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
+func main() {
+	taskN := 10000
+	rets := make(chan string, taskN)
+
+	m := graceful.NewManager()
+
+	// define the worker
+	w := rabbitmq.NewWorker(
+		rabbitmq.WithSubj("sample_2"),
+		rabbitmq.WithExchangeType("fanout"),
+		rabbitmq.WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+			var v *job
+			if err := json.Unmarshal(m.Bytes(), &v); err != nil {
+				return err
+			}
+			rets <- v.Message
+			time.Sleep(500 * time.Millisecond)
+			return nil
+		}),
+	)
+	// define the queue
+	q := queue.NewPool(
+		2,
+		queue.WithWorker(w),
+	)
+
+	m.AddRunningJob(func(ctx context.Context) error {
+		for {
+			select {
+			case <-ctx.Done():
+				select {
+				case m := <-rets:
+					fmt.Println("message:", m)
+				default:
+				}
+				return nil
+			case m := <-rets:
+				fmt.Println("message:", m)
+				time.Sleep(50 * time.Millisecond)
+			}
+		}
+	})
+
+	m.AddShutdownJob(func() error {
+		// shutdown the service and notify all the worker
+		q.Release()
+		return nil
+	})
+
+	<-m.Done()
+}

_example/producer-consumer/go.mod

@@ -0,0 +1,16 @@
+module example
+
+go 1.18
+
+require (
+	github.com/appleboy/graceful v0.0.4
+	github.com/golang-queue/queue v0.1.3
+	github.com/golang-queue/rabbitmq v0.0.3-0.20210907015837-3e2e4b448b3d
+)
+
+require (
+	github.com/goccy/go-json v0.9.7 // indirect
+	github.com/rabbitmq/amqp091-go v1.3.4 // indirect
+)
+
+replace github.com/golang-queue/rabbitmq => ../../

_example/producer-consumer/go.sum

@@ -0,0 +1,14 @@
+github.com/appleboy/graceful v0.0.4 h1:Q4LCeq4DFy59qiACLtuH+mSqDERtUzwkQbCWpRaWwvQ=
+github.com/appleboy/graceful v0.0.4/go.mod h1:Q2mVx0t+N0lCDZc5MJudbcpTm6cgGM/J2gZCZIqD9dc=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/goccy/go-json v0.9.7 h1:IcB+Aqpx/iMHu5Yooh7jEzJk1JZ7Pjtmys2ukPr7EeM=
+github.com/goccy/go-json v0.9.7/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
+github.com/golang-queue/queue v0.1.3 h1:FGIrn8e0fN8EmL3glP0rFEcYVtWUGMEeqX4h4nnzh40=
+github.com/golang-queue/queue v0.1.3/go.mod h1:h/PhaoMwT5Jc4sQNus7APgWBUItm6QC9k6JtmwrsRos=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/rabbitmq/amqp091-go v1.3.4 h1:tXuIslN1nhDqs2t6Jrz3BAoqvt4qIZzxvdbdcxWtHYU=
+github.com/rabbitmq/amqp091-go v1.3.4/go.mod h1:ogQDLSOACsLPsIq0NpbtiifNZi2YOz0VTJ0kHRghqbM=
+github.com/stretchr/testify v1.7.3 h1:dAm0YRdRQlWojc3CrCRgPBzG5f941d0zvAKu7qY4e+I=
+go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

_example/producer-consumer/producer/main.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/golang-queue/queue"
+	"github.com/golang-queue/rabbitmq"
+)
+
+type job struct {
+	Message string
+}
+
+func (j *job) Bytes() []byte {
+	b, err := json.Marshal(j)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
+func main() {
+	taskN := 100
+
+	// define the worker
+	w := rabbitmq.NewWorker(
+		rabbitmq.WithSubj("sample_2"),
+		rabbitmq.WithExchangeType("fanout"),
+	)
+
+	// define the queue
+	q := queue.NewPool(
+		0,
+		queue.WithWorker(w),
+	)
+
+	// assign tasks in queue
+	for i := 0; i < taskN; i++ {
+		if err := q.Queue(&job{
+			Message: fmt.Sprintf("handle the job: %d", i+1),
+		}); err != nil {
+			log.Fatal(err)
+		}
+	}
+
+	time.Sleep(1 * time.Second)
+	// shutdown the service and notify all the worker
+	q.Release()
+}