README
¶
pubsub-poc
Example
package main
import (
"context"
"fmt"
"log"
"os"
"os/signal"
"syscall"
"time"
"cloud.google.com/go/pubsub"
pubsubpoc "github.com/allisson/pubsub-poc"
gocloudpubsub "gocloud.dev/pubsub"
_ "gocloud.dev/pubsub/gcppubsub"
)
func publish(ctx context.Context, projectID, topicID string) {
driverURL := fmt.Sprintf("gcppubsub://projects/%s/topics/%s", projectID, topicID)
producer, err := pubsubpoc.OpenProducer(ctx, driverURL)
if err != nil {
log.Fatal(err)
}
counter := 0
for {
counter++
msg := &gocloudpubsub.Message{Body: []byte(fmt.Sprintf(`{"id": %d}`, counter))}
producer.Send(ctx, msg)
time.Sleep(5 * time.Second)
}
}
func main() {
// Use pubsub emulator
// docker run --rm -p "8085:8085" allisson/gcloud-pubsub-emulator
os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8085")
// Fixtures
ctx := context.Background()
projectID := "my-project"
topicID := "my-topic"
subID := "my-subscription"
consumerHandler := func(ctx context.Context, msg *gocloudpubsub.Message) error {
time.Sleep(10 * time.Second)
return nil
}
maxGoroutines := 10
// Create topic
topic, err := pubsubpoc.GCPCreateTopic(ctx, projectID, topicID)
if err != nil {
log.Fatal(err)
}
// Create subscription
subConfig := pubsub.SubscriptionConfig{Topic: topic, AckDeadline: 60 * time.Second}
_, err = pubsubpoc.GCPCreateSubscription(ctx, projectID, subID, subConfig)
if err != nil {
log.Fatal(err)
}
// Publish messages every 5 seconds
go publish(ctx, projectID, topicID)
// Open consumer
driverURL := fmt.Sprintf("gcppubsub://projects/%s/subscriptions/%s", projectID, subID)
consumer, err := pubsubpoc.OpenConsumer(ctx, driverURL, consumerHandler, maxGoroutines)
if err != nil {
log.Fatal(err)
}
// Graceful shutdown
idleChan := make(chan struct{})
go func() {
sigint := make(chan os.Signal, 1)
// interrupt signal sent from terminal
signal.Notify(sigint, os.Interrupt)
// sigterm signal sent from kubernetes
signal.Notify(sigint, syscall.SIGTERM)
<-sigint
// We received an interrupt signal, shut down.
if err := consumer.Shutdown(ctx); err != nil {
log.Printf("consumer_shutdown_error, err=%s\n", err.Error())
}
close(idleChan)
}()
// Start consumer
if err := consumer.Start(ctx); err != nil {
log.Printf("consumer_error, err=%s\n", err.Error())
}
<-idleChan
log.Println("consumer_shutdown_completed")
}
Documentation
¶
Index ¶
- func GCPCreateSubscription(ctx context.Context, projectID, subID string, ...) (*pubsub.Subscription, error)
- func GCPCreateTopic(ctx context.Context, projectID, topicID string) (*pubsub.Topic, error)
- func NewDevelopmentConfig() zap.Config
- func NewDevelopmentEncoderConfig() zapcore.EncoderConfig
- func NewLogger() (logger *zap.Logger, err error)
- func NewProductionConfig() zap.Config
- func NewProductionEncoderConfig() zapcore.EncoderConfig
- type Consumer
- type Handler
- type Producer
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func GCPCreateSubscription ¶
func GCPCreateSubscription(ctx context.Context, projectID, subID string, subConfig pubsub.SubscriptionConfig) (*pubsub.Subscription, error)
GCPCreateSubscription creates a new subscription on google cloud pubsub and deal if subscription already exists.
func GCPCreateTopic ¶
GCPCreateTopic creates a new topic on google cloud pubsub and deal if topic already exists.
func NewDevelopmentConfig ¶
NewDevelopmentConfig is a reasonable development logging configuration. Logging is enabled at DebugLevel and above.
It enables development mode (which makes DPanicLevel logs panic), uses a console encoder, writes to standard error, and disables sampling. Stacktraces are automatically included on logs of WarnLevel and above.
func NewDevelopmentEncoderConfig ¶
func NewDevelopmentEncoderConfig() zapcore.EncoderConfig
NewDevelopmentEncoderConfig returns an opinionated EncoderConfig for development environments.
func NewLogger ¶
NewLogger returns a zap logger based on LOG_CONFIG envvar if LOG_CONFIG=="production" returns zap.NewProduction() else returns zap.NewDevelopment()
func NewProductionConfig ¶
NewProductionConfig is a reasonable production logging configuration. Logging is enabled at InfoLevel and above.
It uses a JSON encoder, writes to standard error, and enables sampling. Stacktraces are automatically included on logs of ErrorLevel and above.
func NewProductionEncoderConfig ¶
func NewProductionEncoderConfig() zapcore.EncoderConfig
NewProductionEncoderConfig returns an opinionated EncoderConfig for production environments.
Types ¶
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer allows you to consume message from a specific subscription.
func OpenConsumer ¶
func OpenConsumer(ctx context.Context, driverURL string, fn Handler, maxGoroutines int) (Consumer, error)
OpenConsumer returns a new consumer.
type Handler ¶
type Handler func(ctx context.Context, msg *gocloudpubsub.Message) error
Handler represents a function to be passed to consumer.
Source Files