Advanced

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.spi.cluster.hazelcast.HazelcastClusterManager;

public class ClusteredVerticle extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    vertx.eventBus().consumer("cluster.hello", msg -> {
      System.out.println("Received on node " + getNodeId() + ": " + msg.body());
      // => Output: Received on node node-1: Hello from node-2
      // => This consumer is reachable from ALL nodes in the cluster
      msg.reply("Pong from " + getNodeId());
    });

    // Send to a consumer on ANY node (round-robin across cluster)
    vertx.eventBus()
      .request("cluster.hello", "Hello from " + getNodeId())
      .onSuccess(reply -> {
        System.out.println("Reply: " + reply.body());
        // => Output: Reply: Pong from node-2
        // => Reply came from a different JVM node
      });

    startPromise.complete();
  }

  private String getNodeId() {
    return System.getProperty("node.id", "node-1");
    // => Node identifier from system property; set at startup
  }

  public static void main(String[] args) {
    HazelcastClusterManager mgr = new HazelcastClusterManager();
    // => Hazelcast cluster manager; discovers peers via multicast or TCP/IP

    VertxOptions options = new VertxOptions()
      .setClusterManager(mgr)
      // => Enables cluster mode; single-node operation if this is omitted
      .setEventBusOptions(new io.vertx.core.eventbus.EventBusOptions()
        .setHost(System.getProperty("node.host", "localhost"))
        // => Host this node advertises to cluster; must be externally reachable
        .setPort(Integer.parseInt(System.getProperty("cluster.port", "18080"))));
        // => Port for inter-node event bus communication

    Vertx.clusteredVertx(options)
      // => Creates a clustered Vert.x instance; async (cluster join takes time)
      // => Returns Future<Vertx> (not Vertx.vertx() which is synchronous)
      .onSuccess(vertx -> {
        System.out.println("Joined cluster; deploying verticle...");
        vertx.deployVerticle(new ClusteredVerticle());
      })
      .onFailure(err -> {
        System.err.println("Failed to join cluster: " + err);
        System.exit(1);
        // => Exit if clustering fails; orchestration platform will restart
      });
  }
}

import io.vertx.core.AbstractVerticle;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.spi.cluster.hazelcast.HazelcastClusterManager;

public class HighAvailabilityDemo extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    System.out.println("HA Verticle started on node: "
      + System.getProperty("node.id", "unknown"));
    // => Output: HA Verticle started on node: node-1
    // => If node-1 dies, this verticle restarts on another available node

    vertx.eventBus().consumer("ha.ping", msg -> {
      msg.reply("pong from " + System.getProperty("node.id", "unknown"));
      // => This consumer is always available as long as cluster has nodes
    });

    startPromise.complete();
  }

  public static void main(String[] args) {
    HazelcastClusterManager mgr = new HazelcastClusterManager();

    Vertx.clusteredVertx(new VertxOptions().setClusterManager(mgr))
      .onSuccess(vertx -> {
        DeploymentOptions haOptions = new DeploymentOptions()
          .setHa(true);
          // => Mark this deployment for HA failover
          // => If this node leaves the cluster, Vert.x redeploys the verticle on another node
          // => Non-HA verticles are NOT redeployed on node failure

        vertx.deployVerticle(HighAvailabilityDemo.class.getName(), haOptions)
          .onSuccess(id -> {
            System.out.println("HA verticle deployed: " + id);
            // => Deployment recorded in cluster-wide registry
            // => Other nodes know to take over if this node fails
          });
      });
  }
}
// To test HA:
// 1. Start two nodes: -Dnode.id=node-1 and -Dnode.id=node-2
// 2. Kill node-1 (Ctrl+C or kill -9)
// 3. Observe: node-2 redeploys the HA verticle automatically

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.shareddata.Lock;
import io.vertx.core.shareddata.SharedData;
import io.vertx.spi.cluster.hazelcast.HazelcastClusterManager;

