旧バージョンのデフォルトの同期メッセージングを使用する

次に、宛先コンフィギュレーターから順に、各クラスを調べます。

宛先コンフィグレーターを調べる

m4q7-able-implモジュールとm4q7-baker-implモジュールには、それぞれ宛先コンフィギュレータークラスM4Q7AbleMessagingConfiguratorとM4Q7BakerMessagingConfiguratorがあります。 それぞれが宛先を作成して構成します。

M4Q7AbleMessagingConfiguratorクラスは、acme/m4q7_able宛先を構成します。

@Component
public class M4Q7AbleMessagingConfigurator {

	@Activate
	private void _activate(BundleContext bundleContext) {
		DestinationConfiguration destinationConfiguration =
			DestinationConfiguration.createSerialDestinationConfiguration(
				"acme/m4q7_able");

		Destination destination = _destinationFactory.createDestination(
			destinationConfiguration);

		_serviceRegistration = bundleContext.registerService(
			Destination.class, destination,
			MapUtil.singletonDictionary(
				"destination.name", destination.getName()));
	}

	@Deactivate
	private void _deactivate() {
		if (_serviceRegistration != null) {
			_serviceRegistration.unregister();
		}
	}

	@Reference
	private DestinationFactory _destinationFactory;

	private ServiceRegistration<Destination> _serviceRegistration;

}

M4Q7BakerMessagingConfiguratorクラスは、acme/m4q7_baker宛先を構成します。

@Component
public class M4Q7BakerMessagingConfigurator {

	@Activate
	private void _activate(BundleContext bundleContext) {
		DestinationConfiguration destinationConfiguration =
			DestinationConfiguration.createSerialDestinationConfiguration(
				"acme/m4q7_baker");

		Destination destination = _destinationFactory.createDestination(
			destinationConfiguration);

		_serviceRegistration = bundleContext.registerService(
			Destination.class, destination,
			MapUtil.singletonDictionary(
				"destination.name", destination.getName()));
	}

	@Deactivate
	private void _deactivate() {
		if (_serviceRegistration != null) {
			_serviceRegistration.unregister();
		}
	}

	@Reference
	private DestinationFactory _destinationFactory;

	private ServiceRegistration<Destination> _serviceRegistration;

}

両方のコンフィギュレータは コンポーネント クラスです。 これらは、 @Reference アノテーションを使用して、 DestinationFactory インスタンスを挿入します。

_activate(BundleContext) メソッドは、 DestinationFactory と DestinationConfiguration を使用して、 シリアル 宛先を作成します。 最後に、 _activate(BundleContext) メソッドは、 BundleContextを使用して、 OSGi サービスに Destination を登録します。

コンフィギュレーターが無効になると、それらの_deactivate()メソッドは宛先サービスの登録を解除します。

リスナーを調べる

m4q7-charlie-impl モジュールの M4Q7CharlieMessageListener クラスは、 acme/m4q7_able Destinationに送信されるメッセージをリッスンします。 これは、 メッセージのリッスン で示されているのと同じ方法で登録されます。

M4Q7CharlieMessageListenerクラス：

@Component(
	property = "destination.name=acme/m4q7_able",
	service = MessageListener.class
)
public class M4Q7CharlieMessageListener implements MessageListener {

	@Override
	public void receive(Message message) {
		if (_log.isInfoEnabled()) {
			_log.info("Received message payload " + message.getPayload());
		}

		_messageBus.sendMessage(
			message.getResponseDestinationName(),
			new Message() {
				{
					setPayload("M4Q7CharlieMessageListener");
					setResponseId(message.getResponseId());
				}
			});
	}

	private static final Log _log = LogFactoryUtil.getLog(
		M4Q7CharlieMessageListener.class);

	@Reference
	private MessageBus _messageBus;

}

M4Q7CharlieMessageListenerがメッセージを受信すると、そのreceive(Message)メソッドはメッセージペイロードをログに記録し、元のメッセージの応答先に応答メッセージを送信します。 このメソッドは、応答メッセージのペイロードをリスナークラス名に設定し、応答メッセージIDを元のメッセージの応答IDに設定します。

m4q7-baker-implモジュールのM4Q7BakerMessageListenerクラスは、M4Q7BakerOSGiCommandsのメッセージの応答先であるacme/m4q7_bakerに送信されたメッセージをリッスンします。

M4Q7BakerMessageListenerクラス：

@Component(
	property = "destination.name=acme/m4q7_baker",
	service = MessageListener.class
)
public class M4Q7BakerMessageListener implements MessageListener {

	@Override
	public void receive(Message message) {
		if (_log.isInfoEnabled()) {
			_log.info("Received message payload " + message.getPayload());
		}
	}

	private static final Log _log = LogFactoryUtil.getLog(
		M4Q7BakerMessageListener.class);

}

M4Q7BakerMessageListenerがメッセージを受信すると、そのreceive(Message)メソッドはメッセージペイロードをログに記録します。

