Following on from my post Service discovery with Vert.x, let’s switch from using the Vert.x built-in Service Discovery to using ZooKeeper.

You can fire up a docker container running ZooKeeper using

docker run --name zookeeper --restart always -d zookeeper

Now we’ll need to add the following to our pom.xml

<dependency>
   <groupId>io.vertx</groupId>
   <artifactId>vertx-service-discovery-bridge-zookeeper</artifactId>
   <version>${vertx.version}</version>
</dependency>

Note: vertx.version is set to 3.5.1

The code to create the Service Discovery to ZooKeeper is, as follows

ServiceDiscovery discovery = ServiceDiscovery.create(vertx)
   .registerServiceImporter(new ZookeeperServiceImporter(),
      new JsonObject()
        .put("connection", "172.17.0.2:2181")
        .put("basePath", "/services/hello-service"));

Replace the ip address with the one create by Docker or if you’re running on localhost.

We register the ZooKeeperServiceImporter and supply at least the “connection”, the “basePath” is not required, a default will be supplied if none is explicitly supplied.

Don’t forget you’ll need the import

import io.vertx.servicediscovery.zookeeper.ZookeeperServiceImporter;

References

Docker ZooKeeper
Vert.x Service Discovery

Code

Source code relating to these posts on Vert.x can be found at VertxSamples. The code differs from the posts in that in some cases it’s been refactored to reduce some code duplication etc.

Obviously you can use whatever configuration code you like to configure your Verticles and/or Vert.x applications, you might use standard configuration files or a Redis store or whatever.

Vert.x has a single library which allows us to abstract such things and offers different storage formats and code to access different storage mechanisms.

Add the following dependency to your pom.xml

<dependency>
   <groupId>io.vertx</groupId>
   <artifactId>vertx-config</artifactId>
   <version>${vertx.version}</version>
</dependency>

vertx.version in my version is 3.5.0

We can actually define multiple storage mechanisms for our configuration data and chain them. For example imagine we have some we have configuration on the localhost and other configuration on a remote HTTP server. We can set things up to get properties that are a combination of both locations.

Note: If location A has property port set to 8080 but location B (which appears after location A in the chain) has port set to 80 then the last property located in the chain is the one we’ll see when querying the configuration code for this property. For other properties the result is a combination of those from A and those from B.

We can also mark configuration sources as optional so, if they do not exist or are down, the chaining will not fail or exception.

Let’s start with a simple config.properties file stored in /src/main/resources. Here’s the file (nothing exciting here)

port=8080

We’ll now create the ConfigStoreOptions for a “properties” file, then add it to the store (we’ll just use a single store for this example) and finally we’ll retrieve the port property from the configuration retriever…

// create the options for a properties file store
ConfigStoreOptions propertyFile = new ConfigStoreOptions()
   .setType("file")
   .setFormat("properties")
   .setConfig(new JsonObject().put("path", "config.properties"));

// add the options to the chain
ConfigRetrieverOptions options = new ConfigRetrieverOptions()
   .addStore(propertyFile);
    // .addStore for other stores here

ConfigRetriever retriever = ConfigRetriever.create(vertx, options);
retriever.getConfig(ar ->
{
   if(ar.succeeded()) {
      JsonObject o  = ar.result();
      int port = o.getInteger("port");
      // use port config
   }
});

Note: Don’t forget to set the configuration store options “path” to the location and name of your file.

We may get to a point whereby we have multiple Vert.x applications running and we want one Verticle to communicate with another – this is easy enough if the IP address and port are fixed but not so easy in more scalable/real-world scenarios where we cannot guarantee these are fixed.

In such situations we can use service discovery to locate other services.

Before we get started with the code, we need to add the following to the pom.xml

<dependency>
   <groupId>io.vertx</groupId>
   <artifactId>vertx-service-discovery</artifactId>
   <version>${vertx.version}</version>
</dependency>

I’m using vertx.version 3.5.0 in my examples.

Publishing/Registering our Verticle with ServiceDiscovery

To register our Verticle with ServiceDisovery we create a Record object which tells the ServiceDiscovery how to access a Verticle, this includes it’s host/IP, port and service root along with a name for other code to use to locate the service. For example

Record record = HttpEndpoint.createRecord(
   "hello-service",
   "localhost",
   8080,
   "/hello");

So this basically says, create a Record named “hello-service” (the key or name of the service) and it’s IP/host is localhost, obviously this is just for my testing. Next we supply the exposed port and finally the root of the service.