public class SharedDataDemo extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    SharedData sd = vertx.sharedData();
    // => SharedData API; cluster-wide when running in clustered mode

    // Cluster-wide distributed map
    sd.<String, String>getClusterWideMap("session-store")
      // => Hazelcast-backed distributed map; accessible from all cluster nodes
      .compose(map -> {
        return map.put("session-abc-123", "user-42")
          // => Write session to cluster-wide map; visible on all nodes
          .compose(v -> map.get("session-abc-123"))
          // => Read back from map (may come from another node)
          .onSuccess(value -> {
            System.out.println("Session value: " + value);
            // => Output: Session value: user-42
            // => Retrieved from Hazelcast distributed map
          });
      })
      .onFailure(err -> System.err.println("Map error: " + err));

    // Distributed lock for cross-node coordination
    sd.getLockWithTimeout("payment-processor-lock", 3000)
      // => Try to acquire distributed lock; wait max 3000ms
      // => Lock is cluster-wide; only ONE node holds it at a time
      .onSuccess(lock -> {
        System.out.println("Acquired distributed lock; processing payment");
        // => Only this node processes the payment; prevents double-processing

        try {
          processPayment();
          // => Critical section: single-node execution guaranteed
        } finally {
          lock.release();
          // => ALWAYS release lock; use try-finally to ensure release even on error
          // => Other nodes waiting for this lock can now proceed
          System.out.println("Lock released");
        }

        startPromise.complete();
      })
      .onFailure(err -> {
        System.err.println("Could not acquire lock (timeout or error): " + err);
        // => Another node holds the lock; retry or fail gracefully
        startPromise.complete();
      });
  }

  private void processPayment() {
    System.out.println("Processing payment exclusively...");
    // => In production: idempotent payment processing logic
  }

  public static void main(String[] args) {
    Vertx.clusteredVertx(new VertxOptions()
      .setClusterManager(new HazelcastClusterManager()))
      .onSuccess(vertx -> vertx.deployVerticle(new SharedDataDemo()));
  }
}

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.ext.web.client.WebClient;
import io.vertx.servicediscovery.Record;
import io.vertx.servicediscovery.ServiceDiscovery;
import io.vertx.servicediscovery.ServiceDiscoveryOptions;
import io.vertx.servicediscovery.types.HttpEndpoint;

public class ServiceDiscoveryDemo extends AbstractVerticle {

  private ServiceDiscovery discovery;
  // => Service discovery instance; backed by cluster-wide map in clustered mode

  @Override
  public void start(Promise<Void> startPromise) {
    discovery = ServiceDiscovery.create(vertx, new ServiceDiscoveryOptions()
      .setBackendConfiguration(new io.vertx.core.json.JsonObject()
        .put("backend-name", "io.vertx.servicediscovery.backend.hazelcast.HazelcastBackend")));
        // => Hazelcast backend: service records shared across cluster nodes

    // Register THIS service in the discovery registry
    Record record = HttpEndpoint.createRecord(
      "user-service",
      // => Service name used for lookup
      "user-service.internal",
      // => Hostname/IP of this service instance
      8080,
      // => Port number
      "/api/users");
      // => Root path of this service's API

    discovery.publish(record)
      .onSuccess(publishedRecord -> {
        System.out.println("Registered as: " + publishedRecord.getName()
          + " (id: " + publishedRecord.getRegistration() + ")");
        // => Output: Registered as: user-service (id: a1b2c3d4-...)
        // => Other services can now discover this instance
        startPromise.complete();
      })
      .onFailure(startPromise::fail);
  }

  // Method to look up and call a discovered service
  void callUserService() {
    HttpEndpoint.getWebClient(discovery,
      new io.vertx.core.json.JsonObject().put("name", "user-service"))
      // => Looks up a healthy instance of "user-service" in the registry
      // => Returns a configured WebClient pointing to that instance
      .onSuccess(client -> {
        client.get("/api/users/42")
          .send()
          .onSuccess(response -> {
            System.out.println("User: " + response.bodyAsJsonObject());
            // => HTTP call to dynamically discovered service instance
            ServiceDiscovery.releaseServiceObject(discovery, client);
            // => Release client when done; decrements usage counter for load balancing
          });
      })
      .onFailure(err -> System.err.println("Service unavailable: " + err));
      // => Fails if no healthy "user-service" instances are registered
  }

