Activadores de Google Cloud Pub/Sub

Las funciones de Cloud Functions se pueden activar mediante mensajes publicados en temas de Pub/Sub en el mismo proyecto de Cloud en el que se encuentran las funciones. Pub/Sub es un bus de mensajes distribuido a nivel global que se escala de forma automática cuando lo necesitas y brinda la base para que compiles tus propios servicios globales sólidos.

Tipos de eventos

Cloud Functions usa un solo evento de Pub/Sub y tiene el valor de tipo de activador google.pubsub.topic.publish.

Este evento se envía cuando se publica un mensaje en un tema de Pub/Sub que se especifica cuando se implementa una función. Cada mensaje publicado en este tema activará la ejecución de la función con el contenido del mensaje pasado como datos de entrada.

Estructura de eventos

Las funciones de Cloud Functions activadas desde un tema de Pub/Sub recibirán eventos que se ajusten al tipo de PubsubMessage, con la salvedad de que publishTime y messageId no están disponibles directamente en el PubsubMessage. En cambio, puedes acceder a publishTime y messageId a través del ID del evento y las propiedades de marca de tiempo de los metadatos del evento. Se puede acceder a estos metadatos a través del objeto de contexto que se traslada a tu función cuando se invoca.

La carga útil del objeto PubsubMessage, los datos publicados sobre el tema, se almacena como una string codificada en base64 en el atributo data de PubsubMessage. Para extraer la carga útil del objeto PubsubMessage, es posible que debas decodificar el atributo de datos, como se muestra en los ejemplos a continuación.

Código de muestra

Node.js

/**
 * Background Cloud Function to be triggered by Pub/Sub.
 * This function is exported by index.js, and executed when
 * the trigger topic receives a message.
 *
 * @param {object} message The Pub/Sub message.
 * @param {object} context The event metadata.
 */
exports.helloPubSub = (message, context) => {
  const name = message.data
    ? Buffer.from(message.data, 'base64').toString()
    : 'World';

  console.log(`Hello, ${name}!`);
};

Python

def hello_pubsub(event, context):
    """Background Cloud Function to be triggered by Pub/Sub.
    Args:
         event (dict):  The dictionary with data specific to this type of
         event. The `data` field contains the PubsubMessage message. The
         `attributes` field will contain custom attributes if there are any.
         context (google.cloud.functions.Context): The Cloud Functions event
         metadata. The `event_id` field contains the Pub/Sub message ID. The
         `timestamp` field contains the publish time.
    """
    import base64

    print("""This Function was triggered by messageId {} published at {}
    """.format(context.event_id, context.timestamp))

    if 'data' in event:
        name = base64.b64decode(event['data']).decode('utf-8')
    else:
        name = 'World'
    print('Hello {}!'.format(name))

Go


// Package helloworld provides a set of Cloud Functions samples.
package helloworld

import (
	"context"
	"log"
)

// PubSubMessage is the payload of a Pub/Sub event.
type PubSubMessage struct {
	Data []byte `json:"data"`
}

// HelloPubSub consumes a Pub/Sub message.
func HelloPubSub(ctx context.Context, m PubSubMessage) error {
	name := string(m.Data) // Automatically decoded from base64.
	if name == "" {
		name = "World"
	}
	log.Printf("Hello, %s!", name)
	return nil
}

Java


import com.google.cloud.functions.BackgroundFunction;
import com.google.cloud.functions.Context;
import functions.eventpojos.PubSubMessage;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.logging.Level;
import java.util.logging.Logger;

public class HelloPubSub implements BackgroundFunction<PubSubMessage> {
  private static final Logger logger = Logger.getLogger(HelloPubSub.class.getName());

  @Override
  public void accept(PubSubMessage message, Context context) {
    String name = "world";
    if (message != null && message.getData() != null) {
      name = new String(
          Base64.getDecoder().decode(message.getData().getBytes(StandardCharsets.UTF_8)),
          StandardCharsets.UTF_8);
    }
    logger.info(String.format("Hello %s!", name));
    return;
  }
}
.

Publica un mensaje desde una función

También puedes publicar un mensaje en un tema de Pub/Sub desde una función. Esto te permite activar invocaciones posteriores de Cloud Functions con mensajes de Cloud Pub/Sub. Puedes usar esta técnica para hacer lo siguiente:

  • Unir llamadas a funciones secuenciales
  • Distribuir (o “fan-out”) grupos de tareas en paralelo en varias instancias de Cloud Functions

En el siguiente ejemplo, una función de HTTP publish envía un mensaje a un tema de Pub/Sub y, a su vez, activa una función subscribe.

En este fragmento, se muestra la función publish que publica un mensaje en un tema de Pub/Sub:

Node.js

