OpenShift での AMQ Online の使用

本書では、変数を含むコードブロックが紹介されていますが、これは、お客様のシステム環境に固有の値に置き換える必要があります。このドキュメントでは、そのようなテキストはイタリックモノスペースのスタイルで指定しています。

たとえば、次のコードブロックで、my-namespace はインストールで使用されている namespace に置き換えます。

sed -i 's/amq-online-infra/my-namespace/' install/bundles/enmasse-with-standard-authservice/*.yaml

標準アドレス空間は、次の異なるアドレスタイプを 5 つサポートしています。

ブローカーアドレス空間は、次の 2 つのアドレスタイプをサポートします。

このアドレス空間の例では、AddressSpace を作成するために必要なオプションのみを示しています。

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard 1
  plan: standard-unlimited 2

このアドレス空間の例は、AddressSpace の認証サービスを設定する方法を示しています。

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  authenticationService:
    name: standard-authservice 1

このアドレス空間の例は、外部認証サービスのホスト名、ポート番号、およびレルムをオーバーライドする方法を示しています。オーバーライドを指定できるかどうかは、AMQ Online 管理者が外部認証サービスをどのように設定しているかによって異なります。

メッセージングテナントがホスト名、ポート番号、およびレルムを上書きできるように外部認証サービスを設定する方法は、オーバーライドを許可する外部認証サービスの例 を参照してください。

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  authenticationService:
    name: external-authservice 1
    type: external
    overrides: 2
      realm: amq-online-infra-space-standard-auth
      host: standard-authservice-amq-online-infra.apps.wfd-28d9.openshiftworkshop.com
      port: 5671
      caCertSecret:
        name: my-ca-cert

これらのアドレス空間の例は、AddressSpace の外部エンドポイントを設定して、OpenShift クラスター外のメッセージングエンドポイントにアクセスする方法を示しています。

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  authenticationService:
    name: standard-authservice
  endpoints:
  - name: messaging 1
    service: messaging 2
    expose:
     type: loadbalancer 3
     loadBalancerPorts: 4
     - amqp
     - amqps
      annotations: 5
        mykey: myvalue
      loadBalancerSourceRanges: 6
      - 10.0.0.0/8

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  authenticationService:
    name: standard-authservice
  endpoints:
  - name: messaging 1
    service: messaging 2
    expose:
     type: route
     routeServicePort: amqps 3
     routeTlsTermination: passthrough 4
     routeHost: messaging.example.com 5

次のアドレス空間の例は、さまざまな証明書プロバイダーを使用して AddressSpace のエンドポイントを設定する方法を示しています。証明書プロバイダーは、AddressSpace のエンドポイントに対して証明書を発行する方法を決定します。

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  authenticationService:
    name: standard-authservice
  endpoints:
  - name: messaging
    service: messaging
    cert:
     provider: openshift 1

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  authenticationService:
    name: standard-authservice
  endpoints:
  - name: messaging
    service: messaging
    cert:
     provider: selfsigned 1

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  authenticationService:
    name: standard-authservice
  endpoints:
  - name: messaging
    service: messaging
    cert:
     provider: certBundle 1
     tlsKey: Y2VydGJ1bmRsZXByb3ZpZGVyY2VydA== 2
     tlsCert: Y2VydGJ1bmRsZXByb3ZpZGVyY2VydA== 3

次の 3 つのエクスポートタイプを使用して、アドレス空間情報をエクスポートできます。

service.host: messaging.svc
service.port.amqp: 5672
external.host: external.example.com
external.port: 5671
ca.crt: // PEM formatted CA

  externalName:  messaging.svc
    ports:
    - name: amqp
      port: 5672
      protocol: TCP
      targetPort: 5672

認証に相互 TLS を使用しない場合は、SASL PLAIN を使用できます。ユーザー名とパスワードがプレーンテキストとして送信されるため、TLS を有効にしないことはお勧めしません。

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  connectors:
  - name: remote1 1
    endpointHosts: 2
    - host: messaging.example.com
      port: 5672
    - host: messaging2.example.com
    idleTimeout: 3
    maxFrameSize: 4
    tls: {} 5
    credentials: 6
      username:
        value: test
      password:
        valueFromSecret:
          name: password-secret
          key: password.txt
    role: 7
    addresses: 8
    - name: p1
      pattern: "prices/*"
    - name: p2
      pattern: "clients/*/1"

