pubsub_go

README

pubsub-go

pubsub-go is a Go module that makes it easy to publish and receive messages from GCP's Pub/Sub service.

Usage

Create a new PubSub

import pubsub_go "github.com/clearchanneloutdoor/pubsub-go"

func main() {
    ctx := context.Background()
    config := pubsub_go.Config{
        ProjectID:              "projectID",
        IsLocal:                true,
        ServiceAccountFilePath: "./path/to/settings.json",
    }

    ps, err := pubsub_go.NewClient(ctx, config)
    if err != nil {
        // Handle error
    }
}

Create a Topic

import "cloud.google.com/go/pubsub"

func main() {
    // Initialize new pubsub-go PubSub

    // define a configs for topic creation
    cfg := pubsub_go.TopicConfig{
		Settings: pubsub.TopicConfig{
			RetentionDuration: time.Hour * 24 * time.Duration(3),
		},
	}
    if err := ps.CreateTopic("topic", cfg); err != nil {
        // Handle error
    }

    // send a message to the topic}

Create Multiple Subscriptions with Filters for a Topic

import "cloud.google.com/go/pubsub"

func main() {
    // Initialize new pubsub-go PubSub

    // Specify the name of the topic for those subscriptions we are about to create
    tid := "topic"
    // Define a map for a list of Subscription Names and their (optional) filter definitions
    subs := map[string]string{
        "topic-sub": "",
        "topic-sub-ca": "attributes.region = \"CA\"",
        "topic-sub-nv": "attributes.region = \"NV\"",
        "topic-sub-tx": "attributes.region = \"TX\"",
    }
    cfg := pubsub_go.SubscriptionConfig{
		Settings: pubsub.SubscriptionConfig{
			EnableMessageOrdering: true,
			RetainAckedMessages:   false,
		},
	}
    // Create subscriptions
    if err := ps.CreateSubscriptions(tid, subs, cfg); err != nil {
        // handle error
    }

}

Publish Message

func main() {
    // Initialize new pubsub-go PubSub

    message := pubsub_go.Message{
        Attributes: nil,
        Message:    "Hello World",
        Topic:      "topic",
    }

    if err := ps.Publish(message); err != nil {
        // Handle error
    }
}

Receive Messages

func main() {
    // Initialize new pubsub-go PubSub

    messages := make(chan *pubsub.Message)

    go func() {
        for {
            msg := <-messages

            // Process message
            fmt.Printf("Got message: %s\n", string(msg.Data))

            // Acknowledge message
            msg.Ack()
        }
    }()

    if err := ps.Receive("subscription", messages); err != nil {
        // Handle error
    }
}

Receive Messages with ReceiveSettings

import (
	"cloud.google.com/go/pubsub"
)

func main() {
    // Initialize new pubsub-go PubSub

    messages := make(chan *pubsub.Message)

    go func() {
        for {
            msg := <-messages

            // Process message
            fmt.Printf("Got message: %s\n", string(msg.Data))

            // Acknowledge message
            msg.Ack()
        }
    }()

    rs := ps.ReceiveSettings{
		Settings: pubsub.ReceiveSettings{
			MaxExtension: (15 * time.Second),
			MaxOutstandingMessages: 1000,
			NumGoroutines:          10,
		},
	}

    if err := ps.ReceiveWithSettings("subscription", rs, messages); err != nil {
        // Handle error
    }
}

Running GCP PubSub Locally

Google publishes an emulator for GCP PubSub, so you can run it locally. This repo includes a script that will spin up a docker container with the emulator started so running a local dev environment is easier.

Huge shout out to @anguillanneuf, who wrote a blog post that made building this script much easier.

Dependencies

• Docker
• Openssl

Start

Open a terminal and navigate to this project's directory. once there run the following command...

./local-pubsub.sh

This will take you through a wizard to get all information necessary to start a Project, Topic, and Subscription.

There is an export command that is out put once the script has completed that you'll need to copy and paste across all your open terminal windows. The reason being is that there is no way to set the GCP PubSub endpoint directly in your application; the GCP libray looks to an environment to know which endpoint to use.

The -m Option

Sending a message directly to your queue using something like curl isn't clearly documented. Rather than sending a json payload, you send a base64 encoded string of the message data.

Execute the following command to go throw a wizard that will output a properly formatted request for you.

./local-pubsub.sh -m

Documentation

Overview

Package pubsub_go TODO: Description

Index

• type Config
• type Message
• type PubSub
  • func NewPubSub(c Config) (*PubSub, error)
  • func (ps *PubSub) CreateSubscriptions(tid string, sids map[string]string, cfg *SubscriptionConfig) error
  • func (ps *PubSub) CreateTopic(tid string, cfg *TopicConfig) error
  • func (ps *PubSub) Publish(m Message) error
  • func (ps *PubSub) Receive(subscription string, messages chan<- *pubsub.Message) error
• type PublishSettings
• type ReceiveSettings
• type SubscriptionConfig
• type TopicConfig

Types

type Config

type Config struct {
	IsLocal                bool
	ProjectID              string
	PublishSettings        PublishSettings
	ReceiveSettings        ReceiveSettings
	ServiceAccountFilePath string
}

Config provides the information needed to securely connect to Google Cloud's PubSub and to configure any publishing and subscription options.

type Message

type Message struct {
	Attributes map[string]string
	Message    interface{}
	Topic      string
}

Message holds the data needed for publishing a message to PubSub.

type PubSub

type PubSub struct {
	// contains filtered or unexported fields
}

PubSub provides a wrapper client for Google Cloud's PubSub for publishing messages to a topic and receiving messages from a subscription.

func NewPubSub

func NewPubSub(c Config) (*PubSub, error)

NewPubSub creates a new PubSub client with the provided Config.

func (*PubSub) CreateSubscriptions

func (ps *PubSub) CreateSubscriptions(tid string, sids map[string]string, cfg *SubscriptionConfig) error

Create Subscriptions for a Topic based on a map of Subscription Name and Filter

func (*PubSub) CreateTopic

func (ps *PubSub) CreateTopic(tid string, cfg *TopicConfig) error

Create a Topic in Google PubSub if not exist

func (*PubSub) Publish

func (ps *PubSub) Publish(m Message) error

Publish sends a message to a topic along with any attributes that were provided.

func (*PubSub) Receive

func (ps *PubSub) Receive(subscription string, messages chan<- *pubsub.Message) error

Receive subscribes to a topic via the subscription id and passes messages back to the caller through the channel.

type PublishSettings

type PublishSettings struct {
	Settings pubsub.PublishSettings
}

PublishSettings is an extension of Google PubSub's PublishSettings that enables further configuration for publishing messages to a topic.

type ReceiveSettings

type ReceiveSettings struct {
	Settings pubsub.ReceiveSettings
}

ReceiveSettings is an extension of Google PubSub's ReceiveSettings that enables further configuration for receiving messages from a subscription.

type SubscriptionConfig

type SubscriptionConfig struct {
	Settings pubsub.SubscriptionConfig
}

SubscriptionConfig is an extension of Google PubSub's SubscriptionConfig that enables further configuration for a subscription of a topic

type TopicConfig

type TopicConfig struct {
	Settings pubsub.TopicConfig
}

TopicConfig is an extension of Google PubSub's TopicConfig that enables further configuration for a topic