ヘッダーにあるように、Esper と RabbitMQ を組み合わせてプロジェクトをビルドする際に問題があります。したがって、Esperio (バージョン 4.10) の amqp アダプターを使用する必要があります。
マニュアルの基本から始めましたが、何らかの理由でクエリが適切に評価されていません。amqp アダプターの宣言で「EventBusSink」と「EventBusSource」を切り替える必要があることはすでにわかっています (「出力ストリームが定義されていません」などのビルド エラー)。
次のように Spring AMQP を使用して、Rabbitmq に 2 つのキューを設定しています。
<rabbit:connection-factory id="connectionFactory" />
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" />
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:queue name="myQueue" durable="false">
<rabbit:queue-arguments value-type="java.lang.Long">
<entry key="x-message-ttl" value="50000" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue name="myOutputQueue" durable="false">
<rabbit:queue-arguments value-type="java.lang.Long">
<entry key="x-message-ttl" value="50000" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:listener-container
connection-factory="connectionFactory">
<rabbit:listener ref="foo" method="listen"
queue-names="myOutputQueue" />
</rabbit:listener-container>
<bean id="foo" class=" messagebroker.Foo" />
Esper と amqp アダプターを次のように構成します。
Configuration config = new Configuration();
config.addEventType("InputEvent", Event.class);
config.addEventType("OutputEvent", OutputEvent.class);
config.getEngineDefaults().getLogging().setEnableTimerDebug(false);
config.getEngineDefaults().getLogging().setEnableExecutionDebug(true);
config.getEngineDefaults().getLogging().setEnableQueryPlan(true);
epService = EPServiceProviderManager.getDefaultProvider(config);
EPRuntime cepRT = epService.getEPRuntime();
EPAdministrator cepAdm = epService.getEPAdministrator();
cepAdm.getConfiguration().addImport(AMQPSource.class.getPackage().getName() + ".*");
String epl = "Create Dataflow AMQPIncomingDataFlow \n"
// + "Create schema test as (id string), \n"
+ "AMQPSource -> instream<InputEvent> \n"
+ "{host: 'localhost'," + "port: 5672," + "username:'guest',"
+ "password:'guest'," + "queueName: 'myQueue',"
+ "declareDurable: false," + "declareExclusive: false,"
+ "declareAutoDelete: false," + "prefetchCount : 100,"
+ "waitMSecNextMsg: 0," + "consumeAutoAck: true,"
+ "collector: {class: 'AMQPToObjectCollectorSerializable'},"
+ "logMessages: true } \n" + "EventBusSink(instream){}";
cepAdm.createEPL(epl);
EPDataFlowInstance instance = cepRT.getDataFlowRuntime().instantiate(
"AMQPIncomingDataFlow");
instance.start();
cepAdm.getConfiguration().addImport(
AMQPSink.class.getPackage().getName() + ".*");
String epl2 = "Create Dataflow AMQPOutgoingDataFlow \n"
+ "EventBusSource -> outstream<OutputEvent>{} \n"
+ "AMQPSink(outstream)"
+ "{host: 'localhost',"
+ "port: 5672,"
+ "username:'guest',"
+ "password:'guest',"
+ "queueName: 'myOutputQueue',"
+ "declareDurable: false,"
+ "declareExclusive: false,"
+ "declareAutoDelete: false,"
+ "waitMSecNextMsg: 0,"
+ "collector: {class: 'ObjectToAMQPCollectorSerializable'}, logMessages: true }";
cepAdm.createEPL(epl2);
EPDataFlowInstance instance2 = cepRT.getDataFlowRuntime().instantiate(
"AMQPOutgoingDataFlow");
instance2.start();
構成を設定し、1 つのメソッド (コンストラクターまたは init メソッド) で複数のステートメントを発行すると、すべてが正常に機能します。しかし、最初にカスタム構成 (アダプターを使用) で esper を開始し、後のフェーズでステートメントを発行する (そして、後でそれらのステートメントに一致するイベントを送信する) と、ステートメントと一致するものは何もありませんでした。それにもかかわらず、同じ esperProvider (コードでは epService) を使用します。ステートメントを発行するメソッドで構成部分を複製すると、アダプターが既に存在するという例外が発生します。
クエリの例は次のとおりです。
EPStatement statement = epService.getEPAdministrator().createEPL("insert into OutputEvent select e.id as id, e.id as id, e.sourceRef as sourceRef, 'Start' as eventType from pattern[every e = InputEvent((sourceRef = '28853a2b6a88477197d88ee9f89d2ab7' or sourceRef = '39b648697f654b2891e79077d3a18fe6' or sourceRef = 'ea0db5e30d244d4fa6e922bf21c89e6f') and standardEvent = 'create')]");
そして送信されたイベント：
epService.getEPRuntime().sendEvent(new InputEvent (50, 51, "28853a2b6a88477197d88ee9f89d2ab7", "item", 50, false, null, "create"));
epService.getEPRuntime().sendEvent(new InputEvent (50, 52, "f77537e468c84adfa117d6aab48f929c", " item ", 50, false, null, "create"));
epService.getEPRuntime().sendEvent(new InputEvent (50, 53, "1ea4c8af03354a1da134e33783b22b24", " item ",50, false, null, "create"));
epService.getEPRuntime().sendEvent(new InputEvent (50, 50, "ab1ead380b6b48bdb3872de4277195a0", " item ",-1, false, null, "create"));
AEsper には、実行時にステートメントを公開するための制限はありますか?
ありがとう！
ジャナ