gcppubsub

package
v1.8.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 39 Imported by: 0

README

Google Cloud Pub/Sub

Parity grade: B · SDK cloud.google.com/go/pubsub/v2@v2.7.0 · last audited 2026-10-18 (f5ea09414)

Coverage

Metric Value
PARITY entries audited 35 (34 ok, 1 partial)
Feature families 6 (3 ok, 3 partial)
Known gaps 6
Deferred items 0
Resource leaks clean
Known gaps
  • No IAMPolicy RPCs (SetIamPolicy/GetIamPolicy/TestIamPermissions) on topics, subscriptions, snapshots or schemas.
  • BigQuery, Cloud Storage and Bigtable subscriptions, topic ingestion sources and message transforms are rejected UNIMPLEMENTED.
  • PROTOCOL_BUFFER schemas are not compiled: definitions get a lexical check and messages a wire-format or JSON-object check only.
  • Push OIDC tokens are stored but not generated; push endpoints are not verified.
  • Snapshot expiry is a fixed 7 days from creation, not derived from the oldest unacked message.
  • Flow control is per stream; server-side per-subscription limits and publish quotas are not modelled.

More

Documentation

Index

Constants

View Source 
const DefaultPort = 8085

DefaultPort is the dedicated gRPC listener's default TCP port (the standard Pub/Sub emulator port).

Variables

View Source 
var ErrNilAppContext = errors.New("gcppubsub: nil app context")

ErrNilAppContext is returned when Init is called with a nil AppContext.

Functions

This section is empty.

Types

type ConfigProvider 

type ConfigProvider interface {
	GetGCPPubSubSettings() Settings
}

ConfigProvider extracts Pub/Sub configuration from the AppContext Config.

type Handler 

type Handler struct {
	Backend *InMemoryBackend

	Port int
	// contains filtered or unexported fields
}

Handler serves the Pub/Sub gRPC API (Publisher, Subscriber, SchemaService) on a dedicated listener.

func NewHandler 

func NewHandler(backend *InMemoryBackend) *Handler

NewHandler creates a Handler; callers override Port from Settings.

func (*Handler) Addr 

func (h *Handler) Addr() string

Addr returns the bound listener address, or "" before StartWorker.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(*echo.Context) string

ExtractOperation is unused: requests never reach the shared router.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(*echo.Context) string

ExtractResource is unused: requests never reach the shared router.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations lists the supported gRPC methods.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns a 404 handler; the real API is gRPC on the dedicated port.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority is irrelevant because RouteMatcher never matches.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) PublisherAPI 

func (h *Handler) PublisherAPI() pubsubpb.PublisherServer

PublisherAPI exposes the Publisher service for in-process callers (dashboard proxy).

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all in-memory state.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher never matches: Pub/Sub runs on its own gRPC listener.

func (*Handler) Shutdown 

func (h *Handler) Shutdown(ctx context.Context)

Shutdown stops the gRPC server and the push dispatcher, forcing a close after a short grace period.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker 

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker binds the dedicated listener synchronously so a port conflict is returned to the caller.

func (*Handler) SubscriberAPI 

func (h *Handler) SubscriberAPI() pubsubpb.SubscriberServer

SubscriberAPI exposes the Subscriber service for in-process callers (dashboard proxy).

type InMemoryBackend 

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

InMemoryBackend holds all topics, subscriptions, snapshots and schemas under one lock.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates an empty backend.

func (*InMemoryBackend) Acknowledge 

func (b *InMemoryBackend) Acknowledge(name string, ids []string) ([]string, []string, error)

Acknowledge acknowledges messages; ids that are not currently leased are returned as invalid.

func (*InMemoryBackend) CommitSchema 

func (b *InMemoryBackend) CommitSchema(req *pubsubpb.CommitSchemaRequest) (*pubsubpb.Schema, error)

CommitSchema adds a revision to an existing schema.

func (*InMemoryBackend) CreateSchema 