We then publish this record to the ServiceDiscovery object like this

discovery.publish(record, ar ->
{
   if (ar.succeeded()) {
      // publication succeeded
      publishedRecord = ar.result();
   } else {
      // publication failed
   }
});

Upon success we store the Record (in this case we only do this if the call succeeded) so that we can unpublish the service if it’s shutdown.

Let’s look at the full code for a simplified HelloVerticle

public class HelloVerticle extends AbstractVerticle {

    private ServiceDiscovery discovery;
    private Record publishedRecord;

    @Override
    public void start() {
        discovery = new DiscoveryImpl(vertx, 
           new ServiceDiscoveryOptions());

        Router router = Router.router(vertx);
        router.get("/hello").handler(ctx -> {
            ctx.response()
                .putHeader("content-type", "text/plain")
                .end("hello");
        });

        Record record = HttpEndpoint.createRecord(
                "hello-service",
                "localhost",
                8080,
                "/hello");

        discovery.publish(record, ar ->
        {
            if (ar.succeeded()) {
                // publication success
                publishedRecord = ar.result();
            } else {
                // publication failure
            }
        });

        vertx
           .createHttpServer()
           .requestHandler(router::accept)
           .listen(8080, ar -> {
              // handle success/failure 
           });
    }

    @Override
    public void stop() {
        if(discovery != null) {
            discovery.unpublish(publishedRecord.getRegistration(), ar ->
            {
                if (ar.succeeded()) {
                    // Success
                } else {
                    // cannot unpublish the service, 
                    // may have already been removed, 
                    // or the record is not published
                }
            });

            discovery.close();
        }
    }
}

Locating a service via ServiceDiscovery

Let’s take a look at some “consumer” code which will use service discovery to locate our “HelloVerticle”. As expected we need to create access to the ServiceDiscovery object and then we try to locate the Record for a previously added Record.

