Reactive Messaging / Quarkus

Quarkusというクラウドネイティブのフレームワークを試してみました。

環境 Ubuntu 22.04
インストール

$ sudo apt install unzip zip
$ curl -s “https://get.sdkman.io” | bash
$ source “$HOME/.sdkman/bin/sdkman-init.sh”
$ sdk install java
…
Setting java 17.0.6-tem as default.
$ sdk install quarkus
…
Setting quarkus 2.16.1.Final as default.
$ which quarkus
/home/n/.sdkman/candidates/quarkus/current/bin/quarkus

動作確認

$ quarkus create
$ cd code-with-quarkus/
$ quarkus dev

ウェブアクセス
localhost:8080

このサンプルにあるコードをもとに、下記を参考に依存性注入(CDI:Contexts and Dependency Injection)について、実際に動かしてみました。
https://ja.quarkus.io/guides/getting-started

code-with-quarkus/src/main/java/org/acme/GreetingResource.java

package org.acme;

import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

@Path("/hello")
public class GreetingResource {
    @Inject
    GreetingService service;

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("/greeting/{name}")
    public String greeting(String name) {
        return service.greeting(name);
    }

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    public String hello() {
        return "hello!\n";
    }
}

package org.acme;

import javax.inject.Inject;

import javax.ws.rs.GET;

import javax.ws.rs.Path;

import javax.ws.rs.Produces;

import javax.ws.rs.core.MediaType;

@Path("/hello")

public class GreetingResource {

@Inject

GreetingService service;

@GET

@Produces(MediaType.TEXT_PLAIN)

@Path("/greeting/{name}")

public String greeting(String name) {

return service.greeting(name);

}

@GET

@Produces(MediaType.TEXT_PLAIN)

public String hello() {

return "hello!\n";

}

code-with-quarkus/src/main/java/org/acme/getting/started/GreetingService.java

package org.acme;

import javax.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class GreetingService {

    public String greeting(String name) {
        return "hello " + name + "!\n";
    }
}

package org.acme;

import javax.enterprise.context.ApplicationScoped;

@ApplicationScoped

public class GreetingService {

public String greeting(String name) {

return "hello " + name + "!\n";

}

実行結果

$ curl localhost:8080/hello
hello!
$ curl localhost:8080/hello/greeting/everyone
hello everyone!

次に、このブログでも最近よく扱ったリアクティブのサンプルを動かしてみます。

$ git clone https://github.com/quarkusio/quarkus-quickstarts
$ cd quarkus-quickstarts/reactive-messaging-http-quickstart

下記、重要な部分の抜粋です。

reactive-messaging-http-quickstart/src/main/resources/application.properties

mp.messaging.outgoing.outgoing-costs.connector=quarkus-http

# here we are using a URL pointing to a test endpoint
# you can use e.g. an environment variable to change it
mp.messaging.outgoing.outgoing-costs.url=http://localhost:${quarkus.http.port}/cost-collector
%test.mp.messaging.outgoing.outgoing-costs.url=http://localhost:${quarkus.http.test-port}/cost-collector

# POST is the default method. Another possibility is PUT
mp.messaging.outgoing.outgoing-costs.method=POST

mp.messaging.incoming.incoming-costs.connector=quarkus-http

# the incoming-costs channel will be fed via an endpoint on the /costs path
mp.messaging.incoming.incoming-costs.path=/costs

# POST is the default method. Another possibility is PUT
mp.messaging.incoming.incoming-costs.method=POST

quarkus.http.host=0.0.0.0 #リモートからアクセスするために必要

resources/META-INF/resources/index.html

    add = function() {
        var value = document.getElementById('value').value;
        var currency = document.getElementById('currency').value;

        var cost = {
            value: document.getElementById('value').value,
            currency: document.getElementById('currency').value
        };

        fetch('costs', { method: 'POST', body: JSON.stringify(cost) });
    }

    updateCost = function() {
        fetch('cost-collector').then(response => response.text()).then(sum =>
            document.getElementById('content').textContent=sum
        );
    }

    window.setInterval(updateCost, 10000);

add = function() {

var value = document.getElementById('value').value;

var currency = document.getElementById('currency').value;

var cost = {

value: document.getElementById('value').value,

currency: document.getElementById('currency').value

};

fetch('costs', { method: 'POST', body: JSON.stringify(cost) });

}

updateCost = function() {

fetch('cost-collector').then(response => response.text()).then(sum =>

document.getElementById('content').textContent=sum

);

}

window.setInterval(updateCost, 10000);

reactive-messaging-http-quickstart/src/main/java/org/acme/reactivehttp/CostConverter.java

    @Incoming("incoming-costs")
    @Outgoing("outgoing-costs")
    double convert(Cost cost) {
        Double conversionRatio = conversionRatios.get(cost.getCurrency().toUpperCase());
        if (conversionRatio == null) {
            return 0.;
        }
        LOG.info("Convert!");
        return conversionRatio * cost.getValue();
    }

@Incoming("incoming-costs")

@Outgoing("outgoing-costs")

double convert(Cost cost) {

Double conversionRatio = conversionRatios.get(cost.getCurrency().toUpperCase());

if (conversionRatio == null) {

return 0.;

}

LOG.info("Convert!");

return conversionRatio * cost.getValue();

}

動きを確認するためログをいれてみました。(参考 https://ja.quarkus.io/guides/logging)

reactive-messaging-http-quickstart/src/main/java/org/acme/reactivehttp/CostCollector.java

package org.acme.reactivehttp;

import javax.enterprise.context.ApplicationScoped;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import org.jboss.logging.Logger;

@Path("/cost-collector")
@ApplicationScoped
public class CostCollector {
    private static final Logger LOG = Logger.getLogger(CostCollector.class);
    private double sum = 0;
    
    @POST
    public synchronized void consumeCost(String valueAsString) {
        sum += Double.parseDouble(valueAsString);
        LOG.info("consumeCost!");
    }
    @GET
    public synchronized double getSum() {
        LOG.info("getSum!");
        return sum;
    }
}

package org.acme.reactivehttp;

import javax.enterprise.context.ApplicationScoped;

import javax.ws.rs.GET;

import javax.ws.rs.POST;

import javax.ws.rs.Path;

import org.jboss.logging.Logger;

@Path("/cost-collector")

@ApplicationScoped

public class CostCollector {

private static final Logger LOG = Logger.getLogger(CostCollector.class);

private double sum = 0;

@POST

public synchronized void consumeCost(String valueAsString) {

sum += Double.parseDouble(valueAsString);

LOG.info("consumeCost!");

}

@GET

public synchronized double getSum() {

LOG.info("getSum!");

return sum;

}

ボタンをクリックすると、convert()が呼ばれた後に、すぐcomsumeCostが呼ばれています。(incomming-costアノテーションによるものでしょう)　通貨のレートをかけた結果を足し算。
ポーリングで画面な更新している。(テストでは500ミリ秒から10秒間隔に変更)
(他にwebsocketの同様のサンプルもありましたが、正しく動作しているもののws部について、ブラウザのデバッグ画面の表示ができずhttpにしました)

コンソール画面

あとメモとして、リアクティブについて下記に興味深い記述が多数ありましたので、スクリーンショットをとり引用させていただきます。
https://ja.quarkus.io/continuum/