クライアント TLS 証明書を設定すると、SASL EXTERNAL を認証に使用できるようになります。証明書は、インラインで指定するか、秘密の参照を使用して指定できます。

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: myspace
spec:
  type: standard
  plan: standard-unlimited
  connectors:
  - name: remote1 1
    endpointHosts: 2
    - host: messaging.example.com
      port: 5671
    tls:
      caCert: 3
        valueFromSecret:
          name: remote-certs
          key: ca.crt
      clientCert: 4
        valueFromSecret:
          name: remote-certs
          key: tls.crt
      clientKey: 5
        valueFromSecret:
          name: remote-certs
          key: tls.key
    addresses:
    - name: p1
      pattern: "*"

apiVersion: enmasse.io/v1beta1
kind: Address
metadata:
    name: myspace.myqueue 1
spec:
    address: myqueue 2
    type: queue 3
    plan: standard-small-queue 4

トピックとサブスクリプションを使用する場合、次の例に示すように、サブスクリプションは topic: フィールドを使用して所属するトピックを参照します。

apiVersion: enmasse.io/v1beta1
kind: Address
metadata:
    name: myspace.mytopic
spec:
    address: mytopic
    type: topic
    plan: standard-small-topic

apiVersion: enmasse.io/v1beta1
kind: Address
metadata:
    name: myspace.mysub
spec:
    address: mysub
    type: subscription
    plan: standard-small-subscription
    topic: mytopic 1

apiVersion: enmasse.io/v1beta1
kind: Address
metadata:
    name: myspace.myqueue
spec:
    address: myqueue
    type: queue
    plan: standard-small-queue
    messageTtl: 1
        minimim: 30000
        maximum: 300000

messageTtl フィールドは、キューまたはトピックに書き込まれたメッセージの有効な absolute-expiry-time を制限するために使用されます。maximum と minimum の値はミリ秒単位で定義されます。システムは、次の値に基づいて特定のアドレスへの着信メッセージの TTL 値を調整します。

TTL が定義されていない状態で到着したメッセージは、TTL 値が無限大であると見なされます。

期限切れのメッセージは、queue、subscription、または一時トピックサブスクリプションから定期的に自動削除されます。これらのメッセージは失われます。これは 30 秒ごとに発生します。

TTL 制限は、アドレスプランによって課される場合もあります。プランとアドレスの両方に TTL 制限が課されている場合には、アドレス TTL 制限は TTL 制限をさらに絞りこむ以外手段はありません。アドレス status セクションには、有効な TTL 値が表示されます。

フォワーダーを使用すると、以下が可能です。

アドレスフォワーダーを使用するには、最初にアドレス空間のリモート AMQP サーバーへのコネクターを設定する必要があります。アドレス空間コネクターの詳細は、アドレス空間コネクターの例 を参照してください。

アドレス転送は、standard アドレス空間タイプでのみサポートされており、queue および subscritpion アドレスタイプでのみサポートされています。queue アドレスタイプを使用すると、メッセージをリモート AMQP サーバーに転送したり、リモート AMQP サーバーからローカルキューに転送したりできます。subscription アドレスタイプでは、リモート AMQP アドレスへのフォワーダーを作成できますが、メッセージをサブスクリプションにコピーするフォワーダーを作成することはできません。つまり、subscription アドレスタイプは、例に示すように、out 方向の転送のみをサポートします。

次の例では、コネクター remote1 がアドレス空間用に設定されていることを前提としています。

apiVersion: enmasse.io/v1beta1
kind: Address
metadata:
  name: myspace.myqueue
spec:
  address: myqueue
  type: queue
  plan: standard-small-queue
  forwarders:
  - name: f1 1
    remoteAddress: remote1/clients/me/1 2
    direction: out 3

apiVersion: enmasse.io/v1beta1
kind: Address
metadata:
  name: myspace.myqueue
spec:
  address: myqueue
  type: queue
  plan: standard-small-queue
  forwarders:
  - name: f1 1
    remoteAddress: remote1/prices/milk 2
    direction: in 3

