// Path: core/src/main/java/com/github/fhuss/kafka/streams/cep/core/state/internal/Aggregate.java
// public class Aggregate {
// 
//     private String name;
//     private Long sequence;
// 
//     /**
//      * Creates a new {@link Aggregate} instance.
//      * @param name      the aggregate name.
//      * @param sequence  the sequence id.
//      */
//     public Aggregate(final String name, final long sequence) {
//         this.name = name;
//         this.sequence = sequence;
//     }
// 
//     public String getName() {
//         return name;
//     }
// 
//     public Long getSequence() {
//         return sequence;
//     }
// 
//     /**
//      * Sets the sequence identifier of this aggregate.
//      *
//      * @param sequence  the new identifier.
//      * @return  a new {@link Aggregate} instance.
//      */
//     public Aggregate setSequence(final long sequence) {
//         return new Aggregate(name, sequence);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
//         Aggregate aggregate = (Aggregate) o;
//         return Objects.equals(name, aggregate.name) &&
//                 Objects.equals(sequence, aggregate.sequence);
//     }
// 
//     @Override
//     public int hashCode() {
//         return Objects.hash(name, sequence);
//     }
// 
//     @Override
//     public String toString() {
//         return "Aggregate{" +
//                 "name='" + name + '\'' +
//                 ", sequence=" + sequence +
//                 '}';
//     }
// }
// 
// Path: core/src/main/java/com/github/fhuss/kafka/streams/cep/core/state/internal/Aggregated.java
// public class Aggregated<K> {
// 
//     private final K key;
// 
//     private final Aggregate aggregate;
// 
//     /**
//      * Creates a new {@link Aggregated} instance.
//      * @param key       the record key
//      * @param aggregate the instance of {@link Aggregate}.
//      */
//     public Aggregated(final K key, final Aggregate aggregate) {
//         this.key = key;
//         this.aggregate = aggregate;
//     }
// 
//     public K getKey() {
//         return key;
//     }
// 
//     public Aggregate getAggregate() {
//         return aggregate;
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
//         Aggregated<?> that = (Aggregated<?>) o;
//         return Objects.equals(key, that.key) &&
//                 Objects.equals(aggregate, that.aggregate);
//     }
// 
//     @Override
//     public int hashCode() {
// 
//         return Objects.hash(key, aggregate);
//     }
// 
//     @Override
//     public String toString() {
//         return "Aggregated{" +
//                 "key=" + key +
//                 ", aggregate=" + aggregate +
//                 '}';
//     }
// }


import com.github.fhuss.kafka.streams.cep.core.state.internal.Aggregate;
import com.github.fhuss.kafka.streams.cep.core.state.internal.Aggregated;
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.github.fhuss.kafka.streams.cep.core.state;


/**
 * Simple class to wrap a {@link AggregatesStore}.
 *
 * @param <K>   the record key type.
 */
public class States<K> {

    private final AggregatesStore<K> store;

    private final long sequence;

    private final K key;

    /**
     * Creates a new {@link States} instance.
     *
     * @param states    the aggregates store.
     * @param key       the record key used to store aggregates.
     * @param sequence  The sequence number (aka run) used to store aggregates.
     */
    public States(final AggregatesStore<K> states, K key, long sequence) {
        this.store = states;
        this.key = key;
        this.sequence = sequence;
    }

    /**
     * Retrieve the value state for the specified key.
     *
     * @param state the state name.
     * @param <T> the of default value.
     * @return <code>null</code> if no state exists for the given key.
     */
    public <T> T get(final String state) {
        T v = getOrNull(state);
        if (v == null) throw new UnknownAggregateException(state);
        return v;
    }

    /**
     * Retrieve the value state for the specified key.
     *
     * @param key the object key.
     * @param def the default value.
     * @param <T> the of default value.
     * @return {@literal def} if no state exists for the given key.
     */
    public <T> T getOrElse(String key, T def) {
        T val = getOrNull(key);
        return val != null ? val : def;
    }

