Milesoft Pub/Sub Extension (pubsub-ext)

1. Overview & Purpose

The Pub/Sub Extension (pubsub-ext) is a unified, highly optimized, and type-safe wrapper library around the official Google Cloud Pub/Sub Java SDK designed specifically for the Milesoft ecosystem. It standardizes messaging patterns across all downstream applications and microservices by providing standard serialization, message wrapping, and seamless asynchronous publish/subscribe flows.

Key Benefits & Core Capabilities:

• Type-Safe Serialization: Automatic translation between domain-specific Java POJOs and JSON-encoded Google Cloud PubsubMessage payloads using a factory-managed Jackson ObjectMapper.
• Asynchronous Publishing: High-throughput async message dispatch via PubSubTopicService with native callbacks to track and log successes and retryable errors.
• Unified Push Subscription DTOs: Standardized envelope objects (MessageWrapper and Message) conforming to Google Cloud Pub/Sub push subscription HTTP request structures.
• Base64 Message Transmutation: Instant, error-resilient extraction and decoding of inbound base64-encoded push message data fields into strong-typed domain objects.

---

2. Installation & Dependency Configuration

The Pub/Sub Extension is hosted securely inside our Google Cloud Artifact Registry repository.

To configure your application to resolve and pull this dependency:

2.1 Gradle Setup (build.gradle)

Apply the Google Cloud Artifact Registry plugin and configure the Maven repository block:

plugins {
    id "com.google.cloud.artifactregistry.gradle-plugin" version "2.2.5"
    id "java"
}

repositories {
    mavenLocal()
    mavenCentral()
    maven {
        url "artifactregistry://us-west2-maven.pkg.dev/milesoft-repo/repo-maven"
    }
}

dependencies {
    // Milesoft Commons foundation (required by pubsub-ext)
    implementation "io.milesoft:commons:2.0.1"
    
    // Pub/Sub Integration Extension
    implementation "io.milesoft:pubsub-ext:2.0.1"
}

---

3. Configuring Pub/Sub in Spring Boot

To integrate PubSubTopicService within your Spring Boot microservice, define individual strongly typed beans for each topic.

3.1 Externalize Application Properties

Configure your GCP project ID and topic identifiers in your properties file (e.g., application.yml or application.properties):

pubSub:
  project: "your-gcp-project"
  topic:
    userAdded: "prod-user-added"
    accountSaved: "prod-account-saved"

3.2 Secure Bean Configuration

Define the PubSubTopicService beans inside a @Configuration class to enable seamless @Autowired constructor injection across your business logic layer:

import static com.google.common.base.Preconditions.checkArgument;

import io.milesoft.pubsub.services.PubSubTopicService;
import io.milesoft.pubsub.v1.dto.UserAdded;
import io.milesoft.pubsub.v1.dto.AccountSaved;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class PubSubConfig {

    private final String project;
    private final String topicUserAdded;
    private final String topicAccountSaved;

    public PubSubConfig(
            @Value("${pubSub.project}") String project,
            @Value("${pubSub.topic.userAdded}") String topicUserAdded,
            @Value("${pubSub.topic.accountSaved}") String topicAccountSaved) {

        checkArgument(StringUtils.isNotBlank(project), "project must not be blank");
        checkArgument(StringUtils.isNotBlank(topicUserAdded), "topicUserAdded must not be blank");
        checkArgument(StringUtils.isNotBlank(topicAccountSaved), "topicAccountSaved must not be blank");

        this.project = project;
        this.topicUserAdded = topicUserAdded;
        this.topicAccountSaved = topicAccountSaved;
    }

    @Bean
    public PubSubTopicService<UserAdded> userAddedTopicService() {
        return new PubSubTopicService<>(project, topicUserAdded, UserAdded.class);
    }

    @Bean
    public PubSubTopicService<AccountSaved> accountSavedTopicService() {
        return new PubSubTopicService<>(project, topicAccountSaved, AccountSaved.class);
    }
}

---

4. Domain & Data Types

The library defines standardized models for processing Google Cloud Pub/Sub Push Subscription payloads.

4.1 Inbound Message DTO (Message)

Represents the inner Pub/Sub message metadata and payload pushed to HTTP endpoints.

import jakarta.validation.constraints.NotBlank;
import java.time.ZonedDateTime;
import lombok.Getter;
import lombok.Setter;

@Getter
@Setter
public class Message {
    @NotBlank
    private String messageId;      // Unique identifier assigned by GCP Pub/Sub
    
    private ZonedDateTime publishTime; // Timestamp when the message was published
    