  @Override
  public void stop(Promise<Void> stopPromise) {
    discovery.close();
    // => Unregister all published records and close discovery
    // => Prevents stale entries in the discovery registry after shutdown
    stopPromise.complete();
  }

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new ServiceDiscoveryDemo());
  }
}

// Note: requires protobuf-maven-plugin or similar to generate from .proto files
// Proto definition (user.proto):
// syntax = "proto3";
// service UserService {
//   rpc GetUser (GetUserRequest) returns (UserResponse);
//   rpc ListUsers (ListUsersRequest) returns (stream UserResponse);
// }
// message GetUserRequest { string id = 1; }
// message UserResponse { string id = 1; string name = 2; string email = 3; }
// message ListUsersRequest { int32 page = 1; }

import io.grpc.stub.StreamObserver;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.grpc.server.GrpcServer;
import io.vertx.grpc.server.GrpcServerRequest;

public class GrpcServerVerticle extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    GrpcServer grpcServer = GrpcServer.server(vertx);
    // => Creates a Vert.x gRPC server (non-blocking, event loop based)

    // Register unary RPC handler (one request → one response)
    grpcServer.callHandler(UserServiceGrpc.getGetUserMethod(),
      // => Bind to the generated gRPC method descriptor
      request -> {
        // => request is GrpcServerRequest<GetUserRequest, UserResponse>
        request.handler(getUserRequest -> {
          // => Fires when client sends the request message
          String id = getUserRequest.getId();
          // => Access protobuf message field

          System.out.println("gRPC GetUser called for: " + id);
          // => Output: gRPC GetUser called for: user-42

          UserResponse response = UserResponse.newBuilder()
            .setId(id)
            .setName("Alice")
            .setEmail("alice@example.com")
            .build();
            // => Build protobuf response message using generated builder

          request.response()
            .end(response);
            // => Send single response and close the stream
        });
      });

    // Register server-streaming RPC (one request → multiple responses)
    grpcServer.callHandler(UserServiceGrpc.getListUsersMethod(), request -> {
      request.handler(listRequest -> {
        int page = listRequest.getPage();
        // => Page number from request

        var response = request.response();
        // => Streaming response: write multiple messages

        for (int i = (page - 1) * 10; i < page * 10; i++) {
          // => Send 10 users per page
          response.write(UserResponse.newBuilder()
            .setId("user-" + i)
            .setName("User " + i)
            .build());
            // => Each write() sends one message in the stream
        }

        response.end();
        // => End the server stream; client receives all messages then EOF
      });
    });

    vertx.createHttpServer()
      .requestHandler(grpcServer)
      // => gRPC server handles HTTP/2 upgrade internally
      .listen(9090)
      .onSuccess(s -> {
        System.out.println("gRPC server on port 9090");
        // => Output: gRPC server on port 9090
        startPromise.complete();
      })
      .onFailure(startPromise::fail);
  }

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new GrpcServerVerticle());
  }
}
// Stub classes (UserServiceGrpc, UserResponse, etc.) generated from user.proto
// by the protobuf-maven-plugin or gradle protobuf plugin

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.Router;

// Native image requires:
// 1. Reflect on all dynamically loaded classes → native-image.properties
// 2. Resource files included in image → resource-config.json
// 3. Proxy interfaces registered → proxy-config.json

// native-image.properties (in META-INF/native-image/):
// Args = --initialize-at-build-time=io.netty \
//        -H:ReflectionConfigurationFiles=${.}/reflect-config.json \
//        -H:ResourceConfigurationFiles=${.}/resource-config.json

