Handling Requests from Google Pub/Sub Push Subscriptions in Spring Boot Cloud Run Applications
Cloud Run is a serverless application platform, which means apps deployed there can’t be expected to be running all the time and therefore won’t work with the default “pull” subscription type.
A push subscription calls an HTTP endpoint, which is how a Cloud Run service is triggered. There are examples of endpoints built for this purpose in the docs, but the Java example is very basic. Here’s a decent way of writing a request handler that will work with the Spring framework.
Say you have (in Terraform) a topic for events in the lives of characters from the critically-acclaimed Showtime show, Yellowjackets, and a subscription that filters on actions taken by Lottie:
resource "google_pubsub_subscription" "example" {
name = "yellowjackets-events---lottie"
topic = data.google_pubsub_topic.yellowjackets_events.name
filter = "attributes.character = \"Lottie\""
push_config {
push_endpoint = "${google_cloud_run_service.app.status[0].url}/lottie-events"
oidc_token {
service_account_email = data.google_service_account.pubsub_invoker.email
}
}
}Message bodies might look like this:
{
"date": "1996-10-01",
"creepyStatement": "We won't be hungry much longer",
"visionType": "AERIAL_INFERNO"
}The Spring Boot request handler could be written as such:
@RestController
@Slf4j
public class PubSubPushController { @RequestMapping(value = "/lottie-events",
method = RequestMethod.POST,
produces = MediaType.APPLICATION_JSON_VALUE,
consumes = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity<?> receiveLottieEvent(
HttpServletRequest req
) throws Exception {
// (using Jackson)
JsonNode bodyNode = objectMapper.readTree(req.getReader());
String base64Data = bodyNode.get("message")
.get("data").asText();
byte[] data = Base64.getDecoder().decode(base64Data);
LottieEvent event = objectMapper.readValue(
data, LottieEvent.class);
log.info(event.getCreepyStatement()); return new ResponseEntity<>(HttpStatus.OK);
}
}
But I find that a bit noisy for a controller method. It has to read the raw pub/sub message, get the message.data property as text — and that’s a Base64-encoded string, so you have to decode that, and so on. We could abstract that away so we can immediately get what we need from the controller method parameter.
Let’s make a container for the method parameter, where T could be a LottieEvent:
@Data // (Lombok)
public class PubSubPushRequest<T> { private final T body;
private final Map<String, Object> attributes;
// you might want headers here too
}
Then, implement a HandlerMethodArgumentResolver:
@Component
public class PubSubArgumentResolver
implements HandlerMethodArgumentResolver { private final ObjectMapper objectMapper; public PubSubArgumentResolver(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
} @Override
public boolean supportsParameter(MethodParameter parameter) {
return PubSubPushRequest.class
.isAssignableFrom(parameter.getParameterType());
} @Override
public Object resolveArgument(
MethodParameter parameter,
ModelAndViewContainer mavContainer,
NativeWebRequest webRequest,
WebDataBinderFactory binderFactory
) throws Exception {
HttpServletRequest req = webRequest
.getNativeRequest(HttpServletRequest.class);
ParameterizedType dataType = (ParameterizedType)
parameter.getGenericParameterType();
Class<?> dataClass = (Class<?>)
dataType.getActualTypeArguments()[0];
JsonNode bodyNode = objectMapper.readTree(req.getReader());
String base64Data = bodyNode.get("message").get("data")
.asText();
byte[] data = Base64.getDecoder().decode(base64Data);
Object dataObj = objectMapper.readValue(data, dataClass); Map<String, Object> attributes = objectMapper.convertValue(
bodyNode.get("message").get("attributes"),
new TypeReference<Map<String, Object>>() {}); return new PubSubPushRequest<>(dataObj, attributes);
}
}
…and register it:
@Configuration
public class WebMvcConfiguration implements WebMvcConfigurer { private final PubSubArgumentResolver pubSubArgumentResolver; public WebMvcConfiguration(
PubSubArgumentResolver pubSubArgumentResolver
) {
this.pubSubArgumentResolver = pubSubArgumentResolver;
} @Override
public void addArgumentResolvers(
List<HandlerMethodArgumentResolver> resolvers
) {
resolvers.add(pubSubArgumentResolver);
}
}
Finally, we can update the controller method:
// ...
public ResponseEntity<?> receiveLottieEvent(
PubSubPushRequest<LottieEvent> pushRequest
) throws Exception { LottieEvent event = pushRequest.getBody();
// ...
Much better! And PubSubPushRequest can be used for other events from other subscriptions. That is, if you’re sure you really want to know what these ladies are up to.
