Client Management

The Pub/Sub client is the main entry point for interacting with Google Cloud Pub/Sub. It is scoped to a single Google Cloud project and provides methods for creating and accessing topics, subscriptions, and snapshots.

Client Creation

Creating a Client

func NewClient(ctx context.Context, projectID string, opts ...option.ClientOption) (*Client, error)

Creates a new Pub/Sub client for the specified project.

Parameters:

• ctx: Context for the operation
• projectID: Google Cloud project ID, or DetectProjectID for automatic detection
• opts: Optional client options (e.g., credentials, endpoint, connection pool settings)

Returns: Pointer to Client and error

Example:

client, err := pubsub.NewClient(ctx, "my-project-id")
if err != nil {
    log.Fatal(err)
}
defer client.Close()

Automatic Project ID Detection

const DetectProjectID = "*detect-project-id*"

Use this sentinel value to automatically detect the project ID from credentials:

client, err := pubsub.NewClient(ctx, pubsub.DetectProjectID)

Client Configuration

type ClientConfig struct {
    PublisherCallOptions        *vkit.PublisherCallOptions
    SubscriberCallOptions       *vkit.SubscriberCallOptions
    EnableOpenTelemetryTracing  bool
}

Configure custom call options or enable OpenTelemetry tracing.

Client Type

type Client struct {
    // Has unexported fields
}

The Client type provides methods for managing Pub/Sub resources.

Client Methods

Resource Cleanup

func (c *Client) Close() error

Closes the client and releases all resources. Always defer Close() after creating a client.

Project Information

func (c *Client) Project() string

Returns the project ID associated with the client.

Accessing Topics

func (c *Client) Topic(id string) *Topic

Returns a reference to a topic in the client's project. Does not check if the topic exists.

Parameters:

• id: Topic ID (not the full resource name)

Returns: Pointer to Topic

func (c *Client) TopicInProject(id, projectID string) *Topic

Returns a reference to a topic in a specific project.

Parameters:

• id: Topic ID
• projectID: Project ID (can differ from client's project)

Returns: Pointer to Topic

Accessing Subscriptions

func (c *Client) Subscription(id string) *Subscription

Returns a reference to a subscription in the client's project.

Parameters:

• id: Subscription ID (not the full resource name)

Returns: Pointer to Subscription

func (c *Client) SubscriptionInProject(id, projectID string) *Subscription

Returns a reference to a subscription in a specific project.

Parameters:

• id: Subscription ID
• projectID: Project ID (can differ from client's project)

Returns: Pointer to Subscription

Accessing Snapshots

func (c *Client) Snapshot(id string) *Snapshot

Returns a reference to a snapshot in the client's project.

Parameters:

• id: Snapshot ID

Returns: Pointer to Snapshot

Creating Topics

func (c *Client) CreateTopic(ctx context.Context, topicID string) (*Topic, error)

Creates a new topic with default configuration.

Parameters:

• ctx: Context for the operation
• topicID: ID for the new topic

Returns: Pointer to created Topic and error

Example:

topic, err := client.CreateTopic(ctx, "my-topic")
if err != nil {
    log.Fatal(err)
}

func (c *Client) CreateTopicWithConfig(ctx context.Context, topicID string, tc *TopicConfig) (*Topic, error)

Creates a new topic with custom configuration.

Parameters:

• ctx: Context for the operation
• topicID: ID for the new topic
• tc: Topic configuration (labels, schema settings, retention, etc.)

Returns: Pointer to created Topic and error

Example:

topic, err := client.CreateTopicWithConfig(ctx, "my-topic", &pubsub.TopicConfig{
    Labels: map[string]string{
        "env": "production",
    },
    RetentionDuration: 24 * time.Hour,
})

Creating Subscriptions

func (c *Client) CreateSubscription(ctx context.Context, id string, cfg SubscriptionConfig) (*Subscription, error)

Creates a new subscription with the specified configuration.

Parameters:

• ctx: Context for the operation
• id: ID for the new subscription
• cfg: Subscription configuration including topic, ack deadline, delivery settings, etc.

Returns: Pointer to created Subscription and error

Example:

sub, err := client.CreateSubscription(ctx, "my-subscription", pubsub.SubscriptionConfig{
    Topic:       topic,
    AckDeadline: 20 * time.Second,
})
if err != nil {
    log.Fatal(err)
}

Listing Topics

func (c *Client) Topics(ctx context.Context) *TopicIterator

Returns an iterator for listing all topics in the project.

Parameters:

• ctx: Context for the operation

Returns: Pointer to TopicIterator

Example:

it := client.Topics(ctx)
for {
    topic, err := it.Next()
    if err == iterator.Done {
        break
    }
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(topic.ID())
}

Listing Subscriptions

func (c *Client) Subscriptions(ctx context.Context) *SubscriptionIterator

Returns an iterator for listing all subscriptions in the project.

Parameters:

• ctx: Context for the operation

Returns: Pointer to SubscriptionIterator

Example:

it := client.Subscriptions(ctx)
for {
    sub, err := it.Next()
    if err == iterator.Done {
        break
    }
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(sub.ID())
}

Listing Snapshots

func (c *Client) Snapshots(ctx context.Context) *SnapshotConfigIterator

Returns an iterator for listing all snapshots in the project.

Parameters:

• ctx: Context for the operation

Returns: Pointer to SnapshotConfigIterator

Iterator Types

TopicIterator

type TopicIterator struct {
    // Has unexported fields
}

func (it *TopicIterator) Next() (*Topic, error)
func (it *TopicIterator) NextConfig() (*TopicConfig, error)

Iterator for listing topics. Call Next() repeatedly until it returns iterator.Done.

SubscriptionIterator

type SubscriptionIterator struct {
    // Has unexported fields
}

func (it *SubscriptionIterator) Next() (*Subscription, error)
func (it *SubscriptionIterator) NextConfig() (*SubscriptionConfig, error)

Iterator for listing subscriptions. Call Next() repeatedly until it returns iterator.Done.

SnapshotConfigIterator

type SnapshotConfigIterator struct {
    // Has unexported fields
}

func (it *SnapshotConfigIterator) Next() (*SnapshotConfig, error)

Iterator for listing snapshot configurations.

Connection Pool Configuration

Customize the gRPC connection pool size for high-throughput scenarios:

import "google.golang.org/api/option"

opts := []option.ClientOption{
    option.WithGRPCConnectionPool(2),
}
client, err := pubsub.NewClient(ctx, projectID, opts...)

The default connection pool size is min(4, GOMAXPROCS). Each connection supports up to 100 streams.

Authentication Scopes

const (
    ScopePubSub        = "https://www.googleapis.com/auth/pubsub"
    ScopeCloudPlatform = "https://www.googleapis.com/auth/cloud-platform"
)

OAuth2 scopes for Pub/Sub access. The library automatically uses the appropriate scope.

Emulator Support

Set the PUBSUB_EMULATOR_HOST environment variable to use a Pub/Sub emulator:

import "os"

err := os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8085")
if err != nil {
    log.Fatal(err)
}

// Client will now connect to the emulator
client, err := pubsub.NewClient(ctx, "my-project-id")

Error Handling

var ErrEmptyProjectID = errors.New("pubsub: project ID is empty")

Returned when an empty project ID is provided and automatic detection is not used.