// reflect-config.json (partial):
// [
//   { "name": "io.vertx.core.impl.launcher.commands.RunCommand", "allDeclaredMethods": true },
//   { "name": "com.example.MainVerticle", "allDeclaredMethods": true, "allDeclaredConstructors": true }
// ]

public class NativeImageVerticle extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    long startupTime = System.currentTimeMillis();
    // => Capture startup time; native image starts in <100ms vs 2-5s for JVM

    Router router = Router.router(vertx);

    router.get("/hello").handler(ctx -> {
      ctx.json(new JsonObject()
        .put("message", "Hello from native image!")
        .put("uptime", System.currentTimeMillis() - startupTime));
        // => Output: {"message":"Hello from native image!","uptime":45}
        // => 45ms uptime demonstrates sub-100ms startup in native mode
    });

    // Avoid dynamic class loading in native image
    // WRONG: Class.forName("com.example.PluginA") - breaks native image
    // RIGHT: Register all implementations at build time via reflect-config.json

    // Use System.getProperty for runtime config (environment variables work too)
    int port = Integer.parseInt(System.getProperty("port", "8080"));
    // => Properties work in native image; no reflection required

    vertx.createHttpServer()
      .requestHandler(router)
      .listen(port)
      .onSuccess(s -> {
        System.out.println("Native server started in ~"
          + (System.currentTimeMillis() - startupTime) + "ms on port " + port);
        // => Output: Native server started in ~45ms on port 8080
        startPromise.complete();
      })
      .onFailure(startPromise::fail);
  }

  public static void main(String[] args) {
    // Disable classloader isolation (native image has no dynamic classloader)
    VertxOptions options = new VertxOptions()
      .setPreferNativeTransport(true);
      // => Use native epoll/kqueue transport when available
      // => Already compiled in native image; no runtime detection needed

    Vertx vertx = Vertx.vertx(options);
    vertx.deployVerticle(new NativeImageVerticle());
  }
}
// Build command (GraalVM 22+):
// native-image --no-fallback \
//   -cp target/my-app.jar \
//   -H:Name=my-vertx-app \
//   com.example.NativeImageVerticle
//
// Run:
// ./my-vertx-app -Dport=8080
// => Starts in <100ms; uses ~50MB RAM vs ~250MB for JVM

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.net.JksOptions;
import io.vertx.core.net.PemKeyCertOptions;
import io.vertx.ext.web.Router;

public class HttpsVerticle extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    Router router = Router.router(vertx);

    router.get("/secure").handler(ctx -> {
      ctx.response()
        .putHeader("Strict-Transport-Security", "max-age=31536000; includeSubDomains")
        // => HSTS header: tell browsers to always use HTTPS for this domain
        // => max-age=31536000 = 1 year
        .putHeader("X-Content-Type-Options", "nosniff")
        // => Prevent MIME type sniffing attacks
        .putHeader("X-Frame-Options", "DENY")
        // => Prevent clickjacking via iframes
        .end("Secure response");
    });

    HttpServerOptions tlsOptions = new HttpServerOptions()
      .setSsl(true)
      // => Enable TLS; required for HTTPS
      .setKeyStoreOptions(new JksOptions()
        .setPath("server-keystore.jks")
        // => Java KeyStore containing server certificate and private key
        // => Create with: keytool -genkeypair -alias server -keyalg RSA -keystore server.jks
        .setPassword(System.getenv("KEYSTORE_PASSWORD")))
        // => Keystore password from environment; never hardcode
      .addEnabledSecureTransportProtocol("TLSv1.3")
      // => Only allow TLS 1.3; disable older vulnerable versions
      .addEnabledSecureTransportProtocol("TLSv1.2")
      // => Allow TLS 1.2 for older client compatibility
      .removeEnabledSecureTransportProtocol("TLSv1.1")
      // => Disable TLS 1.1 (vulnerable to POODLE, BEAST attacks)
      .removeEnabledSecureTransportProtocol("TLSv1.0")
      // => Disable TLS 1.0 (deprecated by RFC 8996)
      .setUseAlpn(true);
      // => Enable ALPN (Application-Layer Protocol Negotiation)
      // => Allows HTTP/2 negotiation over TLS

    // Alternative: PEM format (more common in cloud/Kubernetes deployments)
    HttpServerOptions pemOptions = new HttpServerOptions()
      .setSsl(true)
      .setKeyCertOptions(new PemKeyCertOptions()
        .setCertPath("/etc/ssl/certs/server.crt")
        // => PEM certificate file (from Let's Encrypt, commercial CA, etc.)
        .setKeyPath("/etc/ssl/private/server.key"))
        // => Private key file; permissions should be 600 (owner read-only)
      .setUseAlpn(true);

    vertx.createHttpServer(tlsOptions)
      .requestHandler(router)
      .listen(443)
      // => Standard HTTPS port; use 8443 for non-root development
      .onSuccess(s -> {
        System.out.println("HTTPS server on port " + s.actualPort());
        // => Output: HTTPS server on port 443
        startPromise.complete();
      })
      .onFailure(startPromise::fail);
  }

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new HttpsVerticle());
  }
}

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.net.JksOptions;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.client.WebClient;