func (b *InMemoryBackend) CreateSchema(req *pubsubpb.CreateSchemaRequest) (*pubsubpb.Schema, error)

CreateSchema stores a new schema with its first revision.

func (*InMemoryBackend) CreateSnapshot 

func (b *InMemoryBackend) CreateSnapshot(req *pubsubpb.CreateSnapshotRequest) (*pubsubpb.Snapshot, error)

CreateSnapshot captures a subscription's unacknowledged messages.

func (*InMemoryBackend) CreateSubscription 

func (b *InMemoryBackend) CreateSubscription(in *pubsubpb.Subscription) (*pubsubpb.Subscription, error)

CreateSubscription stores a new subscription.

func (*InMemoryBackend) CreateTopic 

func (b *InMemoryBackend) CreateTopic(in *pubsubpb.Topic) (*pubsubpb.Topic, error)

CreateTopic stores a new topic.

func (*InMemoryBackend) DeleteSchema 

func (b *InMemoryBackend) DeleteSchema(name string) error

DeleteSchema removes a schema; topics that reference it are re-pointed at the deleted-schema marker.

func (*InMemoryBackend) DeleteSchemaRevision 

func (b *InMemoryBackend) DeleteSchemaRevision(req *pubsubpb.DeleteSchemaRevisionRequest) (*pubsubpb.Schema, error)

DeleteSchemaRevision removes one revision; the last remaining revision cannot be deleted.

func (*InMemoryBackend) DeleteSnapshot 

func (b *InMemoryBackend) DeleteSnapshot(name string) error

DeleteSnapshot removes a snapshot.

func (*InMemoryBackend) DeleteSubscription 

func (b *InMemoryBackend) DeleteSubscription(name string) error

DeleteSubscription removes a subscription.

func (*InMemoryBackend) DeleteTopic 

func (b *InMemoryBackend) DeleteTopic(name string) error

DeleteTopic removes a topic; its subscriptions remain, pointing at the deleted-topic marker.

func (*InMemoryBackend) DetachSubscription 

func (b *InMemoryBackend) DetachSubscription(name string) error

DetachSubscription detaches a subscription from its topic.

func (*InMemoryBackend) GetSchema 

func (b *InMemoryBackend) GetSchema(req *pubsubpb.GetSchemaRequest) (*pubsubpb.Schema, error)

GetSchema returns a schema (or one revision when the name carries @revision).

func (*InMemoryBackend) GetSnapshot 

func (b *InMemoryBackend) GetSnapshot(name string) (*pubsubpb.Snapshot, error)

GetSnapshot returns a snapshot.

func (*InMemoryBackend) GetSubscription 

func (b *InMemoryBackend) GetSubscription(name string) (*pubsubpb.Subscription, error)

GetSubscription returns a subscription.

func (*InMemoryBackend) GetTopic 

func (b *InMemoryBackend) GetTopic(name string) (*pubsubpb.Topic, error)

GetTopic returns a topic.

func (*InMemoryBackend) ListSchemaRevisions 

func (b *InMemoryBackend) ListSchemaRevisions(
	req *pubsubpb.ListSchemaRevisionsRequest,
) ([]*pubsubpb.Schema, string, error)

ListSchemaRevisions lists all revisions of a schema, newest first.

func (*InMemoryBackend) ListSchemas 

func (b *InMemoryBackend) ListSchemas(req *pubsubpb.ListSchemasRequest) ([]*pubsubpb.Schema, string, error)

ListSchemas lists the latest revision of every schema of a project.

func (*InMemoryBackend) ListSnapshots 

func (b *InMemoryBackend) ListSnapshots(req *pubsubpb.ListSnapshotsRequest) ([]*pubsubpb.Snapshot, string, error)

ListSnapshots lists the snapshots of a project.

func (*InMemoryBackend) ListSubscriptions 

func (b *InMemoryBackend) ListSubscriptions(
	req *pubsubpb.ListSubscriptionsRequest,
) ([]*pubsubpb.Subscription, string, error)