In the example, below, we search for the “name”, “hello-service”, this is wrapped into a JsonObject and the result (if successful will contain a Record which matches the search criteria. Using HttpClient we can now simply get the reference to this service and interact with it without ever knowing it’s IP address or port.

ServiceDiscovery discovery = ServiceDiscovery.create(v);
discovery.getRecord(
   new JsonObject().put("name", "hello-service"), found -> {
   if(found.succeeded()) {
      Record match = found.result();
      ServiceReference reference = discovery.getReference(match);
      HttpClient client = reference.get();

      client.getNow("/hello", response ->
         response.bodyHandler(
            body -> 
               System.out.println(body.toString())));
   }
});

Vert.x includes an HttpClient and associated code for interacting with HTTP protocols, obviously this can be used to write client applications or in situations where we might use a reference from service discovery.

This is a very short post which is just mean’t to demonstrate the client capability which will be used in the next post (Service discovery with Vert.x).

HttpClient client = vertx.createHttpClient();
client.getNow(8080, "localhost", "/hello", response ->
{
   response.bodyHandler(
      body -> 
         System.out.println(body.toString()));
});

The HttpClient gives us get, post, head etc. the Now postfixed named methods tend to be simpler syntax including the supply of callback and are composable using fluent style syntax.

See also Creating an HTTP client for more information.

In my previous post on Vert.x, I demonstrated how to create multiple Verticle’s but also how a Verticle needn’t actually be a “service” but in some cases is just a worker process or in the case of the previous post a “main” or startup process.

This leads us to better see how Vert.x is actor-like with regards to Verticles.

In situations where we have multiple Verticles and possible some of those Verticles are simply workers for the other Verticles, we would need some way of communicating across those different parts of the application. Hence Vert.x also comes with an event bus for communicating across Verticles. This is a global event bus across all our Verticles.

Let’s, for sake of simplicity change our previous example so that whenever a service is successfully routed to, an event is sent out of the EventBus to say which services was called.

So, taken the example from my previous post, add the following to the MainVerticle start method

EventBus eventBus = vertx.eventBus();

eventBus.consumer("/svc", m -> {
   System.out.println(m.body());
});

The String /svc is the topic or an address if you prefer that we publish and consume message from. In the example above, we’re listening to the event bus for the anything sent (or published) to the address /svc. How you name you addresses is totally upto you.

Don’t forget to import io.vertx.core.eventbus.EventBus to use the event bus.

Next we’ll add code to the services to send a message on the /svc address when we wish to communicate something. Here’s the code for the HelloVerticle start method with the event bus added.

@Override
public void start() {

   EventBus evBus = vertx.eventBus();

   router.route("/hello").handler(ctx -> {
      evBus.send("/svc", "hello called");

      ctx.response()
         .putHeader("content-type", "text/plain")
         .end("Hello " + ctx.queryParam("name"));
      });

      vertx.createHttpServer()
         .requestHandler(router::accept)
         .listen(8080);

   System.out.println("HTTP server started on port 8080");
}

As you can see from the highlighted lines adding an event bus and sending messages out on it, is very simple. If we updated the WorldVerticle from the previous post also (but sending “world called”) then when run, the MainVerticle will print a line to the console every time the HelloVerticle or WorldVerticle routes are called.

Publish and Send

In the example (above) I used the event buss method send to send messages out on the event bus, but send will only be delivered to at most one of the handlers of the message.

Therefore is we added the following code to each of our service Verticles

evBus.consumer("/svc", m -> {
   System.out.println("Hello " + m.body());
});

After the declaration of the evBus variable.

we’ve now got a requirement that the message on address /svc needs to be handled by multiple consumers. Running up the application as it currently stands would result in only one of the consumers receiving the messages. In my case this was the MainVerticle, however there’s no guarantee which Verticle would receive the messages as this is determined by Vert.x in a round robin fashion.

So here’s the difference between send and publish. If you change the code from send to publish in the service Verticles. Then messages will be routed to all consumers.

In my previous post I looked at creating my first Verticle/Vert.x application. This had a single Verticle acting as a service and running up a server against port 8080. If we therefore assume that by default a Verticle represents a single service then to create multiple service we simply create multi-Verticles.

Let’s create three Verticle’s. The first will act as the “main” verticle (or run as an application) and it will register/deploy the Verticle’s that we’re using to host our two services, the HelloVerticle and WorldVerticle.

Each service will itself create/host an HttpServer. Traditionally this would tend to mean we’d have two ports exposed, one for each service, but Vert.x allows us to create multiple HttpServer’s on the same port and uses something like “round robin” to try to locate a valid route.

Check my previous post for the required pom.xml.

Here’s MainVerticle.java

package com.putridparrot;

import io.vertx.core.AbstractVerticle;
import io.vertx.ext.web.Router;

public class MainVerticle extends AbstractVerticle {
    @Override
    public void start() {
        Router router  = Router.router(vertx);

        vertx.deployVerticle(new HelloVerticle(router));
        vertx.deployVerticle(new WorldVerticle(router));
    }
}

Notice that this Verticle’s job is to both register the other Verticle’s and acts as the main entry point to our application.

You do not need to pass the Router around in this way unless we intend to have multiple routes on the same port. So in this case we’re creating the router that will be used by both services on the same port.

Let’s look at the HelloVerticle.java service (both HelloVerticle and WorldVerticle, in this simple example, are basically the same, but I’ll reproduce all code here anyway).

package com.putridparrot;

import io.vertx.core.AbstractVerticle;
import io.vertx.ext.web.Router;

public class HelloVerticle extends AbstractVerticle {

    private Router router;

    public HelloVerticle(Router router) {
        this.router = router;
    }

    @Override
    public void start() {
        router.route("/hello").handler(ctx -> {
            ctx.response()
                    .putHeader("content-type", "text/plain")
                    .end("Hello " + ctx.queryParam("name"));
        });

        vertx.createHttpServer()
                .requestHandler(router::accept)
                .listen(8080);

        System.out.println("HTTP Hello server started on port 8080");
    }
}

There’s nothing particularly special or different here, compared to my previous posts code. So let’s look at the WorldVerticle.java

package com.putridparrot;

import io.vertx.core.AbstractVerticle;
import io.vertx.ext.web.Router;

public class WorldVerticle extends AbstractVerticle {

    private Router router;

    public WorldVerticle(Router router) {
        this.router = router;
    }

    @Override
    public void start() {
        router.route("/world").handler(ctx -> {
            ctx.response()
                    .putHeader("content-type", "text/plain")
                    .end("World " + ctx.queryParam("name"));
        });

        vertx.createHttpServer()
                .requestHandler(router::accept)
                .listen(8080);

        System.out.println("HTTP World server started on port 8080");
    }
}

What’s interesting is that both HelloVerticle and WorldVerticle create the HTTP server on the same port. As stated earlier Vert.x simply handles this for us without a port conflict and because we’re using a single router across these two Verticle’s Vert.x can correctly located the service for each of the following URL’s

http://localhost:8080/hello?name=Hello
http://localhost:8080/world?name=World