public class MutualTlsDemo extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    // mTLS Server: requires client certificates signed by trusted CA
    HttpServerOptions serverOptions = new HttpServerOptions()
      .setSsl(true)
      .setKeyStoreOptions(new JksOptions()
        .setPath("server-keystore.jks")
        // => Server's certificate and private key
        .setPassword(System.getenv("SERVER_KEYSTORE_PASS")))
      .setTrustStoreOptions(new JksOptions()
        .setPath("trusted-clients.jks")
        // => CA certificates of trusted clients
        // => Only clients with certs signed by these CAs are allowed
        .setPassword(System.getenv("TRUST_STORE_PASS")))
      .setClientAuth(io.vertx.core.http.ClientAuth.REQUIRED);
      // => REQUIRED: reject connections without valid client certificates
      // => REQUEST: accept but don't require client certs
      // => NONE: don't request client certs (default)

    Router router = Router.router(vertx);
    router.get("/internal/data").handler(ctx -> {
      // Client certificate is available after successful mTLS handshake
      javax.security.cert.X509Certificate[] certs =
        ctx.request().peerCertificates();
      // => Returns client's certificate chain; null if no client cert
      if (certs != null && certs.length > 0) {
        String clientDN = certs[0].getSubjectDN().getName();
        // => Distinguished Name from client cert: "CN=payment-service, O=MyOrg"
        System.out.println("Request from: " + clientDN);
        // => Output: Request from: CN=payment-service, O=MyOrg
      }
      ctx.json(new io.vertx.core.json.JsonObject().put("data", "internal-secret"));
    });

    vertx.createHttpServer(serverOptions)
      .requestHandler(router)
      .listen(8443)
      .onSuccess(s -> startPromise.complete())
      .onFailure(startPromise::fail);
  }

  void createMtlsClient() {
    // mTLS Client: presents its certificate to the server
    WebClient client = WebClient.create(vertx,
      new WebClientOptions()
        .setSsl(true)
        .setKeyStoreOptions(new JksOptions()
          .setPath("client-keystore.jks")
          // => Client's certificate and private key (presented to server)
          .setPassword(System.getenv("CLIENT_KEYSTORE_PASS")))
        .setTrustStoreOptions(new JksOptions()
          .setPath("trusted-servers.jks")
          // => Trusted server CA certificates (server cert must be signed by one)
          .setPassword(System.getenv("CLIENT_TRUST_PASS")))));
          // => Prevents connecting to untrusted/self-signed servers

    client.get(8443, "internal-service", "/internal/data")
      .send()
      .onSuccess(r -> System.out.println("mTLS response: " + r.bodyAsJsonObject()));
  }

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new MutualTlsDemo());
  }
}

import io.vertx.config.ConfigRetriever;
import io.vertx.config.ConfigRetrieverOptions;
import io.vertx.config.ConfigStoreOptions;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;

