pubsub_stuck.go

package main

import (
	"context"
	"log"
	"time"

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

func main() {
	srv := pstest.NewServer()
	client := fakePubSubClient(srv)
	defer client.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	s, err := client.CreateSubscription(ctx, "test", pubsub.SubscriptionConfig{
		Topic: client.Topic("test"),
	})
	if err != nil {
		panic(err)
	}

	go publishLots(srv)

	err = s.Receive(ctx, func(_ context.Context, msg *pubsub.Message) {
		log.Println("receive msg")
		cancel()
		time.AfterFunc(time.Second, func() {
			msg.Ack()
			log.Println("acked msg")
		})
	})
	switch err {
	case context.Canceled, nil:
	default:
		panic(err)
	}
}

func publishLots(srv *pstest.Server) {
	for i := 0; i < 1000; i++ {
		data := make([]byte, 300)
		srv.Publish("projects/project/topics/test", data, nil)
	}
}

func fakePubSubClient(srv *pstest.Server) *pubsub.Client {
	ctx := context.Background()
	// Connect to the server without using TLS.
	conn, err := grpc.Dial(srv.Addr, grpc.WithInsecure())
	if err != nil {
		panic(err)
	}

	// Use the connection when creating a pubsub client.
	client, err := pubsub.NewClient(ctx, "project", option.WithGRPCConn(conn))
	if err != nil {
		panic(err)
	}

	// Initialize a topic to publish on
	if _, err = client.CreateTopic(context.Background(), "test"); err != nil {
		panic(err)
	}

	return client
}