/**
 * Publishes a message to a Cloud Pub/Sub Topic.
 *
 * @example
 * gcloud functions call publish --data '{"topic":"[YOUR_TOPIC_NAME]","message":"Hello, world!"}'
 *
 *   - Replace `[YOUR_TOPIC_NAME]` with your Cloud Pub/Sub topic name.
 *
 * @param {object} req Cloud Function request context.
 * @param {object} req.body The request body.
 * @param {string} req.body.topic Topic name on which to publish.
 * @param {string} req.body.message Message to publish.
 * @param {object} res Cloud Function response context.
 */
exports.publish = async (req, res) => {
  if (!req.body.topic || !req.body.message) {
    res
      .status(500)
      .send(
        'Missing parameter(s); include "topic" and "subscription" properties in your request.'
      );
    return;
  }

  console.log(`Publishing message to topic ${req.body.topic}`);

  // References an existing topic
  const topic = pubsub.topic(req.body.topic);

  const messageObject = {
    data: {
      message: req.body.message,
    },
  };
  const messageBuffer = Buffer.from(JSON.stringify(messageObject), 'utf8');

  // Publishes a message
  try {
    await topic.publish(messageBuffer);
    res.status(200).send('Message published.');
  } catch (err) {
    console.error(err);
    res.status(500).send(err);
    return Promise.reject(err);
  }
};

Este fragmento muestra la función subscribe que se activa cuando el mensaje se publica en el tema de Pub/Sub:

Node.js

/**
 * Triggered from a message on a Cloud Pub/Sub topic.
 *
 * @param {object} pubsubMessage The Cloud Pub/Sub Message object.
 * @param {string} pubsubMessage.data The "data" property of the Cloud Pub/Sub Message.
 */
exports.subscribe = (pubsubMessage) => {
  // Print out the data from Pub/Sub, to prove that it worked
  console.log(Buffer.from(pubsubMessage.data, 'base64').toString());
};

En producción, puedes usar la utilidad de línea de comandos cURL para invocar la función publish, de la siguiente manera:

curl https://GCF_REGION-GCP_PROJECT_ID.cloudfunctions.net/publish -X POST  -d "{\"topic\": \"PUBSUB_TOPIC\", \"message\":\"YOUR_MESSAGE\"}" -H "Content-Type: application/json"

Sin embargo, en el caso de las pruebas y la depuración, puedes utilizar el comando gcloud functions call para llamar a la función directamente. En los siguientes pasos, se describe cómo ejecutar el ejemplo anterior con el comando gcloud functions call:

  1. Crea un tema de Pub/Sub, en el que MY_TOPIC sea el nombre del tema nuevo que estás creando:

    gcloud pubsub topics create MY_TOPIC
  2. Implementa la función publish, en la cual RUNTIME es el nombre del entorno de ejecución que estás utilizando, como nodejs8:

    gcloud functions deploy publish --trigger-http --runtime RUNTIME
  3. Implementa la función subscribe:

    gcloud functions deploy subscribe --trigger-topic MY_TOPIC --runtime RUNTIME
  4. Invoca la función publish directamente mediante el comando gcloud functions call y proporciona los datos obligatorios, como JSON en el argumento --data:

    gcloud functions call publish --data '{"topic":"MY_TOPIC","message":"Hello World!"}'
  5. Verifica los registros de la función subscribe. Ten en cuenta que los resultados aparecen en el registro luego de unos minutos:

    gcloud functions logs read subscribe

Deberías ver un resultado similar a este:

D    ...Function execution started
I    ...{"data":{"message":"Hello World!"}}
D    ...Function execution took 753 ms, finished with status: 'ok'

Implementa la función

Mediante el siguiente comando de gcloud, se implementa una función que se activa cuando un mensaje se publica en un tema de Pub/Sub:

gcloud functions deploy FUNCTION_NAME --trigger-topic TOPIC_NAME FLAGS...
Argumento Descripción
FUNCTION_NAME Es el nombre de la función.
--trigger-topic TOPIC_NAME Es el nombre del tema de Pub/Sub al que se suscribe la función. Si el tema no existe, se crea durante la implementación.
FLAGS... Marcas adicionales que debes especificar durante la implementación, como --runtime. Para obtener una referencia completa, consulta la documentación de gcloud functions deploy.

Consulta el Instructivo de Pub/Sub para ver un ejemplo completo de cómo usar los activadores de Pub/Sub.

Activadores de Cloud Pub/Sub heredados

Con el comando de gcloud que aparece a continuación, se implementa una función que se activa mediante notificaciones heredadas de Pub/Sub en un tema específico. Estas notificaciones son compatibles con las funciones heredadas que ya consumen estos eventos. Sin embargo, recomendamos usar la marca --trigger-topic en su lugar, ya que las notificaciones heredadas podrían quitarse en el futuro.

gcloud functions deploy FUNCTION_NAME \
--trigger-resource TOPIC_NAME \
--trigger-event providers/cloud.pubsub/eventTypes/topic.publish \
FLAGS...
.

Próximos pasos

Consulta el Instructivo de Pub/Sub para ver un ejemplo de cómo implementar una función en segundo plano que se activa mediante Pub/Sub.