public class ProductionConfigVerticle extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    ConfigRetrieverOptions options = new ConfigRetrieverOptions()
      .addStore(new ConfigStoreOptions()
        .setType("file")
        .setFormat("json")
        .setConfig(new JsonObject()
          .put("path", "config/defaults.json")))
        // => Base defaults from bundled config file
      .addStore(new ConfigStoreOptions()
        .setType("env"))
        // => Environment variables override defaults
        // => DB_HOST overrides defaults.json "db.host"
        // => HTTP_PORT overrides defaults.json "http.port"
      .addStore(new ConfigStoreOptions()
        .setType("sys"));
        // => Java system properties (-Dkey=value) take highest precedence
        // => Useful for developer overrides during testing

    ConfigRetriever.create(vertx, options)
      .getConfig()
      .onSuccess(config -> {
        System.out.println("Config loaded:");
        // => Resolved config merges all three sources (env overrides file)

        int port = config.getInteger("HTTP_PORT", 8080);
        // => HTTP_PORT env var; default 8080
        String dbHost = config.getString("DB_HOST", "localhost");
        // => DB_HOST env var; default localhost
        int dbPort = config.getInteger("DB_PORT", 5432);
        String dbUser = config.getString("DB_USER", "postgres");
        String dbPass = config.getString("DB_PASSWORD", "");
        // => DB_PASSWORD: never set default in code; fail if missing in production
        String jwtSecret = config.getString("JWT_SECRET", "");
        // => JWT_SECRET: critical; validate non-empty before starting

        if (dbPass.isEmpty()) {
          System.err.println("DB_PASSWORD is required");
          startPromise.fail("DB_PASSWORD environment variable not set");
          // => Fail fast: missing required config prevents startup
          return;
        }

        System.out.printf("Starting: port=%d db=%s:%d%n", port, dbHost, dbPort);
        // => Output: Starting: port=8080 db=postgres-svc:5432
        // => In Docker/Kubernetes, DB_HOST is the service name (postgres-svc)

        startHttpServer(startPromise, port);
      })
      .onFailure(err -> {
        System.err.println("Config load failed: " + err);
        startPromise.fail(err);
      });
  }

  private void startHttpServer(Promise<Void> sp, int port) {
    vertx.createHttpServer()
      .requestHandler(ctx -> ctx.response().end("OK"))
      .listen(port)
      .onSuccess(s -> sp.complete())
      .onFailure(sp::fail);
  }

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new ProductionConfigVerticle());
  }
}
// Dockerfile:
// FROM eclipse-temurin:21-jre-alpine
// WORKDIR /app
// COPY target/my-app-fat.jar app.jar
// EXPOSE 8080
// ENV HTTP_PORT=8080
// CMD ["java", "-jar", "app.jar"]
//
// docker run -e DB_HOST=postgres-svc -e DB_PASSWORD=secret my-app

// This example shows the Maven/Docker build configuration rather than runtime code

// pom.xml additions for fat JAR:
// <plugin>
//   <groupId>io.reactiverse</groupId>
//   <artifactId>vertx-maven-plugin</artifactId>
//   <version>1.0.27</version>
//   <executions>
//     <execution>
//       <id>vmp</id>
//       <goals><goal>initialize</goal><goal>package</goal></goals>
//     </execution>
//   </executions>
//   <configuration>
//     <verticle>com.example.MainVerticle</verticle>
//     <!-- Verticle to deploy; embedded in MANIFEST.MF -->
//   </configuration>
// </plugin>

// Build command:
// mvn package -DskipTests
// => Creates target/my-app-1.0-fat.jar (~20MB bundled)

// Runtime verticle that demonstrates fat JAR patterns
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;