    private <T> T getOrNull(String state) {
// Path: core/src/main/java/com/github/fhuss/kafka/streams/cep/core/state/internal/Aggregate.java
// public class Aggregate {
// 
//     private String name;
//     private Long sequence;
// 
//     /**
//      * Creates a new {@link Aggregate} instance.
//      * @param name      the aggregate name.
//      * @param sequence  the sequence id.
//      */
//     public Aggregate(final String name, final long sequence) {
//         this.name = name;
//         this.sequence = sequence;
//     }
// 
//     public String getName() {
//         return name;
//     }
// 
//     public Long getSequence() {
//         return sequence;
//     }
// 
//     /**
//      * Sets the sequence identifier of this aggregate.
//      *
//      * @param sequence  the new identifier.
//      * @return  a new {@link Aggregate} instance.
//      */
//     public Aggregate setSequence(final long sequence) {
//         return new Aggregate(name, sequence);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
//         Aggregate aggregate = (Aggregate) o;
//         return Objects.equals(name, aggregate.name) &&
//                 Objects.equals(sequence, aggregate.sequence);
//     }
// 
//     @Override
//     public int hashCode() {
//         return Objects.hash(name, sequence);
//     }
// 
//     @Override
//     public String toString() {
//         return "Aggregate{" +
//                 "name='" + name + '\'' +
//                 ", sequence=" + sequence +
//                 '}';
//     }
// }
// 
// Path: core/src/main/java/com/github/fhuss/kafka/streams/cep/core/state/internal/Aggregated.java
// public class Aggregated<K> {
// 
//     private final K key;
// 
//     private final Aggregate aggregate;
// 
//     /**
//      * Creates a new {@link Aggregated} instance.
//      * @param key       the record key
//      * @param aggregate the instance of {@link Aggregate}.
//      */
//     public Aggregated(final K key, final Aggregate aggregate) {
//         this.key = key;
//         this.aggregate = aggregate;
//     }
// 
//     public K getKey() {
//         return key;
//     }
// 
//     public Aggregate getAggregate() {
//         return aggregate;
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
//         Aggregated<?> that = (Aggregated<?>) o;
//         return Objects.equals(key, that.key) &&
//                 Objects.equals(aggregate, that.aggregate);
//     }
// 
//     @Override
//     public int hashCode() {
// 
//         return Objects.hash(key, aggregate);
//     }
// 
//     @Override
//     public String toString() {
//         return "Aggregated{" +
//                 "key=" + key +
//                 ", aggregate=" + aggregate +
//                 '}';
//     }
// }


// Path: core/src/main/java/com/github/fhuss/kafka/streams/cep/core/state/States.java
import com.github.fhuss.kafka.streams.cep.core.state.internal.Aggregate;
import com.github.fhuss.kafka.streams.cep.core.state.internal.Aggregated;

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.github.fhuss.kafka.streams.cep.core.state;


/**
 * Simple class to wrap a {@link AggregatesStore}.
 *
 * @param <K>   the record key type.
 */
public class States<K> {

    private final AggregatesStore<K> store;

    private final long sequence;

    private final K key;

    /**
     * Creates a new {@link States} instance.
     *
     * @param states    the aggregates store.
     * @param key       the record key used to store aggregates.
     * @param sequence  The sequence number (aka run) used to store aggregates.
     */
    public States(final AggregatesStore<K> states, K key, long sequence) {
        this.store = states;
        this.key = key;
        this.sequence = sequence;
    }

    /**
     * Retrieve the value state for the specified key.
     *
     * @param state the state name.
     * @param <T> the of default value.
     * @return <code>null</code> if no state exists for the given key.
     */
    public <T> T get(final String state) {
        T v = getOrNull(state);
        if (v == null) throw new UnknownAggregateException(state);
        return v;
    }

    /**
     * Retrieve the value state for the specified key.
     *
     * @param key the object key.
     * @param def the default value.
     * @param <T> the of default value.
     * @return {@literal def} if no state exists for the given key.
     */
    public <T> T getOrElse(String key, T def) {
        T val = getOrNull(key);
        return val != null ? val : def;
    }

    private <T> T getOrNull(String state) {
        Aggregated<K> aggregated = new Aggregated<>(key, new Aggregate(state, sequence));