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.

Events

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

Attributes

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.

Go

main.go:

package main

import (
	"context"
	"encoding/hex"
	"errors"
	"log"

	"cloud.google.com/go/pubsub"
	"google.golang.org/api/option"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"

	"github.com/chirpstack/chirpstack/api/go/v4/integration"
)

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) {
	msg.Ack()

	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)
	default:
		log.Printf("handler for event %s is not implemented", event)
		return
	}

	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 {
		panic(err)
	}

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

Python

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

main.py:

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':
                    self.up(msg.message.data)
                elif event == 'join':
                    self.join(msg.message.data)
                else:
                    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)
        
        pl.ParseFromString(body)
        return pl


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

    # GCP Project ID
    "project-id",

    # GCP Pub/Sub Subsciption name
    "subscription-name",
)
h.receive()