第11章 メッセージの傍受
AMQ Broker では、ブローカーに出入りするパケットを傍受して、パケットの監査やメッセージのフィルターを実行できます。インターセプターはインターセプトするパケットを変更できます。これにより、強力になりますが、潜在的に危険性があります。
ビジネス要件を満たすためのインターセプターを開発できます。インターセプターはプロトコル固有であるため、適切なインターフェイスを実装する必要があります。
インターセプターは、ブール値を返す intercept() メソッドを実装する必要があります。値が true の場合、メッセージパケットは続行されます。false の場合、プロセスは中止され、他のインターセプターは呼び出されず、メッセージパケットはこれ以上処理されません。
11.1. インターセプターの作成
独自の受信インターセプターおよび発信インターセプターを作成できます。すべてのインターセプターはプロトコル固有で、サーバーに出入りするパケットに対して呼び出されます。これにより、監査パケットなどのビジネス要件を満たすインターセプターを作成できます。インターセプターは、傍受するパケットを変更できます。これにより、危険が伴います。そのため、注意して使用してください。
インターセプターとその依存関係は、ブローカーの Java クラスに配置する必要があります。<broker_instance_dir>/lib ディレクトリーは、デフォルトでクラスパスに含まれているため、使用することができます。
手順
以下の例は、渡された各パケットのサイズをチェックするインターセプターを作成する方法を示しています。この例では、プロトコルごとに特定のインターフェイスを実装します。
適切なインターフェイスを実装し、その
intercept()メソッドをオーバーライドします。AMQP プロトコルを使用している場合は、
org.apache.activemq.artemis.protocol.amqp.broker.AmqpInterceptorインターフェイスを実装してください。package com.example; import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage; import org.apache.activemq.artemis.protocol.amqp.broker.AmqpInterceptor; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements AmqpInterceptor { private final int ACCEPTABLE_SIZE = 1024; @Override public boolean intercept(final AMQPMessage message, RemotingConnection connection) { int size = message.getEncodeSize(); if (size <= ACCEPTABLE_SIZE) { System.out.println("This AMQPMessage has an acceptable size."); return true; } return false; } }Core Protocol を使用している場合、インターセプターは、
org.apache.artemis.activemq.api.core.Interceptorインターフェイスを実装する必要があります。package com.example; import org.apache.artemis.activemq.api.core.Interceptor; import org.apache.activemq.artemis.core.protocol.core.Packet; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements Interceptor { private final int ACCEPTABLE_SIZE = 1024; @Override boolean intercept(Packet packet, RemotingConnection connection) throws ActiveMQException { int size = packet.getPacketSize(); if (size <= ACCEPTABLE_SIZE) { System.out.println("This Packet has an acceptable size."); return true; } return false; } }MQTT プロトコルを使用している場合は、
org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptorインターフェイスを実装してください。package com.example; import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor; import io.netty.handler.codec.mqtt.MqttMessage; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements Interceptor { private final int ACCEPTABLE_SIZE = 1024; @Override boolean intercept(MqttMessage mqttMessage, RemotingConnection connection) throws ActiveMQException { byte[] msg = (mqttMessage.toString()).getBytes(); int size = msg.length; if (size <= ACCEPTABLE_SIZE) { System.out.println("This MqttMessage has an acceptable size."); return true; } return false; } }STOMP プロトコルを使用している場合は、
org.apache.activemq.artemis.core.protocol.stomp.StompFrameInterceptorインターフェイスを実装してください。package com.example; import org.apache.activemq.artemis.core.protocol.stomp.StompFrameInterceptor; import org.apache.activemq.artemis.core.protocol.stomp.StompFrame; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; public class MyInterceptor implements Interceptor { private final int ACCEPTABLE_SIZE = 1024; @Override boolean intercept(StompFrame stompFrame, RemotingConnection connection) throws ActiveMQException { int size = stompFrame.getEncodedSize(); if (size <= ACCEPTABLE_SIZE) { System.out.println("This StompFrame has an acceptable size."); return true; } return false; } }
