eventbus

README

EventBus library

Preparation

First, to be able to publish or subscribe to pub/sub topics, you need to configure it in provisioning profile. For that, you need to:

1. Open .gitlab/terraform/pubsub.tf
2. For the topics you need to publish to, add corresponding names to publishes array
3. For the topics you need to subscribe, add them to subscriptions array

I.e:

  subscriptions = [
    "content"
  ]
  publishes = [
    "circle"
  ]

Configuration

This library uses config library for client configuration. For eventbus client to be able to initialize, following parameters have to be added to configuration .yaml file.

Environment id

Environment id is required for generating correct subscription name, and is also used to set filtering attribute to message payload.

app:
  env:
    id: ~

Service name

Service name is used for generating correct subscription name. In general, subscriptions naming scheme is as follows: ${topicName}.${serviceName}.${envID}.

env:
  service:
    name: example-api

Project id

Project id is used to connect to pub/sub instance.

env:
  pubsub:
    project: picnic-stg

Topics names

List of topics that are going to be used in the project. Keys here are predefined in set of constants inside the library, and values correspond to real names of topics in pub/sub instance.

env:
  pubsub:
    topic:
      profile: profile
      circle: circle
      content: content
      comment: comment

Initialization

Load config

To start with, make sure your service configuration has loaded properly. Configuration loading should be performed upon service initialization and can be done in two ways:

1. Using config library

package main

import (
	"log"

	"gitlab.com/picnic-app/backend/libs/golang/config"
)

func main() {
	err := config.Load()
	if err != nil {
		log.Fatal(err)
	}
}

1. Using core library

err := core.LoadConfig()

Create default client

Once configuration has been loaded, you can create default client, provided that all required parameters are set correctly in corresponding .yaml file.

cli, err := eventbus.DefaultClient(context.Background())
if err != nil {
    log.Fatal(err)
}
defer cli.Close()

Getting topic

Topic object can be retrieved using eventbus client interface method Topic(topicID string). To get topicID value there's a helper method eventbus.TopicID(), which returns all the topic ids registered in the client configuration. Only pre-defined set of topics can be used this way.

So, for example, to get topic for circles, you can use this code:

topic = cli.Topic(eventbus.TopicID().Circle)

Using arbitrary string topic id here is allowed, although not recommended, unless you know exactly what you are doing (if you don't - most likely it won't work).

Topic object provides simple interface with 2 methods:

type Topic interface {
    Pub(ctx context.Context, msg any) error
    Sub(ctx context.Context, handler Handler) error
}

Publishing events to topic

First, you need to create event payload like this:

import "gitlab.com/picnic-app/backend/libs/golang/eventbus/event"

evt := event.NewEvent(event.TypeCircleJoin, event.JoinCirclePayload{
    CircleID: "b7102e85-e217-4dda-a3b1-0f3636f5e3b2",
    UserID:   "05703f5d-b905-423d-a6f3-a043f3e3c6cc",
})

Library includes most of the events that are used system-wide. If there's no event that you need - it can be added by issued merge request with its definition.

As before, creation of event of some custom type which is not defined by the library is possible, but not recommended. If you need to do so - just pass string value as type argument to event.NewEvent method along with any serializable payload (struct with json tags). Please note that receiver of event will need to know how to deserialize it.

Once event has been created, publish it to a registered topic like this:

err := topic.Pub(context.Background(), evt)

It's possible to get topic and publish event in a single line:

err := cli.Topic(eventbus.TopicID().Circle).Pub(context.Background(), evt)

Subscribing to events in topic

To be able to subscribe to topic, you need to create event handler first. Handler is an interface defined by the library with the following signature:

type Handler func(ctx context.Context, evt *event.Event)

Minimal event handling would look like this:

func evtHandler(ctx context.Context, evt *event.Event) {
    fmt.Printf("%+v", evt)
}

Next, you need to pass it the to Sub() method of topic object:

err := topic.Sub(context.Background(), evtHandler)

Upon receiving event from corresponding topic subscription, handler will be invoked with deserialized event as an argument. It is possible to use passed context for synchronization since it will be passed as the first argument to handler function.

Examples

Example code showing library usage can be found at examples/pubsub.go file. It can be run by issuing make examples command in the library root folder.

TODO

More sophisticated event generation and handling routines can/should be implemented.

Documentation

Index

• func TopicID() embedTopics
• func WithSubMaxOutstandingMessages(messages int) func(*pubsub.ReceiveSettings)
• type Client
  • func DefaultClient(ctx context.Context) (Client, error)
  • func FromConfig(ctx context.Context, cfg Config) (Client, error)
  • func NewClient(ctx context.Context, projectID, svcID, envID, adcCredPath string) (Client, error)
  • func NewWithPubSubClientClient(cli *pubsub.Client, projectID, svcID, envID string) Client
• type Config
• type Logger
  • func NewLogger(opts ...LoggerOpt) *Logger
  • func (m Logger) Middleware(next internal.Handler) func(context.Context, event.Event) error
• type LoggerOpt
  • func WithRedactPayloadFunc(f func(string, any) any) LoggerOpt
  • func WithTopic(topic string) LoggerOpt
• type Topic

Functions

func TopicID

func TopicID() embedTopics

TopicID ...

func WithSubMaxOutstandingMessages

func WithSubMaxOutstandingMessages(messages int) func(*pubsub.ReceiveSettings)

WithSubMaxOutstandingMessages ...

Types

type Client

type Client interface {
	internal.Client
}

Client ...

func DefaultClient

func DefaultClient(ctx context.Context) (Client, error)

DefaultClient creates eventbus client from environment configuration

func FromConfig

func FromConfig(ctx context.Context, cfg Config) (Client, error)

FromConfig creates eventbus client with configuration

func NewClient

func NewClient(ctx context.Context, projectID, svcID, envID, adcCredPath string) (Client, error)

NewClient ...

func NewWithPubSubClientClient

func NewWithPubSubClientClient(cli *pubsub.Client, projectID, svcID, envID string) Client

NewWithPubSubClientClient ...

type Config

type Config interface {
	internal.Config
}

Config ...

type Logger

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

Logger is a logger middleware.

func NewLogger

func NewLogger(
	opts ...LoggerOpt,
) *Logger

NewLogger creates a new logger middleware.

func (Logger) Middleware

func (m Logger) Middleware(next internal.Handler) func(context.Context, event.Event) error

Middleware is a logger middleware.

type LoggerOpt

type LoggerOpt func(*Logger)

LoggerOpt is a logger middleware option.

func WithRedactPayloadFunc

func WithRedactPayloadFunc(f func(string, any) any) LoggerOpt

WithRedactPayloadFunc sets the func that will redact event payloads before logging them.

func WithTopic

func WithTopic(topic string) LoggerOpt

WithTopic sets the topic name.

type Topic

type Topic interface {
	internal.Topic
}