public class FatJarMainVerticle extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    System.out.println("Fat JAR verticle starting...");
    // => "Fat JAR" = all dependencies bundled; java -jar runs without classpath setup

    // Demonstrate reading build metadata from JAR manifest
    String version = getClass().getPackage().getImplementationVersion();
    // => Read version from MANIFEST.MF (set by vertx-maven-plugin)
    System.out.println("Version: " + (version != null ? version : "dev"));
    // => Output: Version: 1.0.0 (in production) or Version: dev (in IDE)

    int port = Integer.parseInt(System.getenv().getOrDefault("PORT", "8080"));
    // => Cloud platforms (Heroku, Fly.io) set PORT automatically

    vertx.createHttpServer()
      .requestHandler(req -> {
        if ("/health".equals(req.path())) {
          req.response().end("OK");
          // => Health check for container orchestration
        } else {
          req.response()
            .putHeader("Content-Type", "application/json")
            .end("{\"service\":\"my-app\",\"version\":\"" + version + "\"}");
        }
      })
      .listen(port)
      .onSuccess(s -> {
        System.out.println("Started on port " + s.actualPort());
        startPromise.complete();
      })
      .onFailure(startPromise::fail);
  }

  public static void main(String[] args) {
    // Override vertx thread pool sizes for container CPU allocation
    int cpuCount = Runtime.getRuntime().availableProcessors();
    // => Docker containers may have fractional CPU; detect actual allocation
    System.setProperty("vertx.eventLoopPoolSize", String.valueOf(cpuCount * 2));
    // => Set 2 event loop threads per CPU; tune based on I/O vs CPU ratio
    System.setProperty("vertx.workerPoolSize", String.valueOf(cpuCount * 20));
    // => Worker pool for executeBlocking; higher for blocking-heavy workloads

    Vertx.vertx().deployVerticle(new FatJarMainVerticle());
  }
}
// Dockerfile (multi-stage build):
// FROM maven:3.9-eclipse-temurin-21 AS build
// WORKDIR /build
// COPY pom.xml .
// RUN mvn dependency:go-offline  # Cache dependencies layer
// COPY src ./src
// RUN mvn package -DskipTests
//
// FROM eclipse-temurin:21-jre-alpine
// WORKDIR /app
// RUN adduser -D -h /app appuser && chown appuser:appuser /app
// USER appuser
// COPY --from=build /build/target/*-fat.jar app.jar
// EXPOSE 8080
// HEALTHCHECK --interval=10s CMD wget -qO- http://localhost:8080/health || exit 1
// ENTRYPOINT ["java", "-XX:+UseContainerSupport", "-jar", "app.jar"]
// => -XX:+UseContainerSupport: JVM respects Docker memory/CPU limits (Java 10+)

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.pgclient.PgBuilder;
import io.vertx.pgclient.PgConnectOptions;
import io.vertx.sqlclient.Pool;
import io.vertx.sqlclient.PoolOptions;
import io.vertx.sqlclient.Row;
import io.vertx.sqlclient.RowStream;
import io.vertx.sqlclient.SqlConnection;
import io.vertx.sqlclient.Tuple;

public class AdvancedSqlVerticle extends AbstractVerticle {

  private Pool pool;

