Dead Letter Topics - DeliveryAttempts

main.go

package main

// gucci main

main_test.go

package main

import (
	"fmt"
	"strconv"
	"sync"
	"testing"
	"time"

	"golang.org/x/net/context"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsub/pstest"
	"google.golang.org/api/option"
	"google.golang.org/grpc"
)

func TestPSTest(t *testing.T) {
	ctx := context.Background()
	srv := pstest.NewServer()

	conn, err := grpc.Dial(srv.Addr, grpc.WithInsecure())
	if err != nil {
		panic(err)
	}

	client, err := pubsub.NewClient(ctx, "some-project", option.WithGRPCConn(conn))
	if err != nil {
		panic(err)
	}
	defer client.Close()

	topicName := "test-topic"
	deadLetterTopicName := fmt.Sprintf("%s-%s", topicName, "dlq")
	deadLetterTopicFQDN := fmt.Sprintf("projects/some-project/topics/%s", deadLetterTopicName)
	topic, err := client.CreateTopic(ctx, topicName)
	if err != nil {
		panic(err)
	}

	_, err = client.CreateTopic(ctx, deadLetterTopicName)
	if err != nil {
		panic(err)
	}

	ackDeadline := 20 * time.Second
	deadLetterPolicy := &pubsub.DeadLetterPolicy{
		DeadLetterTopic:     deadLetterTopicFQDN,
		MaxDeliveryAttempts: 5,
	}

	sub, err := client.CreateSubscription(
		ctx,
		"sub-name",
		pubsub.SubscriptionConfig{
			Topic:            topic,
			AckDeadline:      ackDeadline,
			DeadLetterPolicy: deadLetterPolicy,
		},
	)

	if err != nil {
		panic(err)
	}

	go func() {
		for i := 0; i < 10; i++ {
			srv.Publish("projects/some-project/topics/test-topic", []byte(strconv.Itoa(i)), nil)
		}
	}()

	ctx, cancel := context.WithCancel(ctx)
	var mu sync.Mutex
	count := 0
	err = sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
		mu.Lock()
		defer mu.Unlock()
		count++
		if count >= 100 {
			cancel()
		}
		fmt.Printf("%v count:%v\n", msg.ID, count)
		fmt.Printf("%v delivery attempts: %v\n", msg.ID, msg.DeliveryAttempt)
		msg.Nack()
	})
	if err != nil {
		panic(err)
	}
}

output

