java - Apache Flink によるステートフルな複雑なイベント処理

Question

同じ識別子を持つ 2 つのイベントに基づいて、定義された時間枠内に 2 つのイベントが発生したかどうかを検出したいと考えています。たとえば、DoorEvent次のようになります。

<doorevent>
  <door>
    <id>1</id>
    <status>open</status>
  </door>
  <timestamp>12345679</timestamp>
</doorevent> 

<doorevent>
  <door>
    <id>1</id>
    <status>close</status>
  </door>
  <timestamp>23456790</timestamp>
</doorevent>

以下の例の私のDoorEventJava クラスは、同じ構造を持っています。

ID 1 のドアが開いてから 5 分以内に閉じることを検出したいと考えています。この目的のために、Apache flink CEP ライブラリを使用しようとしています。着信ストリームには、たとえば 20 のドアからのすべての開閉メッセージが含まれています。

Pattern<String, ?> pattern = Pattern.<String>begin("door_open").where(
    new SimpleCondition<String>() {
        private static final long serialVersionUID = 1L;
        public boolean filter(String doorevent) {
            DoorEvent event = new DoorEvent().parseInstance(doorevent, DataType.XML);
            if (event.getDoor().getStatus().equals("open")){
                // save state of door as open
                return true;
            }
            return false;                           
        }
    }
)
.followedByAny("door_close").where(
    new SimpleCondition<String>() {
            private static final long serialVersionUID = 1L;
            public boolean filter(String doorevent) throws JsonParseException, JsonMappingException, IOException {
                DoorEvent event = new DoorEvent().parseInstance(doorevent, DataType.XML);
                if (event.getDoor().getStatus().equals("close")){
                    // check if close is of previously opened door
                    return true;
                }
                return false;
            }
        }
)
.within(Time.minutes(5));

ステップでドア 1 が閉じているドアであり、他のドアではないことがわかるdoor_openように、ドア 1 の状態を開いている状態として保存するにはどうすればよいですか?door_close