ListSubscriptions lists the subscriptions of a project.

func (*InMemoryBackend) ListTopicSnapshots 

func (b *InMemoryBackend) ListTopicSnapshots(req *pubsubpb.ListTopicSnapshotsRequest) ([]string, string, error)

ListTopicSnapshots lists the snapshots of a topic.

func (*InMemoryBackend) ListTopicSubscriptions 

func (b *InMemoryBackend) ListTopicSubscriptions(
	req *pubsubpb.ListTopicSubscriptionsRequest,
) ([]string, string, error)

ListTopicSubscriptions lists the attached subscriptions of a topic.

func (*InMemoryBackend) ListTopics 

func (b *InMemoryBackend) ListTopics(req *pubsubpb.ListTopicsRequest) ([]*pubsubpb.Topic, string, error)

ListTopics lists the topics of a project.

func (*InMemoryBackend) ModifyAckDeadline 

func (b *InMemoryBackend) ModifyAckDeadline(name string, ids []string, seconds []int32) ([]string, []string, error)

ModifyAckDeadline extends (or, with zero seconds, releases) the leases of the given ack ids.

func (*InMemoryBackend) ModifyPushConfig 

func (b *InMemoryBackend) ModifyPushConfig(name string, pc *pubsubpb.PushConfig) error

ModifyPushConfig switches a subscription between pull and push delivery.

func (*InMemoryBackend) Publish 

func (b *InMemoryBackend) Publish(req *pubsubpb.PublishRequest) ([]string, error)

Publish appends messages to a topic and fans them out to its subscriptions.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all state.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a snapshot; message leases are not restored, so in-flight messages redeliver.

func (*InMemoryBackend) RollbackSchema 

func (b *InMemoryBackend) RollbackSchema(req *pubsubpb.RollbackSchemaRequest) (*pubsubpb.Schema, error)

RollbackSchema creates a new revision copying an earlier one.

func (*InMemoryBackend) RunPush 

func (b *InMemoryBackend) RunPush(ctx context.Context)

RunPush delivers messages of push subscriptions until ctx is cancelled, then waits for in-flight requests.

func (*InMemoryBackend) Seek 

func (b *InMemoryBackend) Seek(req *pubsubpb.SeekRequest) error

Seek rewinds a subscription to a point in time or to a snapshot.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises the backend state, including unacknowledged messages.

func (*InMemoryBackend) UpdateSnapshot 

func (b *InMemoryBackend) UpdateSnapshot(req *pubsubpb.UpdateSnapshotRequest) (*pubsubpb.Snapshot, error)

UpdateSnapshot applies a masked update (labels, expire_time).

func (*InMemoryBackend) UpdateSubscription 

func (b *InMemoryBackend) UpdateSubscription(req *pubsubpb.UpdateSubscriptionRequest) (*pubsubpb.Subscription, error)

UpdateSubscription applies a masked update.

func (*InMemoryBackend) UpdateTopic 

func (b *InMemoryBackend) UpdateTopic(req *pubsubpb.UpdateTopicRequest) (*pubsubpb.Topic, error)

UpdateTopic applies a masked update.

func (*InMemoryBackend) ValidateMessage 

func (b *InMemoryBackend) ValidateMessage(req *pubsubpb.ValidateMessageRequest) error

ValidateMessage checks a payload against a stored or inline schema.

type Provider 

type Provider struct{}

Provider implements service.Provider for Google Cloud Pub/Sub. It serves gRPC on its own listener (see Handler.StartWorker) and never matches on the shared AWS router.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init creates the backend and handler; the port is bound in StartWorker.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the service provider name.

type Settings 

type Settings struct {
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 172-byte string literal not displayed */
}

Settings holds service-level configuration, embedded in the root CLI command as CLI.GCPPubSub.

func DefaultSettings 

func DefaultSettings() Settings

DefaultSettings returns the settings used when no ConfigProvider is available.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.