送信者を調べる

m4q7-baker-impl モジュールの M4Q7BakerOSGiCommands クラスは、コマンド引数をメッセージで "acme/m4q7_able" 宛先に送信することをトリガーする OSGi コマンドを提供します。

@Component(
	property = {"osgi.command.function=sendMessage", "osgi.command.scope=m4q7"},
	service = M4Q7BakerOSGiCommands.class
)
public class M4Q7BakerOSGiCommands {

	public void sendMessage(String payload) throws MessageBusException {
		Message message = new Message();

		message.setPayload(payload);
		message.setResponseDestinationName("acme/m4q7_baker");

		Object response = _synchronousMessageSender.send(
			"acme/m4q7_able", message, 10000);

		if (_log.isInfoEnabled()) {
			_log.info("Response: " + response);
		}
	}

	private static final Log _log = LogFactoryUtil.getLog(
		M4Q7BakerOSGiCommands.class);

	@Reference(target = "(mode=DEFAULT)")
	private SynchronousMessageSender _synchronousMessageSender;

}

M4Q7BakerOSGiCommandsは、独自のクラスタイプのサービスComponentです。 @Reference アノテーションを使用して、 SynchronousMessageSender を挿入します。これは、 デフォルト モード (アノテーションの target = "(mode=DEFAULT)" 属性によって指定されます) に設定されます。

M4Q7BakerOSGiCommandsの@Componentプロパティは、m4q7スコープでsendMessageと呼ばれるGogoシェルコマンド関数を定義します。 このコマンドは入力Stringを受け取り、M4Q7BakerOSGiCommandsのsendMessage(String)メソッドにマッピングします。

sendMessage(String) メソッドは、Gogo シェル コマンドの String をペイロードとして、 "acme/m4q7_baker" を応答の宛先として、 Message を作成します。

sendMessage(String) メソッドは、 SynchronousMessageSenderの send(String, Message, long) メソッドを呼び出して、 "acme/m4q7_able" 送信先名、メッセージ インスタンス、および 10000 ミリ秒のタイムアウトを渡してメッセージを送信します。 デフォルトモードでは、SynchronousMessageSenderはメッセージバススレッドを使用してメッセージをメッセージリスナーに配信します。 元のメッセージの応答 ID を持つメッセージが "acme/m4q7_baker" 応答宛先で受信されるまで、実行は M4Q7BakerOSGiCommands クラスでブロックされます。 レスポンスを受信すると、 M4Q7BakerOSGiCommands sendMessage(String) メソッドで実行が継続され、メッセージ応答をログに記録します。 一致する応答メッセージを受信する前にタイムアウトが期限切れになると、SynchronousMessageSenderのsend(String, Message, long)メソッドはMessageBusExceptionをスローします。

メッセージリスナーが応答メッセージを返すのを確認したので、応答のタイムアウトをテストできます。

応答タイムアウトのデモを実行する

メッセージ応答ロジックをオフにしてタイムアウトを強制する方法は次のとおりです。

1. M4Q7CharlieMessageListenerのreceive(Message)メソッドで、_messageBus.sendMessage(...)呼び出しをコメントアウトします。

   @Override
   public void receive(Message message) {
       if (_log.isInfoEnabled()) {
           Object payload = message.getPayload();
   
           _log.info("Received message payload " + payload.toString());
       }
   
       // _messageBus.sendMessage(
       //     message.getResponseDestinationName(),
       //     new Message() {
       //         {
       //             setPayload("M4Q7CharlieMessageListener");
       //             setResponseId(message.getResponseId());
       //         }
       //     });
   }
   

2. サンプルプロジェクトを再デプロイします。

   ./gradlew deploy -Ddeploy.docker.container.id=$(docker ps -lq)
   

3. Gogoシェルコマンドフィールドに、m4q7:sendMessageと入力し、その後にメッセージを入力します。 例えば、

   m4q7:sendMessage foo
   

4. Gogoシェルページが次のようになっていることを確認します。

   

5. Dockerコンソールのメッセージが次のようになっていることを確認します。

   INFO  [acme/m4q7_able-2][M4Q7CharlieMessageListener:23] Received message payload foo
   

M4Q7CharlieMessageListenerはメッセージを受信しましたが、応答しませんでした。 SynchronousMessageSenderは、Gogoシェルページに出力されたMessageBusExceptionをスローしました。

タイムアウトと同期してメッセージを送信しました。

次のステップ

ダイレクト モードを使用した同期メッセージングについて詳しく知りたい場合は、「 以前のバージョンでのダイレクト同期メッセージングの使用」を参照してください。

メッセージを送信した後すぐに処理を続行する場合は、「 非同期メッセージングの使用」を参照してください。

関連トピック

• メッセージバス
• メッセージのリッスン
• 非同期メッセージングの使用
• 登録イベントをリッスンする