Merge pull request #51 from siad007/exchangeType

Added amqp exchange-type option.

Author: rustatian

broker/amqp/broker_test.go

@@ -21,6 +21,21 @@ var (
 	}
 )
 
+var (
+	fanoutPipe = &jobs.Pipeline{
+		"broker":   "amqp",
+		"name":     "fanout",
+		"queue":    "fanout-queue",
+		"exchange": "fanout-exchange",
+		"exchange-type": "fanout",
+		"prefetch": 1,
+	}
+
+	fanoutCfg = &Config{
+		Addr: "amqp://guest:guest@localhost:5672/",
+	}
+)
+
 func TestBroker_Init(t *testing.T) {
 	b := &Broker{}
 	ok, err := b.Init(cfg)
@@ -323,3 +338,82 @@ func TestBroker_Consume_With_RoutingKey(t *testing.T) {
 
 	<-waitJob
 }
+
+func TestBroker_Queue_ExchangeType(t *testing.T) {
+	pipeWithKey := pipe.With("exchange-type", "direct")
+
+	assert.Equal(t, pipeWithKey.String("exchange-type", ""), "direct")
+}
+
+func TestBroker_Register_With_ExchangeType(t *testing.T) {
+	b := &Broker{}
+	_, err := b.Init(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pipeWithKey := pipe.With("exchange-type", "fanout")
+
+	assert.NoError(t, b.Register(&pipeWithKey))
+}
+
+func TestBroker_Register_With_WrongExchangeType(t *testing.T) {
+	b := &Broker{}
+	_, err := b.Init(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pipeWithKey := pipe.With("exchange-type", "xxx")
+
+	assert.Error(t, b.Register(&pipeWithKey))
+}
+
+func TestBroker_Consume_With_ExchangeType(t *testing.T) {
+	b := &Broker{}
+	_, err := b.Init(fanoutCfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pipeWithKey := fanoutPipe.With("exchange-type", "fanout")
+
+	err = b.Register(&pipeWithKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ready := make(chan interface{})
+	b.Listen(func(event int, ctx interface{}) {
+		if event == jobs.EventBrokerReady {
+			close(ready)
+		}
+	})
+
+	exec := make(chan jobs.Handler, 1)
+	assert.NoError(t, b.Consume(&pipeWithKey, exec, func(id string, j *jobs.Job, err error) {}))
+
+	go func() { assert.NoError(t, b.Serve()) }()
+	defer b.Stop()
+
+	<-ready
+
+	jid, perr := b.Push(&pipeWithKey, &jobs.Job{
+		Job:     "test",
+		Payload: "body",
+		Options: &jobs.Options{},
+	})
+
+	assert.NotEqual(t, "", jid)
+	assert.NoError(t, perr)
+
+	waitJob := make(chan interface{})
+	exec <- func(id string, j *jobs.Job) error {
+		assert.Equal(t, jid, id)
+		assert.Equal(t, "body", j.Payload)
+		close(waitJob)
+		return nil
+	}
+
+	<-waitJob
+}

broker/amqp/queue.go

@@ -1,6 +1,7 @@
 package amqp
 
 import (
+	"errors"
 	"fmt"
 	"github.com/spiral/jobs/v2"
 	"github.com/streadway/amqp"
@@ -10,11 +11,40 @@ import (
 	"time"
 )
 
+type ExchangeType string
+
+const (
+	Direct  ExchangeType = "direct"
+	Fanout  ExchangeType = "fanout"
+	Topic   ExchangeType = "topic"
+	Headers ExchangeType = "headers"
+)
+
+func (et ExchangeType) IsValid() error {
+	switch et {
+	case Direct, Fanout, Topic, Headers:
+		return nil
+	}
+	return errors.New("unknown exchange-type")
+}
+
+func (et ExchangeType) String() string {
+	switch et {
+	case Direct, Fanout, Topic, Headers:
+		return string(et)
+	default:
+		return "direct"
+	}
+}
+
+
 type queue struct {
-	active              int32
-	pipe                *jobs.Pipeline
-	exchange, name, key string
-	consumer            string
+	active                 int32
+	pipe                   *jobs.Pipeline
+	exchange               string
+	exchangeType           ExchangeType
+	name, key              string
+	consumer               string
 
 	// active consuming channel
 	muc sync.Mutex
@@ -39,8 +69,16 @@ func newQueue(pipe *jobs.Pipeline, lsn func(event int, ctx interface{})) (*queue
 		return nil, fmt.Errorf("missing `queue` parameter on amqp pipeline")
 	}
 
+	exchangeType := ExchangeType(pipe.String("exchange-type", "direct"))
+
+	err := exchangeType.IsValid()
+	if err != nil {
+		return nil, fmt.Errorf(err.Error())
+	}
+
 	return &queue{
 		exchange: pipe.String("exchange", "amqp.direct"),
+		exchangeType: exchangeType,
 		name:     pipe.String("queue", ""),
 		key:      pipe.String("routing-key", pipe.String("queue", "")),
 		consumer: pipe.String("consumer", fmt.Sprintf("rr-jobs:%s-%v", pipe.Name(), os.Getpid())),
@@ -221,7 +259,7 @@ func (q *queue) declare(cp *chanPool, queue string, key string, args amqp.Table)
 		return err
 	}
 
-	err = c.ch.ExchangeDeclare(q.exchange, "direct", true, false, false, false, nil)
+	err = c.ch.ExchangeDeclare(q.exchange, q.exchangeType.String(), true, false, false, false, nil)
 	if err != nil {
 		return cp.closeChan(c, err)
 	}