Cette page explique comment créer et déployer un service récepteur d'événements. Le service cible reçoit les requêtes HTTP contenant l'événement au format CloudEvents.
Les fournisseurs d'événements (sources) peuvent fournir les types d'événements suivants :
Réponse du récepteur d'événements
Votre service de réception doit envoyer une réponse HTTP 2xx pour signaler le succès de la réception d'un événement au routeur. Le routeur traite toutes les autres réponses HTTP comme des échecs de distribution et renvoie l'événement.
Open Source
La structure du corps HTTP de tous les événements est disponible dans le dépôt GitHub de CloudEvents.
Le dépôt contient les éléments suivants pour vous aider à comprendre et utiliser les données CloudEvents dans votre langage de programmation :
- Les tampons de protocole Google pour les charges utiles de données CloudEvents
- Les schémas JSON générés
- Un catalogue de schémas JSON public
Des liens vers des bibliothèques clientes sont également inclus.
Utiliser une bibliothèque de SDK CloudEvents
Vous pouvez développer des services de réception d'événements à l'aide de la bibliothèque de SDK CloudEvents, disponible pour les langages suivants :
Ces bibliothèques Open Source facilitent la transformation de votre requête HTTP en objet CloudEvents idiomatique.
Exemple de code source de récepteur
Cloud Audit Logging
L'exemple de code montre comment lire des événements Cloud Storage à l'aide de Cloud Audit Logging dans un service déployé sur Cloud Run.
Python
@app.route('/', methods=['POST'])
def index():
# Gets the GCS bucket name from the CloudEvent header
# Example: "storage.googleapis.com/projects/_/buckets/my-bucket"
bucket = request.headers.get('ce-subject')
print(f"Detected change in Cloud Storage bucket: {bucket}")
return (f"Detected change in Cloud Storage bucket: {bucket}", 200)Java
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class EventController {
private static final List<String> requiredFields =
Arrays.asList("ce-id", "ce-source", "ce-type", "ce-specversion");
@RequestMapping(value = "/", method = RequestMethod.POST)
public ResponseEntity<String> receiveMessage(
@RequestBody Map<String, Object> body, @RequestHeader Map<String, String> headers) {
for (String field : requiredFields) {
if (headers.get(field) == null) {
String msg = String.format("Missing expected header: %s.", field);
System.out.println(msg);
return new ResponseEntity<String>(msg, HttpStatus.BAD_REQUEST);
}
}
if (headers.get("ce-subject") == null) {
String msg = "Missing expected header: ce-subject.";
System.out.println(msg);
return new ResponseEntity<String>(msg, HttpStatus.BAD_REQUEST);
}
String ceSubject = headers.get("ce-subject");
String msg = "Detected change in Cloud Storage bucket: " + ceSubject;
System.out.println(msg);
return new ResponseEntity<String>(msg, HttpStatus.OK);
}
}Node.js
const express = require('express');
const app = express();
app.use(express.json());
app.post('/', (req, res) => {
if (!req.header('ce-subject')) {
return res
.status(400)
.send('Bad Request: missing required header: ce-subject');
}
console.log(
`Detected change in Cloud Storage bucket: ${req.header('ce-subject')}`
);
return res
.status(200)
.send(
`Detected change in Cloud Storage bucket: ${req.header('ce-subject')}`
);
});
module.exports = app;Go
// Sample audit_storage is a Cloud Run service which handles Cloud Audit Log events with Cloud Storage data.
package main
import (
"fmt"
"log"
"net/http"
"os"
)
// HelloEventsStorage receives and processes a Cloud Audit Log event with Cloud Storage data.
func HelloEventsStorage(w http.ResponseWriter, r *http.Request) {
s := fmt.Sprintf("Detected change in Cloud Storage bucket: %s", string(r.Header.Get("Ce-Subject")))
log.Printf(s)
fmt.Fprintln(w, s)
}
C#
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
public class Startup
{
public void ConfigureServices(IServiceCollection services)
{
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env, ILogger<Startup> logger)
{
if (env.IsDevelopment())
{
app.UseDeveloperExceptionPage();
}
logger.LogInformation("Service is starting...");
app.UseRouting();
app.UseEndpoints(endpoints =>
{
endpoints.MapPost("/", async context =>
{
logger.LogInformation("Handling HTTP POST");
var ceSubject = context.Request.Headers["ce-subject"];
logger.LogInformation($"ce-subject: {ceSubject}");
if (string.IsNullOrEmpty(ceSubject))
{
context.Response.StatusCode = 400;
await context.Response.WriteAsync("Bad Request: expected header Ce-Subject");
return;
}
await context.Response.WriteAsync($"GCS CloudEvent type: {ceSubject}");
});
});
}
}
Pub/Sub
L'exemple de code montre comment lire des événements Pub/Sub dans un service déployé sur Cloud Run.
Python
@app.route('/', methods=['POST'])
def index():
data = request.get_json()
if not data:
msg = 'no Pub/Sub message received'
print(f'error: {msg}')
return f'Bad Request: {msg}', 400
if not isinstance(data, dict) or 'message' not in data:
msg = 'invalid Pub/Sub message format'
print(f'error: {msg}')
return f'Bad Request: {msg}', 400
pubsub_message = data['message']
name = 'World'
if isinstance(pubsub_message, dict) and 'data' in pubsub_message:
name = base64.b64decode(pubsub_message['data']).decode('utf-8').strip()
resp = f"Hello, {name}! ID: {request.headers.get('ce-id')}"
print(resp)
return (resp, 200)Java
import com.example.cloudrun.eventpojos.PubSubBody;
import java.util.Base64;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class EventController {
@RequestMapping(value = "/", method = RequestMethod.POST)
public ResponseEntity<String> receiveMessage(
@RequestBody PubSubBody body, @RequestHeader Map<String, String> headers) {
// Get PubSub message from request body.
PubSubBody.PubSubMessage message = body.getMessage();
if (message == null) {
String msg = "No Pub/Sub message received.";
System.out.println(msg);
return new ResponseEntity<String>(msg, HttpStatus.BAD_REQUEST);
}
String data = message.getData();
if (data == null || data.isEmpty()) {
String msg = "Invalid Pub/Sub message format.";
System.out.println(msg);
return new ResponseEntity<String>(msg, HttpStatus.BAD_REQUEST);
}
String name =
!StringUtils.isEmpty(data) ? new String(Base64.getDecoder().decode(data)) : "World";
String ceId = headers.getOrDefault("ce-id", "");
String msg = String.format("Hello, %s! ID: %s", name, ceId);
System.out.println(msg);
return new ResponseEntity<String>(msg, HttpStatus.OK);
}
}Node.js
const express = require('express');
const {
toMessagePublishedData,
} = require('@google/events/cloud/pubsub/v1/MessagePublishedData');
const app = express();
app.use(express.json());
app.post('/', (req, res) => {
if (!req.body) {
const errorMessage = 'no Pub/Sub message received';
res.status(400).send(`Bad Request: ${errorMessage}`);
console.log(`Bad Request: ${errorMessage}`);
return;
}
if (!req.body.message) {
const errorMessage = 'invalid Pub/Sub message format';
res.status(400).send(`Bad Request: ${errorMessage}`);
console.log(`Bad Request: ${errorMessage}`);
return;
}
// Cast to MessagePublishedEvent for IDE autocompletion
const pubSubMessage = toMessagePublishedData(req.body);
const name =
pubSubMessage.message && pubSubMessage.message.data
? Buffer.from(pubSubMessage.message.data, 'base64').toString().trim()
: 'World';
const result = `Hello, ${name}! ID: ${req.get('ce-id') || ''}`;
console.log(result);
res.send(result);
});
module.exports = app;Go
// Sample pubsub is a Cloud Run service which handles Pub/Sub messages.
package main
import (
"encoding/json"
"fmt"
"log"
"net/http"
"os"
)
// PubSubMessage is the payload of a Pub/Sub event.
// See the documentation for more details:
// https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
type PubSubMessage struct {
Message struct {
Data []byte `json:"data,omitempty"`
ID string `json:"id"`
} `json:"message"`
Subscription string `json:"subscription"`
}
// HelloEventsPubSub receives and processes a Pub/Sub push message.
func HelloEventsPubSub(w http.ResponseWriter, r *http.Request) {
var e PubSubMessage
if err := json.NewDecoder(r.Body).Decode(&e); err != nil {
http.Error(w, "Bad HTTP Request", http.StatusBadRequest)
log.Printf("Bad HTTP Request: %v", http.StatusBadRequest)
return
}
name := string(e.Message.Data)
if name == "" {
name = "World"
}
s := fmt.Sprintf("Hello, %s! ID: %s", name, string(r.Header.Get("Ce-Id")))
log.Printf(s)
fmt.Fprintln(w, s)
}
C#
using CloudNative.CloudEvents;
using CloudNative.CloudEvents.AspNetCore;
using Google.Events.Protobuf.Cloud.PubSub.V1;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
public class Startup
{
public void ConfigureServices(IServiceCollection services)
{
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env, ILogger<Startup> logger)
{
if (env.IsDevelopment())
{
app.UseDeveloperExceptionPage();
}
logger.LogInformation("Service is starting...");
app.UseRouting();
app.UseEndpoints(endpoints =>
{
endpoints.MapPost("/", async context =>
{
var formatter = CloudEventFormatterAttribute.CreateFormatter(typeof(MessagePublishedData));
var cloudEvent = await context.Request.ToCloudEventAsync(formatter);
logger.LogInformation("Received CloudEvent\n" + GetEventLog(cloudEvent));
var messagePublishedData = (MessagePublishedData) cloudEvent.Data;
var pubSubMessage = messagePublishedData.Message;
if (pubSubMessage == null)
{
context.Response.StatusCode = 400;
await context.Response.WriteAsync("Bad request: Invalid Pub/Sub message format");
return;
}
var data = pubSubMessage.Data;
logger.LogInformation($"Data: {data.ToBase64()}");
var name = data.ToStringUtf8();
logger.LogInformation($"Extracted name: {name}");
var id = context.Request.Headers["ce-id"];
await context.Response.WriteAsync($"Hello {name}! ID: {id}");
});
});
}
private string GetEventLog(CloudEvent cloudEvent)
{
return $"ID: {cloudEvent.Id}\n"
+ $"Source: {cloudEvent.Source}\n"
+ $"Type: {cloudEvent.Type}\n"
+ $"Subject: {cloudEvent.Subject}\n"
+ $"DataSchema: {cloudEvent.DataSchema}\n"
+ $"DataContentType: {cloudEvent.DataContentType}\n"
+ $"Time: {cloudEvent.Time?.UtcDateTime:yyyy-MM-dd'T'HH:mm:ss.fff'Z'}\n"
+ $"SpecVersion: {cloudEvent.SpecVersion}\n"
+ $"Data: {cloudEvent.Data}";
}
}