RetryPubSub implements BackgroundFunction<PubSubMessage> { @Override public void accept(PubSubMessage message, Context context) { String bodyJson = new String( Base64.getDecoder().decode(message.getData()), StandardCharsets.UTF_8); JsonElement bodyElement = gson.fromJson(bodyJson, JsonElement.class); boolean retry = false; if (bodyElement != null && bodyElement.isJsonObject()) { JsonObject body = bodyElement.getAsJsonObject(); if (body.has("retry") && body.get("retry").getAsBoolean()) { retry = true; } } if (retry) { throw new RuntimeException("Retrying..."); } else { logger.info("Not retrying..."); } }  @Override void accept(PubSubMessage message, Context context);  }
@Test(expected = RuntimeException.class) public void retryPubsub_handlesRetryMsg() throws IOException { String data = gson.toJson(Map.of("retry", true)); String encodedData = new String( Base64.getEncoder().encode(data.getBytes(StandardCharsets.UTF_8)), StandardCharsets.UTF_8); PubSubMessage pubsubMessage = new PubSubMessage(); pubsubMessage.setData(encodedData); new RetryPubSub().accept(pubsubMessage, null); }
@Test public void retryPubsub_handlesStopMsg() throws IOException { String data = gson.toJson(Map.of("retry", false)); String encodedData = new String( Base64.getEncoder().encode(data.getBytes(StandardCharsets.UTF_8)), StandardCharsets.UTF_8); PubSubMessage pubsubMessage = new PubSubMessage(); pubsubMessage.setData(encodedData); new RetryPubSub().accept(pubsubMessage, null); String logMessage = LOG_HANDLER.getStoredLogRecords().get(0).getMessage(); assertThat("Not retrying...").isEqualTo(logMessage); }
@Test public void retryPubsub_handlesEmptyMsg() throws IOException { PubSubMessage pubsubMessage = new PubSubMessage(); pubsubMessage.setData(""); new RetryPubSub().accept(pubsubMessage, null); String logMessage = LOG_HANDLER.getStoredLogRecords().get(0).getMessage(); assertThat("Not retrying...").isEqualTo(logMessage); }