[] burdz@~/temp/pubsub: ./run.sh
=== RUN   TestPSTest
m0 count:1
m0 delivery attempts: <nil>
m2 count:2
m2 delivery attempts: <nil>
m1 count:3
m1 delivery attempts: <nil>
m7 count:4
m7 delivery attempts: <nil>
m5 count:5
m5 delivery attempts: <nil>
m6 count:6
m6 delivery attempts: <nil>
m4 count:7
m4 delivery attempts: <nil>
m3 count:8
m3 delivery attempts: <nil>
m9 count:9
m9 delivery attempts: <nil>
m8 count:10
m8 delivery attempts: <nil>
m2 count:11
m2 delivery attempts: <nil>
m8 count:12
m8 delivery attempts: <nil>
m6 count:13
m6 delivery attempts: <nil>
m4 count:14
m4 delivery attempts: <nil>
m5 count:15
m5 delivery attempts: <nil>
m3 count:16
m3 delivery attempts: <nil>
m9 count:17
m9 delivery attempts: <nil>
m1 count:18
m1 delivery attempts: <nil>
m0 count:19
m0 delivery attempts: <nil>
m7 count:20
m7 delivery attempts: <nil>
m2 count:21
m2 delivery attempts: <nil>
m3 count:22
m3 delivery attempts: <nil>
m8 count:23
m8 delivery attempts: <nil>
m4 count:24
m4 delivery attempts: <nil>
m1 count:25
m1 delivery attempts: <nil>
m0 count:26
m0 delivery attempts: <nil>
m9 count:27
m9 delivery attempts: <nil>
m5 count:28
m5 delivery attempts: <nil>
m6 count:29
m6 delivery attempts: <nil>
m7 count:30
m7 delivery attempts: <nil>
m7 count:31
m7 delivery attempts: <nil>
m6 count:32
m6 delivery attempts: <nil>
m9 count:33
m9 delivery attempts: <nil>
m3 count:34
m3 delivery attempts: <nil>
m4 count:35
m4 delivery attempts: <nil>
m1 count:36
m1 delivery attempts: <nil>
m8 count:37
m8 delivery attempts: <nil>
m2 count:38
m2 delivery attempts: <nil>
m5 count:39
m5 delivery attempts: <nil>
m0 count:40
m0 delivery attempts: <nil>
m3 count:41
m3 delivery attempts: <nil>
m5 count:42
m5 delivery attempts: <nil>
m0 count:43
m0 delivery attempts: <nil>
m6 count:44
m6 delivery attempts: <nil>
m2 count:45
m2 delivery attempts: <nil>
m7 count:46
m7 delivery attempts: <nil>
m9 count:47
m9 delivery attempts: <nil>
m8 count:48
m8 delivery attempts: <nil>
m1 count:49
m1 delivery attempts: <nil>
m4 count:50
m4 delivery attempts: <nil>
m1 count:51
m1 delivery attempts: <nil>
m7 count:52
m7 delivery attempts: <nil>
m3 count:53
m3 delivery attempts: <nil>
m8 count:54
m8 delivery attempts: <nil>
m0 count:55
m0 delivery attempts: <nil>
m2 count:56
m2 delivery attempts: <nil>
m5 count:57
m5 delivery attempts: <nil>
m6 count:58
m6 delivery attempts: <nil>
m4 count:59
m4 delivery attempts: <nil>
m9 count:60
m9 delivery attempts: <nil>
m9 count:61
m9 delivery attempts: <nil>
m7 count:62
m7 delivery attempts: <nil>
m4 count:63
m4 delivery attempts: <nil>
m0 count:64
m0 delivery attempts: <nil>
m3 count:65
m3 delivery attempts: <nil>
m1 count:66
m1 delivery attempts: <nil>
m6 count:67
m6 delivery attempts: <nil>
m5 count:68
m5 delivery attempts: <nil>
m8 count:69
m8 delivery attempts: <nil>
m2 count:70
m2 delivery attempts: <nil>
m5 count:71
m5 delivery attempts: <nil>
m3 count:72
m3 delivery attempts: <nil>
m8 count:73
m8 delivery attempts: <nil>
m0 count:74
m0 delivery attempts: <nil>
m4 count:75
m4 delivery attempts: <nil>
m9 count:76
m9 delivery attempts: <nil>
m6 count:77
m6 delivery attempts: <nil>
m1 count:78
m1 delivery attempts: <nil>
m2 count:79
m2 delivery attempts: <nil>
m7 count:80
m7 delivery attempts: <nil>
m4 count:81
m4 delivery attempts: <nil>
m9 count:82
m9 delivery attempts: <nil>
m2 count:83
m2 delivery attempts: <nil>
m3 count:84
m3 delivery attempts: <nil>
m8 count:85
m8 delivery attempts: <nil>
m1 count:86
m1 delivery attempts: <nil>
m5 count:87
m5 delivery attempts: <nil>
m0 count:88
m0 delivery attempts: <nil>
m7 count:89
m7 delivery attempts: <nil>
m6 count:90
m6 delivery attempts: <nil>
m0 count:91
m0 delivery attempts: <nil>
m9 count:92
m9 delivery attempts: <nil>
m4 count:93
m4 delivery attempts: <nil>
m2 count:94
m2 delivery attempts: <nil>
m6 count:95
m6 delivery attempts: <nil>
m7 count:96
m7 delivery attempts: <nil>
m3 count:97
m3 delivery attempts: <nil>
m8 count:98
m8 delivery attempts: <nil>
m1 count:99
m1 delivery attempts: <nil>
m5 count:100
m5 delivery attempts: <nil>
--- PASS: TestPSTest (1.86s)
PASS
ok  	test.com/test	1.866s

run.sh

#!/usr/bin/env bash

go test ./...