Init commit

Author: memclutter

.github/workflows/go.yml

@@ -0,0 +1,30 @@
+name: Go
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+
+  test:
+    strategy:
+      matrix:
+        go_version: [ '1.14', '1.15', '1.16', '1.17' ]
+        os: [ 'ubuntu-latest', 'windows-latest', 'macOS-latest' ]
+    runs-on: ${{ matrix.os }}
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Use GO ${{ matrix.go_version }}
+      uses: actions/setup-go@v2
+      with:
+        go-version: ${{ matrix.go_version }}
+
+    - name: Test GO ${{ matrix.go_version }}
+      run: go test ./... -race -coverprofile=coverage.txt -covermode=atomic
+
+    - name: Upload to codecov
+      if: ${{ matrix.os == 'ubuntu-latest' && matrix.go_version == '1.17' }}
+      run: bash <(curl -s https://codecov.io/bash)

.gitignore

@@ -0,0 +1,26 @@
+
+# Created by https://www.toptal.com/developers/gitignore/api/go
+# Edit at https://www.toptal.com/developers/gitignore?templates=go
+
+### Go ###
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+### Go Patch ###
+/vendor/
+/Godeps/
+
+# End of https://www.toptal.com/developers/gitignore/api/go

.idea/.gitignore

@@ -0,0 +1,7 @@
+# goqueue
+
+[![Go](https://github.com/memclutter/goqueue/actions/workflows/go.yml/badge.svg)](https://github.com/memclutter/goqueue/actions/workflows/go.yml)
+[![codecov](https://codecov.io/gh/memclutter/goqueue/branch/main/graph/badge.svg?token=57IA9OCZFD)](https://codecov.io/gh/memclutter/goqueue)
+
+Simple rabbitmq queue wrapper for golang.
+

.idea/goqueue.iml

@@ -0,0 +1,8 @@
+package goqueue
+
+import (
+	log "github.com/sirupsen/logrus"
+	"github.com/streadway/amqp"
+)
+
+type Callback func(amqp.Delivery, *log.Entry) (Retry, error)

.idea/modules.xml

@@ -0,0 +1,110 @@
+package goqueue
+
+import (
+	"fmt"
+	log "github.com/sirupsen/logrus"
+	"github.com/streadway/amqp"
+	"time"
+)
+
+type Consumer interface {
+	Consume() error
+}
+
+func NewConsumer(queue string, retryIntervals []time.Duration, callback Callback, amqpConn *amqp.Connection, consumerLog *log.Entry) Consumer {
+	if consumerLog == nil {
+		consumerLog = log.WithFields(log.Fields{"_default": true})
+	}
+	consumerLog = consumerLog.WithFields(log.Fields{"queue": queue})
+
+	return &DefaultConsumer{
+		queue:          queue,
+		retryIntervals: retryIntervals,
+		callback:       callback,
+		amqpConn:       amqpConn,
+		log:            consumerLog,
+	}
+}
+
+type DefaultConsumer struct {
+	queue          string
+	retryIntervals []time.Duration
+	callback       Callback
+	amqpConn       *amqp.Connection
+	amqpCh         *amqp.Channel
+	log            *log.Entry
+}
+
+func (c DefaultConsumer) Consume() error {
+
+	var err error
+
+	// Open channel in rabbitmq
+	c.amqpCh, err = c.amqpConn.Channel()
+	if err != nil {
+		return fmt.Errorf("open channel error: %v", err)
+	}
+	defer func() {
+		if err := c.amqpCh.Close(); err != nil {
+			c.log.Errorf("close channel error (ignore): %v", err)
+		}
+	}()
+
+	// Declare exchanges, queues, set qos etc
+	if err = c.declare(); err != nil {
+		return fmt.Errorf("declare error: %v", err)
+	}
+
+	// Create co channel for consume deliveries
+	deliveries, err := c.amqpCh.Consume(c.queue, "", false, false, false, false, nil)
+	if err != nil {
+		return fmt.Errorf("error create consume channel: %v", err)
+	}
+
+	// Process deliveries
+	for delivery := range deliveries {
+
+		// How more this delivery retry
+		retry := getRetry(delivery.Headers)
+
+		// Context logger for delivery
+		deliveryLog := c.log.WithFields(log.Fields{
+			"deliveryTag": delivery.DeliveryTag,
+			"retry":       retry,
+		})
+		deliveryLog.Infof("receive delivery")
+		deliveryLog.Debugf("body: %s", delivery.Body)
+
+		// Process with retries logic
+		c.ProcessDelivery(delivery, deliveryLog, retry)
+	}
+
+	return nil
+}
+
+func (c DefaultConsumer) ProcessDelivery(delivery amqp.Delivery, deliveryLog *log.Entry, retry int) {
+	defer func() {
+		if err := delivery.Ack(false); err != nil {
+			deliveryLog.Errorf("ack error (ignored): %v", err)
+		}
+	}()
+
+	// Callback for user code to process delivery. If error, check retryStatus
+	// - RetryNext - redelivery this delivery for retry after some intervals
+	// - RetryStop - ignore
+	retryStatus, err := c.callback(delivery, deliveryLog)
+	if err != nil {
+		deliveryLog.Warnf("callback error: %v", err)
+		switch retryStatus {
+		case RetryNext:
+			deliveryLog.Infof("retry next")
+			if err := Publish(c.amqpCh, c.queue, delivery.Body, retry); err != nil {
+				deliveryLog.Errorf("error retry: %v", err)
+			}
+		case RetryStop:
+			deliveryLog.Infof("retry stop")
+		default:
+			deliveryLog.Warnf("retry ignore")
+		}
+	}
+}