password タイプの場合には、追加フィールドの パスワード を、そのユーザーのパスワードの base64 エンコード値に設定する必要があります。リソースの読み取り時にパスワードは出力されません。

パスワードは、コマンドラインで base64 エンコードできます。たとえば、my-password をエンコードするには以下を実行します。

$ echo -n my-password | base64
bXktcGFzc3dvcmQ=

serviceaccount タイプの場合、username フィールドには、認証に使用される OpenShift サービスアカウント名が含まれている必要があります。メッセージングクライアントに接続するときは、文字列 @@serviceaccount@@ をユーザー名として使用し、サービスアカウントトークンをパスワードとして使用します。アプリケーションで使用される AMQP クライアントは、SASL メカニズムタイプ PLAIN を使用するように設定する必要があります。

AMQ Online では、標準のコマンドラインツールを使用してユーザーを作成できます。

ユーザーは、標準のコマンドラインツールを使用して削除できます。

コマンドラインを使用して、既存のユーザーの権限を編集できます。

次の AMQ Online Python の例を使用して、アプリケーションを AMQ Online に接続できます。この例では、myqueue という名前の queue タイプのアドレスを作成したと想定しています。

from __future__ import print_function, unicode_literals
from proton import Message
from proton.handlers import MessagingHandler
from proton.reactor import Container

class HelloWorld(MessagingHandler):
    def __init__(self, server, address):
        super(HelloWorld, self).__init__()
        self.server = server
        self.address = address

    def on_start(self, event):
        conn = event.container.connect(self.server)
        event.container.create_receiver(conn, self.address)
        event.container.create_sender(conn, self.address)

    def on_sendable(self, event):
        event.sender.send(Message(body="Hello World!"))
        event.sender.close()

    def on_message(self, event):
        print(event.message.body)
        event.connection.close()

Container(HelloWorld("amqps://_messaging-route-hostname_:443", "myqueue")).run()

class CapabilityOptions(ReceiverOption):
    def apply(self, receiver):
        receiver.source.capabilities.put_object(symbol("topic"))

def on_start(self, event):
        event.container.create_receiver(conn, self.address, options=CapabilityOptions())

次の AMQ Online JMS の例を使用して、アプリケーションを AMQ Online に接続できます。この例では、myqueue という名前の queue タイプのアドレスを作成したと想定しています。

package org.apache.qpid.jms.example;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

public class HelloWorld {
    public static void main(String[] args) throws Exception {
        try {
            // The configuration for the Qpid InitialContextFactory has been supplied in
            // a jndi.properties file in the classpath, which results in it being picked
            // up automatically by the InitialContext constructor.
            Context context = new InitialContext();

            ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
            Destination queue = (Destination) context.lookup("myQueueLookup");

            Connection connection = factory.createConnection(System.getProperty("USER"), System.getProperty("PASSWORD"));
            connection.setExceptionListener(new MyExceptionListener());
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            MessageProducer messageProducer = session.createProducer(queue);
            MessageConsumer messageConsumer = session.createConsumer(queue);

            TextMessage message = session.createTextMessage("Hello world!");
            messageProducer.send(message, DeliveryMode.NON_PERSISTENT, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);
            TextMessage receivedMessage = (TextMessage) messageConsumer.receive(2000L);

            if (receivedMessage != null) {
                System.out.println(receivedMessage.getText());
            } else {
                System.out.println("No message received within the given timeout!");
            }

            connection.close();
        } catch (Exception exp) {
            System.out.println("Caught exception, exiting.");
            exp.printStackTrace(System.out);
            System.exit(1);
        }
    }

    private static class MyExceptionListener implements ExceptionListener {
        @Override
        public void onException(JMSException exception) {
            System.out.println("Connection ExceptionListener fired, exiting.");
            exception.printStackTrace(System.out);
            System.exit(1);
        }
    }
}

jndi.properties を使用:

connectionfactory.myFactoryLookup = amqps://messaging-route-hostname:443?transport.trustAll=true&transport.verifyHost=false
queue.myQueueLookup = myqueue

次の AMQ Online JavaScript の例を使用して、アプリケーションを AMQ Online に接続できます。この例では、myqueue という名前の queue タイプのアドレスを作成したと想定しています。