    @NotBlank
    private String data;          // Base64-encoded serialized JSON payload
}

4.2 Push Subscription Envelope (MessageWrapper)

An outer envelope wrapping the incoming Pub/Sub message, designed to align perfectly with GAE/Cloud Run HTTP Push targets.

import jakarta.validation.Valid;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import lombok.Getter;
import lombok.Setter;

@Getter
@Setter
public class MessageWrapper {
    @NotNull
    @Valid
    private Message message;      // The nested Pub/Sub message body
    
    @NotBlank
    private String subscription;   // Full subscription path (e.g. projects/.../subscriptions/...)
}

---

5. API Reference Guide

5.1 PubSubTopicService<M>

A thread-safe, generic orchestration service targeting a specific GCP Pub/Sub topic.

Constructor:

public PubSubTopicService(String project, String topic, Class<M> messageClass, Module... modules)

• project: The GCP Project ID.
• topic: The target Pub/Sub topic name.
• messageClass: The target Java payload class for serialization and type safety.
• modules: Optional Jackson modules to customize the underlying ObjectMapper behavior.

Core Method:

Method Signature	Return Type	Description
publishAsync(M message)	void	Encodes and publishes the custom payload object to GCP asynchronously. Standard callbacks log success or intercept/log ApiException details on failure.

---

5.2 MessageTransformer<M>

A core transformer class responsible for marshaling and unmarshaling message payloads.

Constructor:

public MessageTransformer(Class<M> messageClass, Module... modules)

• messageClass: The Java payload class to serialize to or deserialize from.
• modules: Optional Jackson modules for customization.

Core Methods:

Method Signature	Return Type	Description
toExternal(M message)	PubsubMessage	Serializes a custom domain object into standard JSON, wraps it as binary ByteString, and configures the "Content-Type": "application/json" metadata attribute.
fromDto(MessageWrapper dto)	M	Decodes the base64-encoded data field of the incoming MessageWrapper and parses it back into a strong-typed Java object.

---

6. Practical Real-World Code Examples

These integration examples represent standard, battle-tested publishing and subscribing patterns across the Milesoft ecosystem.

Example A: Asynchronous Event Publishing

This service demonstrates constructing and asynchronously publishing an event when a new user is added or updated.

import io.milesoft.pubsub.services.PubSubTopicService;
import io.milesoft.pubsub.v1.dto.UserAdded;
import org.springframework.stereotype.Service;

@Service
public class AccountPubSubService {

    private final PubSubTopicService<UserAdded> userAddedTopicService;

    public AccountPubSubService(PubSubTopicService<UserAdded> userAddedTopicService) {
        this.userAddedTopicService = userAddedTopicService;
    }

    public void emitUserAdded(String appId, String accountId, String email) {
        // Construct standard event DTO
        final UserAdded event = new UserAdded();
        event.setAppId(appId);
        event.setAccountId(accountId);
        event.setEmail(email);

        // Dispatches the message asynchronously to Google Cloud Pub/Sub
        userAddedTopicService.publishAsync(event);
    }
}

Example B: Secure Subscriber Web Controller

This example demonstrates setting up an HTTP endpoint inside a downstream application to securely receive GCP Push Subscription payloads, decode them, and process the actual payload object.

import io.milesoft.pubsub.dto.MessageWrapper;
import io.milesoft.pubsub.transformers.MessageTransformer;
import io.milesoft.pubsub.v1.dto.UserAdded;
import jakarta.validation.Valid;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequestMapping("/api/v2/subscriptions")
public class UserSubscriptionController {

    private final MessageTransformer<UserAdded> userAddedTransformer;

    public UserSubscriptionController() {
        // Initialize transformer for the specific target DTO type
        this.userAddedTransformer = new MessageTransformer<>(UserAdded.class);
    }

    @PostMapping("/user-added")
    public ResponseEntity<Void> handleUserAddedPushNotification(@Valid @RequestBody MessageWrapper payload) {
        try {
            // 1. Decode base64 payload and deserialize to strong-typed POJO
            final UserAdded event = userAddedTransformer.fromDto(payload);
            
            log.info("Received user-added event for email: {}", event.getEmail());
            
            // 2. Perform business logic (e.g. provision resources, send welcome emails)
            // ...

            // 3. Return 204 OK/No Content to acknowledge successful processing to GCP Pub/Sub
            return ResponseEntity.status(HttpStatus.NO_CONTENT).build();
        } catch (Exception e) {
            log.error("Failed to process inbound Pub/Sub push notification", e);
            
            // Return a 500/Internal Server Error so GCP retries delivery
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
}