feat: support durable queues (#47)

Author: w-h-a

cmd/engine.go

@@ -143,6 +143,7 @@ func initBroker() broker.Broker {
 	case string(broker.Rabbit):
 		return rabbit.NewBroker(
 			broker.WithLocation(config.BrokerLocation()),
+			broker.WithDurable(config.BrokerDurable()),
 		)
 	default:
 		return memorybroker.NewBroker()

internal/engine/clients/broker/options.go

@@ -6,6 +6,7 @@ type Option func(o *Options)
 
 type Options struct {
 	Location string
+	Durable  bool
 	Context  context.Context
 }
 
@@ -15,6 +16,12 @@ func WithLocation(location string) Option {
 	}
 }
 
+func WithDurable(durable bool) Option {
+	return func(o *Options) {
+		o.Durable = durable
+	}
+}
+
 func NewOptions(opts ...Option) Options {
 	options := Options{
 		Context: context.Background(),
@@ -27,21 +34,21 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
-type PublishOption func(o *PublishOptions)
+type SubscribeOption func(o *SubscribeOptions)
 
-type PublishOptions struct {
+type SubscribeOptions struct {
 	Queue   string
 	Context context.Context
 }
 
-func PublishWithQueue(queue string) PublishOption {
-	return func(o *PublishOptions) {
+func SubscribeWithQueue(queue string) SubscribeOption {
+	return func(o *SubscribeOptions) {
 		o.Queue = queue
 	}
 }
 
-func NewPublishOptions(opts ...PublishOption) PublishOptions {
-	options := PublishOptions{
+func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
+	options := SubscribeOptions{
 		Context: context.Background(),
 	}
 
@@ -52,21 +59,21 @@ func NewPublishOptions(opts ...PublishOption) PublishOptions {
 	return options
 }
 
-type SubscribeOption func(o *SubscribeOptions)
+type PublishOption func(o *PublishOptions)
 
-type SubscribeOptions struct {
+type PublishOptions struct {
 	Queue   string
 	Context context.Context
 }
 
-func SubscribeWithQueue(queue string) SubscribeOption {
-	return func(o *SubscribeOptions) {
+func PublishWithQueue(queue string) PublishOption {
+	return func(o *PublishOptions) {
 		o.Queue = queue
 	}
 }
 
-func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
-	options := SubscribeOptions{
+func NewPublishOptions(opts ...PublishOption) PublishOptions {
+	options := PublishOptions{
 		Context: context.Background(),
 	}
 

internal/engine/clients/broker/rabbit/broker.go

@@ -51,7 +51,7 @@ func (b *rabbitBroker) Subscribe(ctx context.Context, callback func(ctx context.
 
 			if _, err := rbch.QueueDeclare(
 				options.Queue,
-				false, // durable
+				b.options.Durable,
 				false, // delete when unused
 				false, // exclusive
 				false, // no-wait,
@@ -145,7 +145,7 @@ func (b *rabbitBroker) Publish(ctx context.Context, data []byte, opts ...broker.
 
 	if _, err := rbch.QueueDeclare(
 		options.Queue,
-		false, // durable
+		b.options.Durable,
 		false, // delete when unused
 		false, // exclusive
 		false, // no-wait,
@@ -154,16 +154,22 @@ func (b *rabbitBroker) Publish(ctx context.Context, data []byte, opts ...broker.
 		return err
 	}
 
+	publishing := amqp.Publishing{
+		ContentType: "application/json",
+		Body:        data,
+	}
+
+	if b.options.Durable {
+		publishing.DeliveryMode = amqp.Persistent
+	}
+
 	if err := rbch.PublishWithContext(
 		ctx,
 		"",            // exchange
 		options.Queue, // routing key
 		false,         // mandatory
 		false,         // immediate
-		amqp.Publishing{
-			ContentType: "application/json",
-			Body:        data,
-		},
+		publishing,
 	); err != nil {
 		return broker.ErrPublishing
 	}

internal/engine/config/config.go

@@ -28,6 +28,7 @@ type config struct {
 	workerQueues       map[string]int
 	broker             string
 	brokerLocation     string
+	brokerDurable      bool
 	runner             string
 	runnerHost         string
 	runnerRegistryUser string
@@ -48,6 +49,7 @@ func New() {
 			workerQueues:       map[string]int{string(task.Scheduled): 1, string(task.Cancelled): 1},
 			broker:             "memory",
 			brokerLocation:     "",
+			brokerDurable:      false,
 			runner:             "docker",
 			runnerHost:         "unix:///var/run/docker.sock",
 			runnerRegistryUser: "",
@@ -123,6 +125,11 @@ func New() {
 			instance.brokerLocation = brokerLocation
 		}
 
+		brokerDurable := os.Getenv("BROKER_DURABLE")
+		if brokerDurable == "true" {
+			instance.brokerDurable = true
+		}
+
 		r := os.Getenv("RUNNER")
 		if len(r) > 0 {
 			if _, ok := runner.RuntimeTypes[r]; ok {
@@ -239,6 +246,14 @@ func BrokerLocation() string {
 	return instance.brokerLocation
 }
 
+func BrokerDurable() bool {
+	if instance == nil {
+		panic("cfg is nil")
+	}
+
+	return instance.brokerDurable
+}
+
 func Runner() string {
 	if instance == nil {
 		panic("cfg is nil")