var container = require('rhea');
container.on('connection_open', function (context) {
    context.connection.open_receiver('myqueue');
    context.connection.open_sender('myqueue');
});
container.on('message', function (context) {
    console.log(context.message.body);
    context.connection.close();
});
container.on('sendable', function (context) {
    context.sender.send({body:'Hello World!'});
    context.sender.detach();
});
container.connect({username: 'username', password: 'password', port:443, host:'messaging-route-hostname', transport:'tls', rejectUnauthorized:false});

var container = require('rhea');
var WebSocket = require('ws');

container.on('connection_open', function (context) {
    context.connection.open_receiver('myqueue');
    context.connection.open_sender('myqueue');
});
container.on('message', function (context) {
    console.log(context.message.body);
    context.connection.close();
});
container.on('sendable', function (context) {
    context.sender.send({body:'Hello World!'});
    context.sender.detach();
});

var ws = container.websocket_connect(WebSocket);
container.connect({username: 'username', password: 'password', connection_details: ws("wss://messaging-route-hostname:443", ["binary"], {rejectUnauthorized: false})});

C++ クライアントには、Python と同じオプションを持つ同等の simple_recv と simple_send の例があります。ただし、C++ ライブラリーは URL に対して同じレベルの処理を実行しません。特に、TLS の使用を意味する amqps:// は使用できないので、例を次のように変更する必要があります。

#include <proton/connection.hpp>
#include <proton/container.hpp>
#include <proton/default_container.hpp>
#include <proton/delivery.hpp>
#include <proton/message.hpp>
#include <proton/messaging_handler.hpp>
#include <proton/ssl.hpp>
#include <proton/thread_safe.hpp>
#include <proton/tracker.hpp>
#include <proton/url.hpp>

#include <iostream>

#include "fake_cpp11.hpp"

class hello_world : public proton::messaging_handler {
  private:
    proton::url url;

  public:
    hello_world(const std::string& u) : url(u) {}

    void on_container_start(proton::container& c) OVERRIDE {
        proton::connection_options co;
        co.ssl_client_options(proton::ssl_client_options());
        c.client_connection_options(co);
        c.connect(url);
    }

    void on_connection_open(proton::connection& c) OVERRIDE {
        c.open_receiver(url.path());
        c.open_sender(url.path());
    }

    void on_sendable(proton::sender &s) OVERRIDE {
        proton::message m("Hello World!");
        s.send(m);
        s.close();
    }

    void on_message(proton::delivery &d, proton::message &m) OVERRIDE {
        std::cout << m.body() << std::endl;
        d.connection().close();
    }
};

int main(int argc, char **argv) {
    try {
        std::string url = argc > 1 ? argv[1] : "messaging-route-hostname:443/myqueue";

        hello_world hw(url);
        proton::default_container(hw).run();

        return 0;
    } catch (const std::exception& e) {
        std::cerr << e.what() << std::endl;
    }

    return 1;
}

void on_container_start(proton::container& cont) override {
        proton::connection conn = cont.connect(conn_url_);
        proton::receiver_options opts {};
        proton::source_options sopts {};

        sopts.capabilities(std::vector<proton::symbol> { "topic" });
        opts.source(sopts);

        conn.open_receiver(address_, opts);
    }

次の AMQ Online .NET の例を使用して、アプリケーションを AMQ Online に接続できます。この例では、myqueue という名前の queue タイプのアドレスを作成したと想定しています。

using System;
using Amqp;

namespace Test
{
    public class Program
    {
        public static void Main(string[] args)
        {
            String url = (args.Length > 0) ? args[0] : "amqps://messaging-route-hostname:443";
            String address = (args.Length > 1) ? args[1] : "myqueue";

            Connection.DisableServerCertValidation = true;
            Connection connection = new Connection(new Address(url));
            Session session = new Session(connection);
            SenderLink sender = new SenderLink(session, "test-sender", address);

            Message messageSent = new Message("Test Message");
            sender.Send(messageSent);

            ReceiverLink receiver = new ReceiverLink(session, "test-receiver", address);
            Message messageReceived = receiver.Receive(TimeSpan.FromSeconds(2));
            Console.WriteLine(messageReceived.Body);
            receiver.Accept(messageReceived);

            sender.Close();
            receiver.Close();
            session.Close();
            connection.Close();
        }
    }
}