  @Override
  public void start(Promise<Void> startPromise) {
    pool = PgBuilder.pool().using(vertx)
      .with(new PoolOptions().setMaxSize(10))
      .connectingTo(new PgConnectOptions()
        .setHost("localhost").setDatabase("myapp")
        .setUser("postgres").setPassword("secret"))
      .build();

    io.vertx.ext.web.Router router = io.vertx.ext.web.Router.router(vertx);

    // Streaming large result sets (avoid loading all rows into memory)
    router.get("/export/users").handler(ctx -> {
      pool.getConnection()
        .onSuccess(conn -> {
          ctx.response()
            .setChunked(true)
            // => Enable chunked transfer for streaming response
            .putHeader("Content-Type", "application/json");

          ctx.response().write("[");
          // => Start JSON array manually for streaming

          conn.prepare("SELECT id, name, email FROM users ORDER BY id")
            .onSuccess(stmt -> {
              RowStream<Row> stream = stmt.createStream(50);
              // => Stream rows from DB 50 at a time (fetch size)
              // => Prevents loading 1M rows into memory

              boolean[] first = {true};
              // => Track first row to avoid leading comma

              stream.handler(row -> {
                String prefix = first[0] ? "" : ",";
                first[0] = false;
                ctx.response().write(prefix + new JsonObject()
                  .put("id", row.getInteger("id"))
                  .put("name", row.getString("name"))
                  .put("email", row.getString("email"))
                  .encode());
                // => Write each row as JSON object in the array
              });

              stream.endHandler(v -> {
                ctx.response().write("]").end();
                // => Close JSON array and flush response
                conn.close();
                // => Return connection to pool after streaming completes
              });

              stream.exceptionHandler(err -> {
                System.err.println("Stream error: " + err);
                conn.close();
                // => Release connection even on error
                ctx.fail(500, err);
              });
            })
            .onFailure(err -> {
              conn.close();
              ctx.fail(500, err);
            });
        })
        .onFailure(err -> ctx.fail(503, err));
    });

    // Bulk insert with prepared batch
    router.post("/users/bulk").handler(ctx -> {
      io.vertx.ext.web.handler.BodyHandler.create().handle(ctx);
      JsonArray users = ctx.body().asJsonArray();
      // => Array of user objects to bulk insert

      java.util.List<Tuple> batch = new java.util.ArrayList<>();
      for (int i = 0; i < users.size(); i++) {
        JsonObject user = users.getJsonObject(i);
        batch.add(Tuple.of(user.getString("name"), user.getString("email")));
        // => Build batch of parameterized tuples
      }

      pool.preparedQuery("INSERT INTO users (name, email) VALUES ($1, $2)")
        .executeBatch(batch)
        // => executes all tuples in a single database round-trip
        // => Much more efficient than N separate INSERT statements
        .onSuccess(rows -> ctx.response().setStatusCode(201)
          .json(new JsonObject().put("inserted", users.size())))
        .onFailure(err -> ctx.fail(500, err));
    });

    vertx.createHttpServer()
      .requestHandler(router)
      .listen(8080)
      .onSuccess(s -> startPromise.complete())
      .onFailure(startPromise::fail);
  }

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new AdvancedSqlVerticle());
  }
}

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.MessageConsumer;

public class BackpressureDemo extends AbstractVerticle {

  @Override
  public void start(Promise<Void> startPromise) {
    MessageConsumer<String> consumer =
      vertx.eventBus().localConsumer("high.volume.events");
    // => Subscribe to high-volume event stream

    consumer.pause();
    // => Start paused; resume only when ready to process
    // => Prevents buffer overflow on startup

    consumer.handler(msg -> {
      String event = msg.body();
      // => Process the event
      System.out.println("Processing: " + event);
      // => Output: Processing: event-1234

      vertx.executeBlocking(() -> {
        // => Offload processing to worker thread
        processEvent(event);
        // => Could be slow: DB write, file I/O, external API call
        return null;
      })
      .onComplete(ar -> {
        if (consumer.isPaused()) {
          consumer.resume();
          // => Resume consuming after processing completes
          // => Ensures only one event processed at a time (backpressure)
        }
      });

      consumer.pause();
      // => Pause immediately after receiving; prevents flooding worker pool
    });

    consumer.resume();
    // => Start consuming events
    // => Will pause again after first event until processing completes

    startPromise.complete();

    // Simulate high-volume event producer
    for (int i = 0; i < 100; i++) {
      final int n = i;
      vertx.setTimer(n * 10, id ->
        vertx.eventBus().publish("high.volume.events", "event-" + n));
      // => Publish 100 events every 10ms = 100 events/second
      // => Consumer processes at its own pace; backpressure prevents overload
    }
  }

  private void processEvent(String event) {
    try { Thread.sleep(50); }
    catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    // => Simulate 50ms processing; producer sends 10/s; consumer handles 20/s
    // => Queue grows but processor keeps up; replace with real logic in production
  }

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new BackpressureDemo());
  }
}