GCP Pub/Sub

The Google Cloud Platform Pub/Sub integration publishes all the events to a configurable GCP Pub/Sub topic. Using the GCP console (or APIs) you are able to create one or multiple Pub/Sub subscriptions, for integrating this with your application(s) or store the data in one of the storage options provided by the Google Cloud Platform.


The GCP Pub/Sub integration exposes all events as documented by Event types.


The following attributes are added to each Pub/Sub message:

  • event: the event type
  • devEui: the device EUI to which the event relates

Example code

The following code example demonstrates how to consume integration events using a GCP Pub/Sub Subscription.



package main

import (



type handler struct {
	json         bool
	client       *pubsub.Client
	subscription *pubsub.Subscription

func (h *handler) receive() error {
	for {
		if err := h.subscription.Receive(context.TODO(), h.receiveFunc); err != nil {
			return err

func (h *handler) receiveFunc(ctx context.Context, msg *pubsub.Message) {

	event, ok := msg.Attributes["event"]
	if !ok {
		log.Printf("event attribute is missing")

	var err error

	switch event {
	case "up":
		err = h.up(msg.Data)
	case "join":
		err = h.join(msg.Data)
		log.Printf("handler for event %s is not implemented", event)

	if err != nil {
		log.Printf("handling event '%s' returned error: %s", event, err)

func (h *handler) up(b []byte) error {
	var up integration.UplinkEvent
	if err := h.unmarshal(b, &up); err != nil {
		return err
	log.Printf("Uplink received from %s with payload: %s", up.GetDeviceInfo().DevEui, hex.EncodeToString(up.Data))
	return nil

func (h *handler) join(b []byte) error {
	var join integration.JoinEvent
	if err := h.unmarshal(b, &join); err != nil {
		return err
	log.Printf("Device %s joined with DevAddr %s", join.GetDeviceInfo().DevEui, join.DevAddr)
	return nil

func (h *handler) unmarshal(b []byte, v proto.Message) error {
	if h.json {
		return protojson.UnmarshalOptions{
			DiscardUnknown: true,
			AllowPartial:   true,
		}.Unmarshal(b, v)
	return proto.Unmarshal(b, v)

func newHandler(json bool, projectID, subscriptionName, credentialsFile string) (*handler, error) {
	var opts []option.ClientOption

	if credentialsFile != "" {
		opts = append(opts, option.WithCredentialsFile(credentialsFile))

	client, err := pubsub.NewClient(context.TODO(), projectID, opts...)
	if err != nil {
		return nil, err

	subscription := client.Subscription(subscriptionName)
	exists, err := subscription.Exists(context.TODO())
	if err != nil {
		return nil, err
	if !exists {
		return nil, errors.New("subscription does not exist")

	return &handler{
		json:         json,
		subscription: subscription,
		client:       client,
	}, nil

func main() {
	h, err := newHandler(
		false,               // set to true when using JSON encoding
		"project-id",        // replace with GCP project ID
		"subscription-name", // replace with GCP Pub/Sub Subscription name
		"",                  // path to GCP credentials file
	if err != nil {

	if err := h.receive(); err != nil {


Please refer to the Setting up authentication section for creating a service-account and setting up the credentials.


from chirpstack_api import integration
from google.cloud import pubsub_v1
from google.protobuf.json_format import Parse

class Handler:
    def __init__(self, json, project_id, subscription_name):
        self.json = json
        self.project_id = project_id
        self.subscription_name = subscription_name

    def receive(self):
        subscriber = pubsub_v1.SubscriberClient()
        sub_path = subscriber.subscription_path(self.project_id, self.subscription_name)

        while True:
            resp = subscriber.pull(sub_path, max_messages=10)
            for msg in resp.received_messages:
                event = msg.message.attributes['event']
                subscriber.acknowledge(sub_path, [msg.ack_id])

                if event == 'up':
                elif event == 'join':
                    print('handler for event %s is not implemented' % event)

    def up(self, body):
        up = self.unmarshal(body, integration.UplinkEvent())
        print('Uplink received from: %s with payload: %s' % (up.dev_eui.hex(), up.data.hex()))

    def join(self, body):
        join = self.unmarshal(body, integration.JoinEvent())
        print('Device: %s joined with DevAddr: %s' % (join.dev_eui.hex(), join.dev_addr.hex()))

    def unmarshal(self, body, pl):
        if self.json:
            return Parse(body, pl)
        return pl

h = Handler(
    # True -  JSON marshaler
    # False - Protobuf marshaler (binary)

    # GCP Project ID

    # GCP Pub/Sub Subsciption name