I have recently started using Google’s event queue “Cloud Pub/Sub” for a few projects at work, and I enjoyed how straightforward it is to configure and use in production. Their documentation is usually great, but often misses how to tie everything for your local environment.
Here is a quick guide how I set up my integration test locally using the Pub/Sub emulator. Code examples are written in Go.
Starting the emulator
To run the emulator, you first need the gcloud cli (setup instructions). Next can we install the emulator.
gcloud components install pubsub-emulator
gcloud components update
Finally, we start the emulator on localhost:8085.
gcloud beta emulators pubsub start --project=test-project
Output:
Executing: /Users/test/Downloads/google-cloud-sdk/platform/pubsub-emulator/bin/cloud-pubsub-emulator --host=localhost --port=8085
[pubsub] This is the Google Pub/Sub fake.
[pubsub] Implementation may be incomplete or differ from the real system.
....
Code
We can use the Google Pub/Sub client to publish or subscribe on a topic.
go get -u cloud.google.com/go/pubsub
Setup the client
We need to point the pubsub.Client to our emulator. For this purpose, pubsub.NewClient uses a magic environment variable PUBSUB_EMULATOR_HOST.
You can set it in a shell using $(gcloud beta emulators pubsub env-init). For our test code, we use os.Setenv.
Note: If we don’t set PUBSUB_EMULATOR_HOST, the client will default to an actual Google project. Be careful :-).
// Initialize client
os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8085")
client, err := pubsub.NewClient(ctx, "test-project")
...
#### Receive message
#### Receive message
// Create topic
topic, err := client.CreateTopic(ctx, "some-topic")
...
Receive message
Now that we have a topic, we could publish a message. However, we don’t have any subscriptions yet, and so the published message would be discared.
- Push - Pub/Sub is responsible for pushing any incoming messages on a topic to a specific endpoint in the service
- Pull - Service is responsible for setting up a connection to a Pub/Sub topic, and waits for incoming messages.
Creating a pull subscription
sub, err := client.CreateSubscription(ctx, "some-subscription-id", pubsub.SubscriptionConfig{
Topic: topic,
})
// Wrap context for cancellation, and cancel the first
cctx, cancel := context.WithCancel(ctx)
// Receive will block until the context is cancelled, or we get a non-recoverable error
err = sub.Receive(cctx, func(_ context.Context, m *pubsub.Message) {
m.Ack()
cancel() // for test purpose, shut down after first message
})
Note: If we don’t receive a message for some reason, the Receive call will be stuck forever. I will fix this in the full test code example.
Creating a push subscription
For the push example, we need a endpoint that can receive and decode the PubSub message.
func pubSubHandler(w http.ResponseWriter, r *http.Request) {
var event pubSubMessage
if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
defer r.Body.Close()
// Do something with event
// ...
}
Then all we need to do is to add pubsub.PushConfig with our endpoint, when we configure create a subscription.
_, err = client.CreateSubscription(ctx, "some-subscription-id-2", pubsub.SubscriptionConfig{
Topic: topic,
PushConfig: pubsub.PushConfig{
Endpoint: "localhost:8080/pubsub/event",
},
})
Publishing a message
Finally, we can publish a message to Pub/Sub. Keep in mind that Publish is asynchronous, and you should use res.Get afterwards if the result is important.
res := topic.Publish(ctx, &pubsub.Message{
Data: []byte("hello test"),
})
// Wait for result, if you care about it :-)
messageID, err = res.Get(ctx)
...
Summary
In this guide I showed how to do the following:
- Start the Pub/Sub emulator
- Configure the Go client
- Create topics
- Create pull and push subscriptions
- Publish and receive events
If you are interested in the full code I have included two testcases below.
Appendix: Complete test code
package main
import (
"cloud.google.com/go/pubsub"
"context"
"encoding/json"
"github.com/google/uuid"
"github.com/magnuswahlstrand/gcp-examples/pubsub/common"
"github.com/stretchr/testify/require"
"net/http"
"net/http/httptest"
"os"
"testing"
"time"
)
// Test 1: Pull subscription
func TestSendWithPull(t *testing.T) {
topicID := "topic-" + uuid.New().String()
subscriptionID := "sub-" + uuid.New().String()
ctx := context.Background()
os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8085")
client, err := pubsub.NewClient(ctx, "fake-project")
require.NoError(t, err)
topic, err := client.CreateTopic(ctx, topicID)
require.NoError(t, err)
sub, err := client.CreateSubscription(ctx, subscriptionID, pubsub.SubscriptionConfig{
Topic: topic,
})
require.NoError(t, err)
// Act: publish message
res := topic.Publish(ctx, &pubsub.Message{
Data: []byte("hello test"),
})
_, err = res.Get(ctx)
require.NoError(t, err)
cctx, cancel := context.WithCancel(ctx)
okCh := make(chan string)
go func() {
// Use a callback to receive messages via subscription.
// Receive will block until the context is cancelled, or we get a non-recoverable error
err = sub.Receive(cctx, func(_ context.Context, m *pubsub.Message) {
m.Ack()
okCh <- string(m.Data)
cancel()
})
}()
select {
case msg := <-okCh:
require.Equal(t, "hello test", msg)
case <-time.After(300 * time.Millisecond):
require.Fail(t, "did not receive message within deadline")
cancel()
}
}
// Test 2: Push subscription
func TestSendWithPush(t *testing.T) {
topicID := "topic-" + uuid.New().String()[0:8]
subscriptionID := "sub-" + uuid.New().String()[0:8]
okCh := make(chan string)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var event pubSubMessage
if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
defer r.Body.Close()
okCh <- string(event.Message.Data)
}))
ctx := context.Background()
os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8085")
client, err := pubsub.NewClient(ctx, "fake-project")
require.NoError(t, err)
topic, err := client.CreateTopic(ctx, topicID)
require.NoError(t, err)
_, err = client.CreateSubscription(ctx, subscriptionID, pubsub.SubscriptionConfig{
Topic: topic,
PushConfig: pubsub.PushConfig{
Endpoint: server.URL,
},
})
require.NoError(t, err)
// Act: publish message
_ = topic.Publish(ctx, &pubsub.Message{
Data: []byte("hello test"),
})
select {
case msg := <-okCh:
require.Equal(t, "hello test", msg)
case <-time.After(500 * time.Millisecond):
require.Fail(t, "did not receive message within deadline")
}
}