You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2021/05/10 20:48:37 UTC
[ignite-3] 08/09: WIP
This is an automated email from the ASF dual-hosted git repository.
agura pushed a commit to branch ignite-14389
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit a858e14b71af45a4734f48aa2c86a8ae41c3bd36
Author: Andrey Gura <ag...@apache.org>
AuthorDate: Wed May 5 18:25:00 2021 +0300
WIP
---
.../ignite/internal/affinity/AffinityManager.java | 32 +-
.../org/apache/ignite/internal/util}/Cursor.java | 10 +-
.../java/org/apache/ignite/lang/ByteArray.java | 16 +-
.../metastorage/client}/CompactedException.java | 2 +-
.../ignite/metastorage/client}/Condition.java | 228 +++----
.../ignite/metastorage/client}/Conditions.java | 48 +-
.../apache/ignite/metastorage/client}/Entry.java | 5 +-
.../ignite/metastorage/client/EntryEvent.java} | 18 +-
.../org/apache/ignite/metastorage/client}/Key.java | 2 +-
.../metastorage/client/MetaStorageService.java | 105 ++--
.../ignite/metastorage/client}/Operation.java | 47 +-
.../client}/OperationTimeoutException.java | 2 +-
.../ignite/metastorage/client}/Operations.java | 14 +-
.../ignite/metastorage/client}/WatchEvent.java | 19 +-
.../ignite/metastorage/client}/WatchListener.java | 6 +-
modules/metastorage-server/pom.xml | 6 +-
.../{EntryEvent.java => AbstractCondition.java} | 44 +-
.../server/{EntryEvent.java => Condition.java} | 44 +-
.../internal/metastorage/server/EntryEvent.java | 2 +-
.../metastorage/server/ExistenceCondition.java | 74 +++
.../metastorage/server/KeyValueStorage.java | 4 +-
.../internal/metastorage/server/Operation.java | 111 ++++
.../metastorage/server/RevisionCondition.java | 110 ++++
.../server/SimpleInMemoryKeyValueStorage.java | 136 ++++-
.../metastorage/server/ValueCondition.java | 82 +++
.../internal/metastorage/server/WatchEvent.java | 11 +-
.../metastorage/server/ExistenceConditionTest.java | 71 +++
.../metastorage/server/RevisionConditionTest.java | 113 ++++
.../server/SimpleInMemoryKeyValueStorageTest.java | 668 +++++++++++++++++++--
.../metastorage/server/ValueConditionTest.java | 60 ++
.../internal/metastorage/MetaStorageManager.java | 105 ++--
.../metastorage/watch/AggregatedWatch.java | 2 +-
.../internal/metastorage/watch/KeyCriterion.java | 44 +-
.../metastorage/watch/WatchAggregator.java | 49 +-
.../internal/metastorage/WatchAggregatorTest.java | 91 +--
.../apache/ignite/internal/app/IgnitionImpl.java | 95 ++-
.../internal/table/distributed/TableManager.java | 34 +-
37 files changed, 1899 insertions(+), 611 deletions(-)
diff --git a/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java b/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java
index 37690bf..a13d041 100644
--- a/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java
+++ b/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java
@@ -31,11 +31,11 @@ import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
-import org.apache.ignite.metastorage.common.Conditions;
-import org.apache.ignite.metastorage.common.Key;
-import org.apache.ignite.metastorage.common.Operations;
-import org.apache.ignite.metastorage.common.WatchEvent;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.metastorage.client.Conditions;
+import org.apache.ignite.metastorage.client.EntryEvent;
+import org.apache.ignite.metastorage.client.Operations;
+import org.apache.ignite.metastorage.client.WatchEvent;
+import org.apache.ignite.metastorage.client.WatchListener;
import org.jetbrains.annotations.NotNull;
/**
@@ -130,9 +130,9 @@ public class AffinityManager {
String tableInternalPrefix = INTERNAL_PREFIX + "assignment.";
- affinityCalculateSubscriptionFut = metaStorageMgr.registerWatchByPrefix(new Key(tableInternalPrefix), new WatchListener() {
- @Override public boolean onUpdate(@NotNull Iterable<WatchEvent> events) {
- for (WatchEvent evt : events) {
+ affinityCalculateSubscriptionFut = metaStorageMgr.registerWatchByPrefix(new ByteArray(tableInternalPrefix), new WatchListener() {
+ @Override public boolean onUpdate(@NotNull WatchEvent watchEvt) {
+ for (EntryEvent evt : watchEvt.entryEvents()) {
byte[] assignmentVal = evt.newEntry().value();
if (assignmentVal != null && assignmentVal.length == 0) {
@@ -143,16 +143,20 @@ public class AffinityManager {
UUID tblId = UUID.fromString(placeholderValue);
try {
- String name = new String(vaultManager.get(ByteArray.fromString(INTERNAL_PREFIX + tblId.toString())).get().value(), StandardCharsets.UTF_8);
+ byte[] tblNameVal = vaultManager.get(ByteArray.fromString(INTERNAL_PREFIX + tblId)).get().value();
- int partitions = configurationMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY)
- .tables().get(name).partitions().value();
- int replicas = configurationMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY)
- .tables().get(name).replicas().value();
+ String name = new String(tblNameVal, StandardCharsets.UTF_8);
+
+ int partitions = configurationMgr.configurationRegistry()
+ .getConfiguration(TablesConfiguration.KEY).tables().get(name).partitions().value();
+
+ int replicas = configurationMgr.configurationRegistry()
+ .getConfiguration(TablesConfiguration.KEY).tables().get(name).replicas().value();
var key = evt.newEntry().key();
+
metaStorageMgr.invoke(
- Conditions.key(key).value().eq(assignmentVal),
+ Conditions.value(key).eq(assignmentVal),
Operations.put(key, ByteUtils.toBytes(
RendezvousAffinityFunction.assignPartitions(
baselineMgr.nodes(),
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Cursor.java b/modules/core/src/main/java/org/apache/ignite/internal/util/Cursor.java
similarity index 79%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Cursor.java
rename to modules/core/src/main/java/org/apache/ignite/internal/util/Cursor.java
index dcc1c94..e1ddaf2 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Cursor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/Cursor.java
@@ -15,13 +15,15 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.internal.util;
+
+import java.util.Iterator;
/**
* Closeable cursor.
*
- * @param <T>
+ * @param <T> Type of elements.
*/
-//TODO: Should be replaced by common entity from org.ignite.lang or org.ignite.core package.
-public interface Cursor<T> extends Iterable<T>, AutoCloseable {
+public interface Cursor<T> extends Iterator<T>, Iterable<T>, AutoCloseable {
+
}
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/ByteArray.java b/modules/core/src/main/java/org/apache/ignite/lang/ByteArray.java
index c38be87..9f6013a 100644
--- a/modules/core/src/main/java/org/apache/ignite/lang/ByteArray.java
+++ b/modules/core/src/main/java/org/apache/ignite/lang/ByteArray.java
@@ -22,15 +22,16 @@ import java.util.Arrays;
import org.jetbrains.annotations.NotNull;
/**
- * A class for handling byte array.
+ * A class wraps {@code byte[]} which provides {@code equals}, {@code hashCode} and lexicographical comparison.
*/
public final class ByteArray implements Comparable<ByteArray> {
- /** Byte-wise representation of the {@code ByteArray}. */
+ /** Wrapped byte array. */
@NotNull
private final byte[] arr;
/**
- * Constructs {@code ByteArray} instance from the given byte array. <em>Note:</em> copy of the given byte array will not be
+ * Constructs {@code ByteArray} instance from the given byte array.
+ * <em>Note:</em> copy of the given byte array will not be
* created in order to avoid redundant memory consumption.
*
* @param arr Byte array. Can't be {@code null}.
@@ -40,6 +41,15 @@ public final class ByteArray implements Comparable<ByteArray> {
}
/**
+ * Constructs {@code ByteArray} instance from the given string.
+ *
+ * @param s The string key representation. Can't be {@code null}.
+ */
+ public ByteArray(@NotNull String s) {
+ this(s.getBytes(StandardCharsets.UTF_8));
+ }
+
+ /**
* Constructs {@code ByteArray} instance from the given string. {@link StandardCharsets#UTF_8} charset is used for
* encoding the input string.
*
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/CompactedException.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/CompactedException.java
similarity index 97%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/CompactedException.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/CompactedException.java
index 6ea9f43..c256442 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/CompactedException.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/CompactedException.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
/**
* Thrown when a requested operation on meta storage could not be performed because target revisions were removed
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Condition.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Condition.java
similarity index 67%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Condition.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Condition.java
index 54049de..e060b8d 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Condition.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Condition.java
@@ -15,12 +15,15 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
-import java.util.Arrays;
+import java.util.Collection;
/**
- * Represents a condition for conditional update.
+ * Represents a condition for meta storage conditional update.
+ *
+ * @see MetaStorageService#invoke(Condition, Operation, Operation)
+ * @see MetaStorageService#invoke(Condition, Collection, Collection)
*/
public final class Condition {
/** Actual condition implementation. */
@@ -36,22 +39,11 @@ public final class Condition {
}
/**
- * Tests the given entry on satisfaction of the condition.
- *
- * @param e Entry.
- * @return The result of condition test. {@code true} - if the entry satisfies to the condition,
- * otherwise - {@code false}.
- */
- public boolean test(Entry e) {
- return cond.test(e);
- }
-
- /**
* Represents condition on entry revision. Only one type of condition could be applied to
* the one instance of condition. Subsequent invocations of any method which produces condition will throw
* {@link IllegalStateException}.
*/
- public static final class RevisionCondition implements InnerCondition {
+ public static final class RevisionCondition extends AbstractCondition {
/**
* The type of condition.
*
@@ -62,16 +54,13 @@ public final class Condition {
/** The revision as the condition argument. */
private long rev;
- /** Key of entry, which will be tested for condition. */
- private final Key key;
-
/**
- * Creates a new condition for the given {@code key}.
+ * Constructs a condition by a revision for an entry identified by the given key.
*
- * @param key Key of entry, to be tested for the condition.
+ * @param key Identifies an entry which condition will be applied to.
*/
- RevisionCondition(Key key) {
- this.key = key;
+ RevisionCondition(byte[] key) {
+ super(key);
}
/**
@@ -176,70 +165,27 @@ public final class Condition {
return new Condition(this);
}
- /** {@inheritDoc} */
- @Override public boolean test(Entry e) {
- if ((e.key() == key) || (e.key() != null && e.key().equals(key))) {
- int res = Long.compare(e.revision(), rev);
-
- return type.test(res);
- }
- else
- return false;
- }
-
/**
* Defines possible condition types which can be applied to the revision.
*/
enum Type {
/** Equality condition type. */
- EQUAL {
- @Override public boolean test(long res) {
- return res == 0;
- }
- },
+ EQUAL,
/** Inequality condition type. */
- NOT_EQUAL {
- @Override public boolean test(long res) {
- return res != 0;
- }
- },
+ NOT_EQUAL,
/** Greater than condition type. */
- GREATER {
- @Override public boolean test(long res) {
- return res > 0;
- }
- },
+ GREATER,
/** Less than condition type. */
- LESS {
- @Override public boolean test(long res) {
- return res < 0;
- }
- },
+ LESS,
/** Less than or equal to condition type. */
- LESS_OR_EQUAL {
- @Override public boolean test(long res) {
- return res <= 0;
- }
- },
+ LESS_OR_EQUAL,
/** Greater than or equal to condition type. */
- GREATER_OR_EQUAL {
- @Override public boolean test(long res) {
- return res >= 0;
- }
- };
-
- /**
- * Interprets comparison result.
- *
- * @param res The result of comparison.
- * @return The interpretation of the comparison result.
- */
- public abstract boolean test(long res);
+ GREATER_OR_EQUAL
}
}
@@ -248,7 +194,7 @@ public final class Condition {
* the one instance of condition. Subsequent invocations of any method which produces condition will throw
* {@link IllegalStateException}.
*/
- public static final class ValueCondition implements InnerCondition {
+ public static final class ValueCondition extends AbstractCondition {
/**
* The type of condition.
*
@@ -259,16 +205,13 @@ public final class Condition {
/** The value as the condition argument. */
private byte[] val;
- /** Key of entry, which will be tested for condition. */
- private final Key key;
-
/**
- * Creates a new condition for the given {@code key}.
+ * Constructs a condition by a value for an entry identified by the given key.
*
- * @param key Key of entry, to be tested for the condition.
+ * @param key Identifies an entry which condition will be applied to.
*/
- ValueCondition(Key key) {
- this.key = key;
+ ValueCondition(byte[] key) {
+ super(key);
}
/**
@@ -305,42 +248,79 @@ public final class Condition {
return new Condition(this);
}
- /** {@inheritDoc} */
- @Override public boolean test(Entry e) {
- if ((e.key() == key) || (e.key() != null && e.key().equals(key))) {
- int res = Arrays.compare(e.value(), val);
+ /**
+ * Defines possible condition types which can be applied to the value.
+ */
+ enum Type {
+ /** Equality condition type. */
+ EQUAL,
- return type.test(res);
- }
- else
- return false;
+ /** Inequality condition type. */
+ NOT_EQUAL
+ }
+ }
+
+ /**
+ * Represents condition on an entry existence. Only one type of condition could be applied to
+ * the one instance of condition. Subsequent invocations of any method which produces condition will throw
+ * {@link IllegalStateException}.
+ */
+ public static final class ExistenceCondition extends AbstractCondition {
+ /**
+ * The type of condition.
+ *
+ * @see Type
+ */
+ private Type type;
+
+ /**
+ * Constructs a condition on existence an entry identified by the given key.
+ *
+ * @param key Identifies an entry which condition will be applied to.
+ */
+ ExistenceCondition(byte[] key) {
+ super(key);
+ }
+
+ /**
+ * Produces the condition of type {@link Type#EXISTS}. This condition tests the existence of an entry
+ * identified by the given key.
+ *
+ * @return The condition of type {@link Type#EXISTS}.
+ * @throws IllegalStateException In case when the condition is already defined.
+ */
+ public Condition exists() {
+ validate(type);
+
+ this.type = Type.EXISTS;
+
+ return new Condition(this);
+ }
+
+ /**
+ * Produces the condition of type {@link Type#NOT_EXISTS}. This condition tests the non-existence of an entry
+ * identified by the given key.
+ *
+ * @return The condition of type {@link Type#NOT_EXISTS}.
+ * @throws IllegalStateException In case when the condition is already defined.
+ */
+ public Condition notExists() {
+ validate(type);
+
+ this.type = Type.NOT_EXISTS;
+
+ return new Condition(this);
}
/**
* Defines possible condition types which can be applied to the value.
*/
enum Type {
- /** Equality condition type. */
- EQUAL {
- @Override public boolean test(long res) {
- return res == 0;
- }
- },
+ /** Existence condition type. */
+ EXISTS,
- /** Inequality condition type. */
- NOT_EQUAL {
- @Override public boolean test(long res) {
- return res != 0;
- }
- };
-
- /**
- * Interprets comparison result.
- *
- * @param res The result of comparison.
- * @return The interpretation of the comparison result.
- */
- public abstract boolean test(long res);
+ /** Non-existence condition type. */
+ NOT_EXISTS
}
}
@@ -359,12 +339,36 @@ public final class Condition {
*/
private interface InnerCondition {
/**
- * Tests the given entry on satisfaction of the condition.
+ * Returns key which identifies an entry which condition will be applied to.
+ *
+ * @return Key which identifies an entry which condition will be applied to.
+ */
+ byte[] key();
+ }
+
+ /**
+ * Defines an abstract condition with the key which identifies an entry which condition will be applied to.
+ */
+ private static abstract class AbstractCondition implements InnerCondition {
+ /** Entry key. */
+ private final byte[] key;
+
+ /**
+ * Constructs a condition with the given entry key.
*
- * @param e Entry.
- * @return The result of condition test. {@code true} - if the entry satisfies to the condition,
- * otherwise - {@code false}.
+ * @param key Key which identifies an entry which condition will be applied to.
*/
- boolean test(Entry e);
+ public AbstractCondition(byte[] key) {
+ this.key = key;
+ }
+
+ /**
+ * Returns the key which identifies an entry which condition will be applied to.
+ *
+ * @return Key which identifies an entry which condition will be applied to.
+ */
+ @Override public byte[] key() {
+ return key;
+ }
}
}
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Conditions.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Conditions.java
similarity index 54%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Conditions.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Conditions.java
index 87fa238..3d4f370 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Conditions.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Conditions.java
@@ -15,7 +15,10 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
+
+import org.apache.ignite.lang.ByteArray;
+import org.jetbrains.annotations.NotNull;
/**
* This class contains fabric methods which produce conditions needed for conditional multi update functionality
@@ -24,47 +27,46 @@ package org.apache.ignite.metastorage.common;
* @see Condition
*/
public final class Conditions {
-
- /** Key. */
- private Key key;
-
- /**
- * Creates new condition for entry with concrete key.
- *
- * @param key Key
- */
- private Conditions(Key key) {
- this.key = key;
- }
-
/**
* Creates condition on entry revision.
*
+ * @param key Identifies an entry which condition will be applied to. Can't be {@code null}.
* @return Condition on entry revision.
* @see Condition.RevisionCondition
*/
- public Condition.RevisionCondition revision() {
- return new Condition.RevisionCondition(key);
+ public static Condition.RevisionCondition revision(@NotNull ByteArray key) {
+ return new Condition.RevisionCondition(key.bytes());
}
/**
* Creates condition on entry value.
*
+ * @param key Identifies an entry which condition will be applied to. Can't be {@code null}.
* @return Condition on entry value.
* @see Condition.ValueCondition
*/
- public Condition.ValueCondition value() {
- return new Condition.ValueCondition(key);
+ public static Condition.ValueCondition value(@NotNull ByteArray key) {
+ return new Condition.ValueCondition(key.bytes());
+ }
+
+ /**
+ * Creates condition on entry existence.
+ *
+ * @param key Identifies an entry which condition will be applied to. Can't be {@code null}.
+ * @return Condition on entry existence.
+ */
+ public static Condition exists(@NotNull ByteArray key) {
+ return new Condition.ExistenceCondition(key.bytes()).exists();
}
/**
- * Creates key-based condition.
+ * Creates condition on entry not existence.
*
- * @param key Key of condition.
- * @return Key-based condition instance.
+ * @param key Identifies an entry which condition will be applied to. Can't be {@code null}.
+ * @return Condition on entry not existence.
*/
- public static Conditions key(Key key) {
- return new Conditions(key);
+ public static Condition notExists(@NotNull ByteArray key) {
+ return new Condition.ExistenceCondition(key.bytes()).notExists();
}
/**
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Entry.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Entry.java
similarity index 93%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Entry.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Entry.java
index e7391ba..24a9c79 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Entry.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Entry.java
@@ -15,8 +15,9 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
+import org.apache.ignite.lang.ByteArray;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -34,7 +35,7 @@ public interface Entry {
*
* @return The key.
*/
- @NotNull Key key();
+ @NotNull ByteArray key();
/**
* Returns a value. Could be {@code null} for empty entry.
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/WatchEvent.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/EntryEvent.java
similarity index 76%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/WatchEvent.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/EntryEvent.java
index bd78ea5..c1a0ee9 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/WatchEvent.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/EntryEvent.java
@@ -15,25 +15,25 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
/**
- * Watch event which can be processed by {@link WatchListener}.
+ * Represent an update event for particular key and entry.
*/
-public final class WatchEvent {
- /** Old (previous) entry */
+public final class EntryEvent {
+ /** Old (previous) entry. */
private final Entry oldEntry;
- /** New (updated) entry. */
+ /** New (current) entry. */
private final Entry newEntry;
/**
- * Constructs an event with given old and new entries.
+ * Constructs event with given old and new entries.
*
* @param oldEntry Old entry.
- * @param newEntry New entry/
+ * @param newEntry New entry.
*/
- public WatchEvent(Entry oldEntry, Entry newEntry) {
+ public EntryEvent(Entry oldEntry, Entry newEntry) {
this.oldEntry = oldEntry;
this.newEntry = newEntry;
}
@@ -48,7 +48,7 @@ public final class WatchEvent {
}
/**
- * Returns new entry.
+ * Rreturns new entry.
*
* @return New entry.
*/
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Key.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Key.java
similarity index 98%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Key.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Key.java
index d497021..8eb1c14 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Key.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Key.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
diff --git a/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/MetaStorageService.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/MetaStorageService.java
index 6c24a59..0939cc9 100644
--- a/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/MetaStorageService.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/MetaStorageService.java
@@ -20,15 +20,9 @@ package org.apache.ignite.metastorage.client;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.metastorage.common.CompactedException;
-import org.apache.ignite.metastorage.common.Condition;
-import org.apache.ignite.metastorage.common.Cursor;
-import org.apache.ignite.metastorage.common.Entry;
-import org.apache.ignite.metastorage.common.Key;
-import org.apache.ignite.metastorage.common.Operation;
-import org.apache.ignite.metastorage.common.OperationTimeoutException;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.internal.util.Cursor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -42,11 +36,11 @@ public interface MetaStorageService {
* @param key Key. Couldn't be {@code null}.
* @return An entry for the given key. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Entry> get(@NotNull Key key);
+ CompletableFuture<Entry> get(@NotNull ByteArray key);
/**
* Retrieves an entry for the given key and the revision upper bound.
@@ -58,11 +52,11 @@ public interface MetaStorageService {
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
* Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Entry> get(@NotNull Key key, long revUpperBound);
+ CompletableFuture<Entry> get(@NotNull ByteArray key, long revUpperBound);
/**
* Retrieves entries for given keys.
@@ -71,11 +65,11 @@ public interface MetaStorageService {
* Collection elements couldn't be {@code null}.
* @return A map of entries for given keys. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Map<Key, Entry>> getAll(Collection<Key> keys);
+ CompletableFuture<Map<ByteArray, Entry>> getAll(Collection<ByteArray> keys);
/**
* Retrieves entries for given keys and the revision upper bound.
@@ -88,11 +82,11 @@ public interface MetaStorageService {
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
* Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Map<Key, Entry>> getAll(Collection<Key> keys, long revUpperBound);
+ CompletableFuture<Map<ByteArray, Entry>> getAll(Collection<ByteArray> keys, long revUpperBound);
/**
* Inserts or updates an entry with the given key and the given value.
@@ -101,11 +95,11 @@ public interface MetaStorageService {
* @param value The value.Couldn't be {@code null}.
* @return Completed future.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Void> put(@NotNull Key key, @NotNull byte[] value);
+ CompletableFuture<Void> put(@NotNull ByteArray key, @NotNull byte[] value);
/**
* Inserts or updates an entry with the given key and the given value and
@@ -115,11 +109,11 @@ public interface MetaStorageService {
* @param value The value.Couldn't be {@code null}.
* @return A previous entry for the given key. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Entry> getAndPut(@NotNull Key key, @NotNull byte[] value);
+ CompletableFuture<Entry> getAndPut(@NotNull ByteArray key, @NotNull byte[] value);
/**
* Inserts or updates entries with given keys and given values.
@@ -127,11 +121,11 @@ public interface MetaStorageService {
* @param vals The map of keys and corresponding values. Couldn't be {@code null} or empty.
* @return Completed future.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Void> putAll(@NotNull Map<Key, byte[]> vals);
+ CompletableFuture<Void> putAll(@NotNull Map<ByteArray, byte[]> vals);
/**
* Inserts or updates entries with given keys and given values and
@@ -140,11 +134,11 @@ public interface MetaStorageService {
* @param vals The map of keys and corresponding values. Couldn't be {@code null} or empty.
* @return A map of entries for given keys. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Map<Key, Entry>> getAndPutAll(@NotNull Map<Key, byte[]> vals);
+ CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(@NotNull Map<ByteArray, byte[]> vals);
/**
* Removes an entry for the given key.
@@ -152,11 +146,11 @@ public interface MetaStorageService {
* @param key The key. Couldn't be {@code null}.
* @return Completed future.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Void> remove(@NotNull Key key);
+ CompletableFuture<Void> remove(@NotNull ByteArray key);
/**
* Removes an entry for the given key.
@@ -164,11 +158,11 @@ public interface MetaStorageService {
* @param key The key. Couldn't be {@code null}.
* @return A previous entry for the given key. Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Entry> getAndRemove(@NotNull Key key);
+ CompletableFuture<Entry> getAndRemove(@NotNull ByteArray key);
/**
* Removes entries for given keys.
@@ -176,11 +170,11 @@ public interface MetaStorageService {
* @param keys The keys collection. Couldn't be {@code null}.
* @return Completed future.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Void> removeAll(@NotNull Collection<Key> keys);
+ CompletableFuture<Void> removeAll(@NotNull Collection<ByteArray> keys);
/**
* Removes entries for given keys and retrieves previous entries.
@@ -190,11 +184,11 @@ public interface MetaStorageService {
* The order of entries in the result list corresponds to the traversal order of {@code keys} collection.
* Couldn't be {@code null}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<Map<Key, Entry>> getAndRemoveAll(@NotNull Collection<Key> keys);
+ CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(@NotNull Collection<ByteArray> keys);
/**
@@ -203,11 +197,11 @@ public interface MetaStorageService {
* <p>Conditional update could be treated as <i>if(condition)-then(success)-else(failure)</i> expression.</p>
*
* @param condition The condition.
- * @param success Batch of updates which will be atomically applied in case of condition evaluation yields {@code true}.
- * @param failure Batch of updates which will be atomically applied in case of condition evaluation yields {@code false}.
- * @return Future result {@code true} if {@code success} updates were applied, otherwise {@code false}.
+ * @param success The update which will be applied in case of condition evaluation yields {@code true}.
+ * @param failure The update which will be applied in case of condition evaluation yields {@code false}.
+ * @return Future result {@code true} if {@code success} update was applied, otherwise {@code false}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
* @see Condition
* @see Operation
@@ -215,28 +209,27 @@ public interface MetaStorageService {
// TODO: https://issues.apache.org/jira/browse/IGNITE-14269: will be replaced by conditional multi update.
@NotNull
CompletableFuture<Boolean> invoke(@NotNull Condition condition,
- @NotNull Collection<Operation> success, @NotNull Collection<Operation> failure);
+ @NotNull Operation success, @NotNull Operation failure);
/**
* Updates an entry for the given key conditionally.
*
* <p>Conditional update could be treated as <i>if(condition)-then(success)-else(failure)</i> expression.</p>
*
- * @param key The key. Couldn't be {@code null}.
* @param condition The condition.
- * @param success The update which will be applied in case of condition evaluation yields {@code true}.
- * @param failure The update which will be applied in case of condition evaluation yields {@code false}.
- * @return A previous entry for the given key.
+ * @param success The updates which will be applied in case of condition evaluation yields {@code true}.
+ * @param failure The updates which will be applied in case of condition evaluation yields {@code false}.
+ * @return Future result {@code true} if {@code success} update was applied, otherwise {@code false}.
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
* @see Condition
* @see Operation
*/
- //TODO: https://issues.apache.org/jira/browse/IGNITE-14269: will be replaced by conditional multi update.
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-14269: will be replaced by conditional multi update.
@NotNull
- CompletableFuture<Entry> getAndInvoke(@NotNull Key key, @NotNull Condition condition,
- @NotNull Operation success, @NotNull Operation failure);
+ CompletableFuture<Boolean> invoke(@NotNull Condition condition,
+ @NotNull Collection<Operation> success, @NotNull Collection<Operation> failure);
/**
* Retrieves entries for the given key range in lexicographic order. Entries will be filtered out by upper bound
@@ -248,26 +241,26 @@ public interface MetaStorageService {
* @return Cursor built upon entries corresponding to the given range and revision.
* @throws OperationTimeoutException If the operation is timed out.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- Cursor<Entry> range(@NotNull Key keyFrom, @Nullable Key keyTo, long revUpperBound);
+ Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound);
/**
* Retrieves entries for the given key range in lexicographic order. Short cut for
- * {@link #range(Key, Key, long)} where {@code revUpperBound == -1}.
+ * {@link #range(ByteArray, ByteArray, long)} where {@code revUpperBound == -1}.
*
* @param keyFrom Start key of range (inclusive). Couldn't be {@code null}.
* @param keyTo End key of range (exclusive). Could be {@code null}.
* @return Cursor built upon entries corresponding to the given range and revision.
* @throws OperationTimeoutException If the operation is timed out.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- Cursor<Entry> range(@NotNull Key keyFrom, @Nullable Key keyTo);
+ Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo);
/**
* Subscribes on meta storage updates matching the parameters.
@@ -281,11 +274,11 @@ public interface MetaStorageService {
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
* Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<IgniteUuid> watch(@Nullable Key keyFrom, @Nullable Key keyTo,
+ CompletableFuture<IgniteUuid> watch(@Nullable ByteArray keyFrom, @Nullable ByteArray keyTo,
long revision, @NotNull WatchListener lsnr);
/**
@@ -299,11 +292,11 @@ public interface MetaStorageService {
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
* Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<IgniteUuid> watch(@NotNull Key key, long revision, @NotNull WatchListener lsnr);
+ CompletableFuture<IgniteUuid> watch(@NotNull ByteArray key, long revision, @NotNull WatchListener lsnr);
/**
* Subscribes on meta storage updates for given keys.
@@ -316,11 +309,11 @@ public interface MetaStorageService {
* @throws OperationTimeoutException If the operation is timed out. Will be thrown on getting future result.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
* Will be thrown on getting future result.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
@NotNull
- CompletableFuture<IgniteUuid> watch(@NotNull Collection<Key> keys, long revision, @NotNull WatchListener lsnr);
+ CompletableFuture<IgniteUuid> watch(@NotNull Collection<ByteArray> keys, long revision, @NotNull WatchListener lsnr);
/**
* Cancels subscription for the given identifier.
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Operation.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Operation.java
similarity index 68%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Operation.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Operation.java
index 9810fe5..bd41824 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Operation.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Operation.java
@@ -15,7 +15,9 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
+
+import org.jetbrains.annotations.Nullable;
/**
* Defines operation for meta storage conditional update (invoke).
@@ -37,37 +39,33 @@ public final class Operation {
/**
* Represents operation of type <i>remove</i>.
*/
- public static final class RemoveOp implements InnerOp {
- /** Key. */
- private final Key key;
-
+ public static final class RemoveOp extends AbstractOp {
/**
- * Creates a new remove operation for the given {@code key}.
+ * Default no-op constructor.
*
- * @param key Key.
+ * @param key Identifies an entry which operation will be applied to.
*/
- RemoveOp(Key key) {
- this.key = key;
+ RemoveOp(byte[] key) {
+ super(key);
}
}
/**
* Represents operation of type <i>put</i>.
*/
- public static final class PutOp implements InnerOp {
- /** Key. */
- private final Key key;
-
+ public static final class PutOp extends AbstractOp {
/** Value. */
private final byte[] val;
/**
* Constructs operation of type <i>put</i>.
*
+ * @param key Identifies an entry which operation will be applied to.
* @param val The value to which the entry should be updated.
*/
- PutOp(Key key, byte[] val) {
- this.key = key;
+ PutOp(byte[] key, byte[] val) {
+ super(key);
+
this.val = val;
}
}
@@ -75,12 +73,12 @@ public final class Operation {
/**
* Represents operation of type <i>no-op</i>.
*/
- public static final class NoOp implements InnerOp {
+ public static final class NoOp extends AbstractOp {
/**
* Default no-op constructor.
*/
NoOp() {
- // No-op.
+ super(null);
}
}
@@ -88,6 +86,19 @@ public final class Operation {
* Defines operation interface.
*/
private interface InnerOp {
- // Marker interface.
+ @Nullable byte[] key();
+ }
+
+ private static class AbstractOp implements InnerOp {
+ @Nullable private final byte[] key;
+
+ public AbstractOp(@Nullable byte[] key) {
+ this.key = key;
+ }
+
+ @Nullable
+ @Override public byte[] key() {
+ return key;
+ }
}
}
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/OperationTimeoutException.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/OperationTimeoutException.java
similarity index 97%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/OperationTimeoutException.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/OperationTimeoutException.java
index 4edcf02..3428f39 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/OperationTimeoutException.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/OperationTimeoutException.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
/**
* Thrown when an operation is not executed within a specified time period. Usually in such cases the operation
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Operations.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Operations.java
similarity index 79%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Operations.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Operations.java
index 31c7449..87f806f 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/Operations.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/Operations.java
@@ -15,7 +15,9 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
+
+import org.apache.ignite.lang.ByteArray;
/**
* This class contains fabric methods which produce operations needed for conditional multi update functionality
@@ -30,20 +32,22 @@ public final class Operations {
/**
* Creates operation of type <i>remove</i>. This type of operation removes entry.
*
+ * @param key Identifies an entry which operation will be applied to.
* @return Operation of type <i>remove</i>.
*/
- public static Operation remove(Key key) {
- return new Operation(new Operation.RemoveOp(key));
+ public static Operation remove(ByteArray key) {
+ return new Operation(new Operation.RemoveOp(key.bytes()));
}
/**
* Creates operation of type <i>put</i>. This type of operation inserts or updates value of entry.
*
+ * @param key Identifies an entry which operation will be applied to.
* @param value Value.
* @return Operation of type <i>put</i>.
*/
- public static Operation put(Key key, byte[] value) {
- return new Operation(new Operation.PutOp(key, value));
+ public static Operation put(ByteArray key, byte[] value) {
+ return new Operation(new Operation.PutOp(key.bytes(), value));
}
/**
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/WatchEvent.java
similarity index 82%
copy from modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java
copy to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/WatchEvent.java
index 561f203..1ddac71 100644
--- a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/WatchEvent.java
@@ -15,7 +15,9 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.metastorage.server;
+package org.apache.ignite.metastorage.client;
+
+import org.jetbrains.annotations.NotNull;
import java.util.Collection;
import java.util.List;
@@ -23,7 +25,7 @@ import java.util.List;
public class WatchEvent {
private final List<EntryEvent> entryEvts;
- private final boolean batch;
+ private final boolean single;
/**
* Constructs an watch event with given entry events collection.
@@ -33,12 +35,16 @@ public class WatchEvent {
public WatchEvent(List<EntryEvent> entryEvts) {
assert entryEvts != null && !entryEvts.isEmpty();
- this.batch = entryEvts.size() > 1;
+ this.single = entryEvts.size() == 1;
this.entryEvts = entryEvts;
}
- public boolean batch() {
- return batch;
+ public WatchEvent(@NotNull EntryEvent entryEvt) {
+ this(List.of(entryEvt));
+ }
+
+ public boolean single() {
+ return single;
}
public Collection<EntryEvent> entryEvents() {
@@ -46,9 +52,6 @@ public class WatchEvent {
}
public EntryEvent entryEvent() {
- if (batch)
- throw new IllegalStateException("Watch event represents a batch of events.");
-
return entryEvts.get(0);
}
}
diff --git a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/WatchListener.java b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/WatchListener.java
similarity index 87%
rename from modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/WatchListener.java
rename to modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/WatchListener.java
index 1ed196e..37c6a6f 100644
--- a/modules/metastorage-common/src/main/java/org/apache/ignite/metastorage/common/WatchListener.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/metastorage/client/WatchListener.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.metastorage.common;
+package org.apache.ignite.metastorage.client;
import org.jetbrains.annotations.NotNull;
@@ -26,11 +26,11 @@ public interface WatchListener {
/**
* The method will be called on each meta storage update.
*
- * @param events A single event or a batch. The batch always contains updates for specific revision.
+ * @param evt A single event or a batch. The batch always contains updates for specific revision.
* @return {@code True} if listener must continue event handling. If returns {@code false} then the listener and
* corresponding watch will be unregistered.
*/
- boolean onUpdate(@NotNull Iterable<WatchEvent> events);
+ boolean onUpdate(@NotNull WatchEvent evt);
/**
* The method will be called in case of an error occurred. The listener and corresponding watch will be
diff --git a/modules/metastorage-server/pom.xml b/modules/metastorage-server/pom.xml
index d73d080..5f0f453 100644
--- a/modules/metastorage-server/pom.xml
+++ b/modules/metastorage-server/pom.xml
@@ -29,20 +29,18 @@
<relativePath>../../parent/pom.xml</relativePath>
</parent>
- <artifactId>metastorage-server</artifactId>
+ <artifactId>ignite-metastorage-server</artifactId>
<version>3.0.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.ignite</groupId>
<artifactId>ignite-core</artifactId>
- <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.ignite</groupId>
- <artifactId>metastorage-common</artifactId>
- <version>${project.version}</version>
+ <artifactId>ignite-metastorage-common</artifactId>
</dependency>
<dependency>
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractCondition.java
similarity index 55%
copy from modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
copy to modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractCondition.java
index 554a3a7..df4acbe 100644
--- a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractCondition.java
@@ -17,42 +17,28 @@
package org.apache.ignite.internal.metastorage.server;
+import org.jetbrains.annotations.NotNull;
+
/**
- * Represent an change event for particular key and entry.
+ * An abstract condition which could be applied to an entry identified by the key.
*/
-public class EntryEvent {
- /** Old (previous) entry. */
- private final Entry oldEntry;
-
- /** New (current) entry. */
- private final Entry entry;
-
- /**
- * Constructs event with given old and new entries.
- *
- * @param oldEntry Old entry.
- * @param curEntry New entry.
- */
- EntryEvent(Entry oldEntry, Entry curEntry) {
- this.oldEntry = oldEntry;
- this.entry = curEntry;
- }
+public abstract class AbstractCondition implements Condition {
+ /** Entry key. */
+ @NotNull
+ private final byte[] key;
/**
- * Returns old entry.
+ * Constructs a condition with the given entry key.
*
- * @return Old entry.
+ * @param key Key identifies an entry which the condition will applied to.
*/
- public Entry oldEntry() {
- return oldEntry;
+ public AbstractCondition(@NotNull byte[] key) {
+ this.key = key;
}
- /**
- * Rreturns new entry.
- *
- * @return New entry.
- */
- public Entry entry() {
- return entry;
+ /** {@inheritDoc} */
+ @NotNull
+ @Override public byte[] key() {
+ return key;
}
}
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/Condition.java
similarity index 56%
copy from modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
copy to modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/Condition.java
index 554a3a7..e20459a 100644
--- a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/Condition.java
@@ -17,42 +17,28 @@
package org.apache.ignite.internal.metastorage.server;
-/**
- * Represent an change event for particular key and entry.
- */
-public class EntryEvent {
- /** Old (previous) entry. */
- private final Entry oldEntry;
+import org.jetbrains.annotations.NotNull;
- /** New (current) entry. */
- private final Entry entry;
-
- /**
- * Constructs event with given old and new entries.
- *
- * @param oldEntry Old entry.
- * @param curEntry New entry.
- */
- EntryEvent(Entry oldEntry, Entry curEntry) {
- this.oldEntry = oldEntry;
- this.entry = curEntry;
- }
+import java.util.Collection;
+/**
+ * Defines interface for condition which could be applied to an entry.
+ *
+ * @see KeyValueStorage#invoke(Condition, Collection, Collection)
+ */
+public interface Condition {
/**
- * Returns old entry.
+ * Returns the key which identifies an entry which condition will applied to.
*
- * @return Old entry.
+ * @return The key which identifies an entry which condition will applied to.
*/
- public Entry oldEntry() {
- return oldEntry;
- }
+ @NotNull byte[] key();
/**
- * Rreturns new entry.
+ * Tests the given entry on condition.
*
- * @return New entry.
+ * @param e Entry which will be tested on the condition. Can't be {@code null}.
+ * @return {@code True} if the given entry satisfies to the condition, otherwise - {@code false}.
*/
- public Entry entry() {
- return entry;
- }
+ boolean test(@NotNull Entry e);
}
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
index 554a3a7..c1e4f56 100644
--- a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/EntryEvent.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.metastorage.server;
/**
- * Represent an change event for particular key and entry.
+ * Represent an update event for particular key and entry.
*/
public class EntryEvent {
/** Old (previous) entry. */
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/ExistenceCondition.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/ExistenceCondition.java
new file mode 100644
index 0000000..506e23b
--- /dev/null
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/ExistenceCondition.java
@@ -0,0 +1,74 @@
+/*
+ * 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 org.apache.ignite.internal.metastorage.server;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Condition tests an entry on existence in meta storage.
+ * Entry exists if it is not empty and not tombstone.
+ */
+public class ExistenceCondition extends AbstractCondition {
+ /** Condition type. */
+ @NotNull
+ private final Type type;
+
+ /**
+ * Constructs existence condition with the given type and for entry identified by the given key.
+ *
+ * @param type Condition type. Can't be {@code null}.
+ * @param key Key of entry which condition will be applied to. Can't be {@code null}.
+ */
+ public ExistenceCondition(@NotNull Type type, @NotNull byte[] key) {
+ super(key);
+
+ this.type = type;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean test(@NotNull Entry e) {
+ boolean res = !(e.empty() || e.tombstone());
+
+ return type.test(res);
+ }
+
+ /** Defines existence condition types. */
+ public enum Type {
+ /** Equality condition type. */
+ EXISTS {
+ @Override public boolean test(boolean res) {
+ return res;
+ }
+ },
+
+ /** Inequality condition type. */
+ NOT_EXISTS {
+ @Override public boolean test(boolean res) {
+ return !res;
+ }
+ };
+
+ /**
+ * Interprets comparison result.
+ *
+ * @param res The result of comparison.
+ * @return The interpretation of the comparison result.
+ */
+ public abstract boolean test(boolean res);
+ }
+}
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
index 5d6da44..f46610c 100644
--- a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java
@@ -19,7 +19,7 @@ package org.apache.ignite.internal.metastorage.server;
import java.util.Collection;
import java.util.List;
-import org.apache.ignite.metastorage.common.Cursor;
+import org.apache.ignite.internal.util.Cursor;
import org.jetbrains.annotations.NotNull;
public interface KeyValueStorage {
@@ -59,6 +59,8 @@ public interface KeyValueStorage {
@NotNull
Collection<Entry> getAndRemoveAll(List<byte[]> keys);
+ boolean invoke(Condition condition, Collection<Operation> success, Collection<Operation> failure);
+
Cursor<Entry> range(byte[] keyFrom, byte[] keyTo);
Cursor<Entry> range(byte[] keyFrom, byte[] keyTo, long revUpperBound);
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/Operation.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/Operation.java
new file mode 100644
index 0000000..aaea75b
--- /dev/null
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/Operation.java
@@ -0,0 +1,111 @@
+/*
+ * 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 org.apache.ignite.internal.metastorage.server;
+
+import java.util.Objects;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Defines operation which will be applied to an entry identified by the key.
+ * <p>
+ * Invariants:
+ * <ul>
+ * <li>Any operation identifies a target entry by not null {@code key} except of {@link Type#NO_OP}.</li>
+ * <li>Only {@link Type#PUT} operation contains value which will be written to meta storage.</li>
+ * </ul>
+ */
+final class Operation {
+ /**
+ * Key identifies an entry which operation will be applied to. Key is {@code null} for {@link Type#NO_OP} operation.
+ */
+ @Nullable
+ private final byte[] key;
+
+ /**
+ * Value which will be associated with the {@link #key}. Value is not {@code null} only for {@link Type#PUT}
+ * operation.
+ */
+ @Nullable
+ private final byte[] val;
+
+ /**
+ * Operation type.
+ * @see Type
+ */
+ @NotNull
+ private final Type type;
+
+ /**
+ * Constructs operation which will be applied to an entry identified by the given key.
+ *
+ * @param type Operation type. Can't be {@code null}.
+ * @param key Key identifies an entry which operation will be applied to.
+ * @param val Value will be associated with an entry identified by the {@code key}.
+ */
+ Operation(@NotNull Type type, @Nullable byte[] key, @Nullable byte[] val) {
+ assert (type == Type.NO_OP && key == null && val == null)
+ || (type == Type.PUT && key != null && val != null)
+ || (type == Type.REMOVE && key != null && val == null)
+ : "Invalid operation parameters: [type=" + type + ", key=" + Objects.toString(key,"null") +
+ ", val=" + Objects.toString(key,"null") + ']';
+
+ this.key = key;
+ this.val = val;
+ this.type = type;
+ }
+
+ /**
+ * Returns a key which identifies an entry which operation will be applied to.
+ *
+ * @return A key which identifies an entry which operation will be applied to.
+ */
+ @Nullable byte[] key() {
+ return key;
+ }
+
+ /**
+ * Returns a value which will be associated with an entry identified by the {@code key}.
+ *
+ * @return A value which will be associated with an entry identified by the {@code key}.
+ */
+ @Nullable byte[] value() {
+ return val;
+ }
+
+ /**
+ * Returns an operation type.
+ *
+ * @return An operation type.
+ */
+ @NotNull Type type() {
+ return type;
+ }
+
+ /** Defines operation types. */
+ enum Type {
+ /** Put operation. */
+ PUT,
+
+ /** Remove operation. */
+ REMOVE,
+
+ /** No-op operation. */
+ NO_OP
+ }
+}
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/RevisionCondition.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/RevisionCondition.java
new file mode 100644
index 0000000..94e8b71
--- /dev/null
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/RevisionCondition.java
@@ -0,0 +1,110 @@
+/*
+ * 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 org.apache.ignite.internal.metastorage.server;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Defines condition on entry revision.
+ */
+public class RevisionCondition extends AbstractCondition {
+ /** Condition type. */
+ @NotNull
+ private final Type type;
+
+ /** Revision which will be tested against an entry revision. Must be positive. */
+ private final long rev;
+
+ /**
+ * Constructs revision condition with the given type, the key and revision.
+ *
+ * @param type Condition type. Can't be {@code null}.
+ * @param key Key identifies an entry which condition will be applied to. Can't be {@code null}.
+ * @param rev Revision which will be tested against an entry revision. Must be positive.
+ */
+ public RevisionCondition(@NotNull Type type, @NotNull byte[] key, long rev) {
+ super(key);
+
+ assert rev > 0 : "Revision must be positive.";
+
+ this.type = type;
+ this.rev = rev;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean test(@NotNull Entry e) {
+ int res = Long.compare(e.revision(), rev);
+
+ return type.test(res);
+ }
+
+ /**
+ * Defines possible condition types which can be applied to the revision.
+ */
+ public enum Type {
+ /** Equality condition type. */
+ EQUAL {
+ @Override public boolean test(long res) {
+ return res == 0;
+ }
+ },
+
+ /** Inequality condition type. */
+ NOT_EQUAL {
+ @Override public boolean test(long res) {
+ return res != 0;
+ }
+ },
+
+ /** Greater than condition type. */
+ GREATER {
+ @Override public boolean test(long res) {
+ return res > 0;
+ }
+ },
+
+ /** Less than condition type. */
+ LESS {
+ @Override public boolean test(long res) {
+ return res < 0;
+ }
+ },
+
+ /** Less than or equal to condition type. */
+ LESS_OR_EQUAL {
+ @Override public boolean test(long res) {
+ return res <= 0;
+ }
+ },
+
+ /** Greater than or equal to condition type. */
+ GREATER_OR_EQUAL {
+ @Override public boolean test(long res) {
+ return res >= 0;
+ }
+ };
+
+ /**
+ * Interprets comparison result.
+ *
+ * @param res The result of comparison.
+ * @return The interpretation of the comparison result.
+ */
+ public abstract boolean test(long res);
+ }
+}
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
index b37c96a..1f3f0e4 100644
--- a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorage.java
@@ -29,7 +29,7 @@ import java.util.NoSuchElementException;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.function.Predicate;
-import org.apache.ignite.metastorage.common.Cursor;
+import org.apache.ignite.internal.util.Cursor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.TestOnly;
@@ -63,14 +63,22 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
@Override public void put(byte[] key, byte[] value) {
synchronized (mux) {
- doPut(key, value);
+ long curRev = rev + 1;
+
+ doPut(key, value, curRev);
+
+ rev = curRev;
}
}
@NotNull
@Override public Entry getAndPut(byte[] key, byte[] bytes) {
synchronized (mux) {
- long lastRev = doPut(key, bytes);
+ long curRev = rev + 1;
+
+ long lastRev = doPut(key, bytes, curRev);
+
+ rev = curRev;
// Return previous value.
return doGetValue(key, lastRev);
@@ -129,15 +137,24 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
@Override
public void remove(byte[] key) {
synchronized (mux) {
- Entry e = doGet(key, LATEST_REV, false);
-
- if (e.empty() || e.tombstone())
- return;
+ long curRev = rev + 1;
- doPut(key, TOMBSTONE);
+ if (doRemove(key, curRev))
+ rev = curRev;
}
}
+ private boolean doRemove(byte[] key, long curRev) {
+ Entry e = doGet(key, LATEST_REV, false);
+
+ if (e.empty() || e.tombstone())
+ return false;
+
+ doPut(key, TOMBSTONE, curRev);
+
+ return true;
+ }
+
@NotNull
@Override public Entry getAndRemove(byte[] key) {
synchronized (mux) {
@@ -204,6 +221,47 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
return res;
}
+ @Override public boolean invoke(Condition condition, Collection<Operation> success, Collection<Operation> failure) {
+ synchronized (mux) {
+ Entry e = get(condition.key());
+
+ boolean branch = condition.test(e);
+
+ Collection<Operation> ops = branch ? success : failure;
+
+ long curRev = rev + 1;
+
+ boolean modified = false;
+
+ for (Operation op : ops) {
+ switch (op.type()) {
+ case PUT:
+ doPut(op.key(), op.value(), curRev);
+
+ modified = true;
+
+ break;
+
+ case REMOVE:
+ modified |= doRemove(op.key(), curRev);
+
+ break;
+
+ case NO_OP:
+ break;
+
+ default:
+ throw new IllegalArgumentException("Unknown operation type: " + op.type());
+ }
+ }
+
+ if (modified)
+ rev = curRev;
+
+ return branch;
+ }
+ }
+
@Override public Cursor<Entry> range(byte[] keyFrom, byte[] keyTo) {
return new RangeCursor(keyFrom, keyTo, rev);
}
@@ -365,9 +423,7 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
return new Entry(key, lastVal.bytes() , lastRev, lastVal.updateCounter());
}
- private long doPut(byte[] key, byte[] bytes) {
- long curRev = ++rev;
-
+ private long doPut(byte[] key, byte[] bytes, long curRev) {
long curUpdCntr = ++updCntr;
// Update keysIdx.
@@ -378,13 +434,25 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
revs.add(curRev);
// Update revsIdx.
- NavigableMap<byte[], Value> entries = new TreeMap<>(CMP);
+ //NavigableMap<byte[], Value> entries = new TreeMap<>(CMP);
Value val = new Value(bytes, curUpdCntr);
- entries.put(key, val);
+ //entries.put(key, val);
+
+ //revsIdx.put(curRev, entries);
- revsIdx.put(curRev, entries);
+ revsIdx.compute(
+ curRev,
+ (rev, entries) -> {
+ if (entries == null)
+ entries = new TreeMap<>(CMP);
+
+ entries.put(key, val);
+
+ return entries;
+ }
+ );
return lastRev;
}
@@ -437,6 +505,7 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
private final byte[] keyFrom;
private final byte[] keyTo;
private final long rev;
+ private final Iterator<Entry> it;
private Entry nextRetEntry;
private byte[] lastRetKey;
private boolean finished;
@@ -445,14 +514,32 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
this.keyFrom = keyFrom;
this.keyTo = keyTo;
this.rev = rev;
+ this.it = createIterator();
}
+ /** {@inheritDoc} */
+ @Override public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Entry next() {
+ return it.next();
+ }
+
+ /** {@inheritDoc} */
@Override public void close() throws Exception {
// TODO: implement.
}
+ /** {@inheritDoc} */
@NotNull
@Override public Iterator<Entry> iterator() {
+ return it;
+ }
+
+ @NotNull
+ Iterator<Entry> createIterator() {
return new Iterator<>() {
@Override public boolean hasNext() {
synchronized (mux) {
@@ -530,20 +617,39 @@ public class SimpleInMemoryKeyValueStorage implements KeyValueStorage {
private class WatchCursor implements Cursor<WatchEvent> {
private final Predicate<byte[]> p;
+ private final Iterator<WatchEvent> it;
private long lastRetRev;
private long nextRetRev = -1;
WatchCursor(long rev, Predicate<byte[]> p) {
this.p = p;
this.lastRetRev = rev - 1;
+ this.it = createIterator();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ /** {@inheritDoc} */
+ @Override public WatchEvent next() {
+ return it.next();
}
+ /** {@inheritDoc} */
@Override public void close() throws Exception {
- // TODO: implement
+ // TODO: implement.
}
+ /** {@inheritDoc} */
@NotNull
@Override public Iterator<WatchEvent> iterator() {
+ return it;
+ }
+
+ @NotNull
+ Iterator<WatchEvent> createIterator() {
return new Iterator<>() {
@Override public boolean hasNext() {
synchronized (mux) {
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/ValueCondition.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/ValueCondition.java
new file mode 100644
index 0000000..444bfdb
--- /dev/null
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/ValueCondition.java
@@ -0,0 +1,82 @@
+/*
+ * 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 org.apache.ignite.internal.metastorage.server;
+
+import java.util.Arrays;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Defines condition on entry value.
+ */
+public class ValueCondition extends AbstractCondition {
+ /** Condition type. */
+ @NotNull
+ private final Type type;
+
+ /** Value which will be tested against an entry value. */
+ @NotNull
+ private final byte[] val;
+
+ /**
+ * Constructs value condition with the given type, key and value.
+ *
+ * @param type Condition type. Can't be {@code null}.
+ * @param key Key identifies an entry which condition will be applied to. Can't be {@code null}.
+ * @param val Value which will be tested against an entry value. Can't be {@code null}.
+ */
+ public ValueCondition(@NotNull Type type, @NotNull byte[] key, @NotNull byte[] val) {
+ super(key);
+
+ this.type = type;
+ this.val = val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean test(@NotNull Entry e) {
+ int res = Arrays.compare(e.value(), val);
+
+ return type.test(res);
+ }
+
+ /**
+ * Defines possible condition types which can be applied to the value.
+ */
+ enum Type {
+ /** Equality condition type. */
+ EQUAL {
+ @Override public boolean test(long res) {
+ return res == 0;
+ }
+ },
+
+ /** Inequality condition type. */
+ NOT_EQUAL {
+ @Override public boolean test(long res) {
+ return res != 0;
+ }
+ };
+
+ /**
+ * Interprets comparison result.
+ *
+ * @param res The result of comparison.
+ * @return The interpretation of the comparison result.
+ */
+ public abstract boolean test(long res);
+ }
+}
diff --git a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java
index 561f203..4f2f305 100644
--- a/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java
+++ b/modules/metastorage-server/src/main/java/org/apache/ignite/internal/metastorage/server/WatchEvent.java
@@ -23,7 +23,7 @@ import java.util.List;
public class WatchEvent {
private final List<EntryEvent> entryEvts;
- private final boolean batch;
+ private final boolean single;
/**
* Constructs an watch event with given entry events collection.
@@ -33,12 +33,12 @@ public class WatchEvent {
public WatchEvent(List<EntryEvent> entryEvts) {
assert entryEvts != null && !entryEvts.isEmpty();
- this.batch = entryEvts.size() > 1;
+ this.single = entryEvts.size() == 1;
this.entryEvts = entryEvts;
}
- public boolean batch() {
- return batch;
+ public boolean single() {
+ return single;
}
public Collection<EntryEvent> entryEvents() {
@@ -46,9 +46,6 @@ public class WatchEvent {
}
public EntryEvent entryEvent() {
- if (batch)
- throw new IllegalStateException("Watch event represents a batch of events.");
-
return entryEvts.get(0);
}
}
diff --git a/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/ExistenceConditionTest.java b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/ExistenceConditionTest.java
new file mode 100644
index 0000000..7d58743
--- /dev/null
+++ b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/ExistenceConditionTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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 org.apache.ignite.internal.metastorage.server;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.ignite.internal.metastorage.server.ExistenceCondition.Type.EXISTS;
+import static org.apache.ignite.internal.metastorage.server.ExistenceCondition.Type.NOT_EXISTS;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for entry existence condition.
+ *
+ * @see ExistenceCondition
+ */
+public class ExistenceConditionTest {
+ /** Entry key. */
+ private static final byte[] KEY = new byte[] {1};
+
+ /** Entry value. */
+ private static final byte[] VAL = new byte[] {1};
+
+ /** Regular entry. */
+ private static final Entry ENTRY = new Entry(KEY, VAL, 1, 1);
+
+ /** Empty entry. */
+ private static final Entry EMPTY = Entry.empty(KEY);
+
+ /** Tombstone entry. */
+ private static final Entry TOMBSTONE = Entry.tombstone(KEY, 1, 1);
+
+ /**
+ * Tests {@link ExistenceCondition.Type#EXISTS} condition for regular, empty and tombstone entries.
+ */
+ @Test
+ public void exists() {
+ Condition cond = new ExistenceCondition(EXISTS, KEY);
+
+ assertTrue(cond.test(ENTRY));
+ assertFalse(cond.test(EMPTY));
+ assertFalse(cond.test(TOMBSTONE));
+ }
+
+ /**
+ * Tests {@link ExistenceCondition.Type#NOT_EXISTS} condition for regular, empty and tombstone entries.
+ */
+ @Test
+ public void notExists() {
+ Condition cond = new ExistenceCondition(NOT_EXISTS, KEY);
+
+ assertFalse(cond.test(ENTRY));
+ assertTrue(cond.test(EMPTY));
+ assertTrue(cond.test(TOMBSTONE));
+ }
+}
diff --git a/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/RevisionConditionTest.java b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/RevisionConditionTest.java
new file mode 100644
index 0000000..2c461aa
--- /dev/null
+++ b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/RevisionConditionTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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 org.apache.ignite.internal.metastorage.server;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Type.EQUAL;
+import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Type.GREATER;
+import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Type.GREATER_OR_EQUAL;
+import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Type.LESS;
+import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Type.LESS_OR_EQUAL;
+import static org.apache.ignite.internal.metastorage.server.RevisionCondition.Type.NOT_EQUAL;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for entry revision condition.
+ *
+ * @see RevisionCondition
+ */
+public class RevisionConditionTest {
+ /** Entry key. */
+ private static final byte[] KEY = new byte[] {1};
+
+ /** Entry value. */
+ private static final byte[] VAL = new byte[] {2};
+
+ /**
+ * Tests revisions equality.
+ */
+ @Test
+ public void eq() {
+ Condition cond = new RevisionCondition(EQUAL, KEY, 1);
+
+ // 1 == 1.
+ assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ }
+
+ /**
+ * Tests revisions inequality.
+ */
+ @Test
+ public void ne() {
+ Condition cond = new RevisionCondition(NOT_EQUAL, KEY, 1);
+
+ // 2 != 1.
+ assertTrue(cond.test(new Entry(KEY, VAL, 2, 1)));
+ }
+
+ /**
+ * Tests that revision is greater than another one.
+ */
+ @Test
+ public void gt() {
+ Condition cond = new RevisionCondition(GREATER, KEY, 1);
+
+ // 2 > 1.
+ assertTrue(cond.test(new Entry(KEY, VAL, 2, 1)));
+ }
+
+ /**
+ * Tests that revision is greater than or equal to another one .
+ */
+ @Test
+ public void ge() {
+ Condition cond = new RevisionCondition(GREATER_OR_EQUAL, KEY, 1);
+
+ // 2 >= 1 (2 > 1).
+ assertTrue(cond.test(new Entry(KEY, VAL, 2, 1)));
+
+ // 1 >= 1 (1 == 1).
+ assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ }
+
+ /**
+ * Tests that revision is less than another one.
+ */
+ @Test
+ public void lt() {
+ Condition cond = new RevisionCondition(LESS, KEY, 2);
+
+ // 1 < 2
+ assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ }
+
+ /**
+ * Tests that revision is less than or equal to another one .
+ */
+ @Test
+ public void le() {
+ Condition cond = new RevisionCondition(LESS_OR_EQUAL, KEY, 2);
+
+ // 1 <= 2 (1 < 2)
+ assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+
+ // 1 <= 1 (1 == 1).
+ assertTrue(cond.test(new Entry(KEY, VAL, 1, 1)));
+ }
+}
diff --git a/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorageTest.java b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorageTest.java
index 27df790..8e138fd 100644
--- a/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorageTest.java
+++ b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/SimpleInMemoryKeyValueStorageTest.java
@@ -23,13 +23,19 @@ import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.stream.Collectors;
-import org.apache.ignite.metastorage.common.Cursor;
-import org.apache.ignite.metastorage.common.Key;
+import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.lang.ByteArray;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static java.util.function.Function.identity;
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
class SimpleInMemoryKeyValueStorageTest {
private KeyValueStorage storage;
@@ -108,10 +114,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- Map<Key, Entry> map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ Map<ByteArray, Entry> map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- Entry e1 = map.get(new Key(key1));
+ Entry e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(1, e1.revision());
@@ -121,7 +127,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val1, e1.value());
// Test rewritten value.
- Entry e2 = map.get(new Key(key2));
+ Entry e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(3, e2.revision());
@@ -131,7 +137,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val2_2, e2.value());
// Test removed value.
- Entry e3 = map.get(new Key(key3));
+ Entry e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(5, e3.revision());
@@ -140,7 +146,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e3.empty());
// Test empty value.
- Entry e4 = map.get(new Key(key4));
+ Entry e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertFalse(e4.tombstone());
@@ -183,10 +189,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- Map<Key, Entry> map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ Map<ByteArray, Entry> map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- Entry e1 = map.get(new Key(key1));
+ Entry e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(1, e1.revision());
@@ -196,7 +202,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val1, e1.value());
// Test while not rewritten value.
- Entry e2 = map.get(new Key(key2));
+ Entry e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(2, e2.revision());
@@ -206,12 +212,12 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val2_1, e2.value());
// Values with larger revision don't exist yet.
- Entry e3 = map.get(new Key(key3));
+ Entry e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertTrue(e3.empty());
- Entry e4 = map.get(new Key(key4));
+ Entry e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertTrue(e4.empty());
@@ -221,10 +227,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- e1 = map.get(new Key(key1));
+ e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(1, e1.revision());
@@ -234,7 +240,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val1, e1.value());
// Test rewritten value.
- e2 = map.get(new Key(key2));
+ e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(3, e2.revision());
@@ -244,7 +250,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val2_2, e2.value());
// Test not removed value.
- e3 = map.get(new Key(key3));
+ e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(4, e3.revision());
@@ -254,7 +260,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val3, e3.value());
// Value with larger revision doesn't exist yet.
- e4 = map.get(new Key(key4));
+ e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertTrue(e4.empty());
@@ -325,10 +331,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- Map<Key, Entry> map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ Map<ByteArray, Entry> map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- Entry e1 = map.get(new Key(key1));
+ Entry e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(4, e1.revision());
@@ -338,7 +344,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val1, e1.value());
// Test rewritten value.
- Entry e2 = map.get(new Key(key2));
+ Entry e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(4, e2.revision());
@@ -348,7 +354,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val2_2, e2.value());
// Test removed value.
- Entry e3 = map.get(new Key(key3));
+ Entry e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(4, e3.revision());
@@ -357,7 +363,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e3.empty());
// Test empty value.
- Entry e4 = map.get(new Key(key4));
+ Entry e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertFalse(e4.tombstone());
@@ -399,10 +405,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(3, entries.size());
- Map<Key, Entry> map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ Map<ByteArray, Entry> map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- Entry e1 = map.get(new Key(key1));
+ Entry e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(0, e1.revision());
@@ -411,7 +417,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertTrue(e1.empty());
// Test rewritten value.
- Entry e2 = map.get(new Key(key2));
+ Entry e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(1, e2.revision());
@@ -421,7 +427,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val2_1, e2.value());
// Test removed value.
- Entry e3 = map.get(new Key(key3));
+ Entry e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(3, e3.revision());
@@ -434,10 +440,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- e1 = map.get(new Key(key1));
+ e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(4, e1.revision());
@@ -447,7 +453,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val1, e1.value());
// Test rewritten value.
- e2 = map.get(new Key(key2));
+ e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(4, e2.revision());
@@ -457,7 +463,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val2_2, e2.value());
// Test removed value.
- e3 = map.get(new Key(key3));
+ e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(4, e3.revision());
@@ -466,7 +472,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e3.empty());
// Test empty value.
- Entry e4 = map.get(new Key(key4));
+ Entry e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertFalse(e4.tombstone());
@@ -618,10 +624,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- Map<Key, Entry> map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ Map<ByteArray, Entry> map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- Entry e1 = map.get(new Key(key1));
+ Entry e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(6, e1.revision());
@@ -630,7 +636,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e1.empty());
// Test rewritten value.
- Entry e2 = map.get(new Key(key2));
+ Entry e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(6, e2.revision());
@@ -639,7 +645,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e2.empty());
// Test removed value.
- Entry e3 = map.get(new Key(key3));
+ Entry e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(5, e3.revision());
@@ -648,7 +654,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e3.empty());
// Test empty value.
- Entry e4 = map.get(new Key(key4));
+ Entry e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertFalse(e4.tombstone());
@@ -693,10 +699,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- Map<Key, Entry> map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ Map<ByteArray, Entry> map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- Entry e1 = map.get(new Key(key1));
+ Entry e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(1, e1.revision());
@@ -706,7 +712,7 @@ class SimpleInMemoryKeyValueStorageTest {
// Test rewritten value.
- Entry e2 = map.get(new Key(key2));
+ Entry e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(3, e2.revision());
@@ -716,7 +722,7 @@ class SimpleInMemoryKeyValueStorageTest {
// Test removed value.
- Entry e3 = map.get(new Key(key3));
+ Entry e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(5, e3.revision());
@@ -725,7 +731,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e3.empty());
// Test empty value.
- Entry e4 = map.get(new Key(key4));
+ Entry e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertFalse(e4.tombstone());
@@ -736,10 +742,10 @@ class SimpleInMemoryKeyValueStorageTest {
assertEquals(4, entries.size());
- map = entries.stream().collect(Collectors.toMap(e -> new Key(e.key()), identity()));
+ map = entries.stream().collect(Collectors.toMap(e -> new ByteArray(e.key()), identity()));
// Test regular put value.
- e1 = map.get(new Key(key1));
+ e1 = map.get(new ByteArray(key1));
assertNotNull(e1);
assertEquals(6, e1.revision());
@@ -748,7 +754,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e1.empty());
// Test rewritten value.
- e2 = map.get(new Key(key2));
+ e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
assertEquals(6, e2.revision());
@@ -757,7 +763,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e2.empty());
// Test removed value.
- e3 = map.get(new Key(key3));
+ e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
assertEquals(5, e3.revision());
@@ -766,7 +772,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertFalse(e3.empty());
// Test empty value.
- e4 = map.get(new Key(key4));
+ e4 = map.get(new ByteArray(key4));
assertNotNull(e4);
assertFalse(e4.tombstone());
@@ -939,6 +945,553 @@ class SimpleInMemoryKeyValueStorageTest {
}
@Test
+ public void invokeWithRevisionCondition_successBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new RevisionCondition(RevisionCondition.Type.EQUAL, key1, 1),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ ),
+ List.of(new Operation(Operation.Type.PUT, key3, val3))
+ );
+
+ // "Success" branch is applied.
+ assertTrue(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Failure" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeWithRevisionCondition_failureBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new RevisionCondition(RevisionCondition.Type.EQUAL, key1, 2),
+ List.of(new Operation(Operation.Type.PUT, key3, val3)),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ )
+ );
+
+ // "Failure" branch is applied.
+ assertFalse(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Success" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeWithExistsCondition_successBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new ExistenceCondition(ExistenceCondition.Type.EXISTS, key1),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ ),
+ List.of(new Operation(Operation.Type.PUT, key3, val3))
+ );
+
+ // "Success" branch is applied.
+ assertTrue(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Failure" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeWithExistsCondition_failureBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new ExistenceCondition(ExistenceCondition.Type.EXISTS, key3),
+ List.of(new Operation(Operation.Type.PUT, key3, val3)),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ )
+ );
+
+ // "Failure" branch is applied.
+ assertFalse(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Success" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeWithNotExistsCondition_successBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new ExistenceCondition(ExistenceCondition.Type.NOT_EXISTS, key2),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ ),
+ List.of(new Operation(Operation.Type.PUT, key3, val3))
+ );
+
+ // "Success" branch is applied.
+ assertTrue(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Failure" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeWithNotExistsCondition_failureBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new ExistenceCondition(ExistenceCondition.Type.NOT_EXISTS, key1),
+ List.of(new Operation(Operation.Type.PUT, key3, val3)),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ )
+ );
+
+ // "Failure" branch is applied.
+ assertFalse(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Success" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeWithValueCondition_successBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new ValueCondition(ValueCondition.Type.EQUAL, key1, val1_1),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ ),
+ List.of(new Operation(Operation.Type.PUT, key3, val3))
+ );
+
+ // "Success" branch is applied.
+ assertTrue(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Failure" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeWithValueCondition_failureBranch() {
+ byte[] key1 = k(1);
+ byte[] val1_1 = kv(1, 11);
+ byte[] val1_2 = kv(1, 12);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1_1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ boolean branch = storage.invoke(
+ new ValueCondition(ValueCondition.Type.EQUAL, key1, val1_2),
+ List.of(new Operation(Operation.Type.PUT, key3, val3)),
+ List.of(
+ new Operation(Operation.Type.PUT, key1, val1_2),
+ new Operation(Operation.Type.PUT, key2, val2)
+ )
+ );
+
+ // "Failure" branch is applied.
+ assertFalse(branch);
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e1 = storage.get(key1);
+
+ assertFalse(e1.empty());
+ assertFalse(e1.tombstone());
+ assertEquals(2, e1.revision());
+ assertEquals(2, e1.updateCounter());
+ assertArrayEquals(val1_2, e1.value());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(3, e2.updateCounter());
+ assertArrayEquals(val2, e2.value());
+
+ // "Success" branch isn't applied.
+ Entry e3 = storage.get(key3);
+
+ assertTrue(e3.empty());
+ }
+
+ @Test
+ public void invokeOperations() {
+ byte[] key1 = k(1);
+ byte[] val1 = kv(1, 1);
+
+ byte[] key2 = k(2);
+ byte[] val2 = kv(2, 2);
+
+ byte[] key3 = k(3);
+ byte[] val3 = kv(3, 3);
+
+ assertEquals(0, storage.revision());
+ assertEquals(0, storage.updateCounter());
+
+ storage.put(key1, val1);
+
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ // No-op.
+ boolean branch = storage.invoke(
+ new ValueCondition(ValueCondition.Type.EQUAL, key1, val1),
+ List.of(new Operation(Operation.Type.NO_OP, null, null)),
+ List.of(new Operation(Operation.Type.NO_OP, null, null))
+ );
+
+ assertTrue(branch);
+
+ // No updates.
+ assertEquals(1, storage.revision());
+ assertEquals(1, storage.updateCounter());
+
+ // Put.
+ branch = storage.invoke(
+ new ValueCondition(ValueCondition.Type.EQUAL, key1, val1),
+ List.of(
+ new Operation(Operation.Type.PUT, key2, val2),
+ new Operation(Operation.Type.PUT, key3, val3)
+ ),
+ List.of(new Operation(Operation.Type.NO_OP, null, null))
+ );
+
+ assertTrue(branch);
+
+ // +1 for revision, +2 for update counter.
+ assertEquals(2, storage.revision());
+ assertEquals(3, storage.updateCounter());
+
+ Entry e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertFalse(e2.tombstone());
+ assertEquals(2, e2.revision());
+ assertEquals(2, e2.updateCounter());
+ assertArrayEquals(key2, e2.key());
+ assertArrayEquals(val2, e2.value());
+
+ Entry e3 = storage.get(key3);
+
+ assertFalse(e3.empty());
+ assertFalse(e3.tombstone());
+ assertEquals(2, e3.revision());
+ assertEquals(3, e3.updateCounter());
+ assertArrayEquals(key3, e3.key());
+ assertArrayEquals(val3, e3.value());
+
+ // Remove.
+ branch = storage.invoke(
+ new ValueCondition(ValueCondition.Type.EQUAL, key1, val1),
+ List.of(
+ new Operation(Operation.Type.REMOVE, key2, null),
+ new Operation(Operation.Type.REMOVE, key3, null)
+ ),
+ List.of(new Operation(Operation.Type.NO_OP, null, null))
+ );
+
+ assertTrue(branch);
+
+ // +1 for revision, +2 for update counter.
+ assertEquals(3, storage.revision());
+ assertEquals(5, storage.updateCounter());
+
+ e2 = storage.get(key2);
+
+ assertFalse(e2.empty());
+ assertTrue(e2.tombstone());
+ assertEquals(3, e2.revision());
+ assertEquals(4, e2.updateCounter());
+ assertArrayEquals(key2, e2.key());
+
+ e3 = storage.get(key3);
+
+ assertFalse(e3.empty());
+ assertTrue(e3.tombstone());
+ assertEquals(3, e3.revision());
+ assertEquals(5, e3.updateCounter());
+ assertArrayEquals(key3, e3.key());
+ }
+
+ @Test
public void compact() {
assertEquals(0, storage.revision());
assertEquals(0, storage.updateCounter());
@@ -1155,15 +1708,15 @@ class SimpleInMemoryKeyValueStorageTest {
WatchEvent watchEvent = it.next();
- assertTrue(watchEvent.batch());
+ assertFalse(watchEvent.single());
- Map<Key, EntryEvent> map = watchEvent.entryEvents().stream()
- .collect(Collectors.toMap(evt -> new Key(evt.entry().key()), identity()));
+ Map<ByteArray, EntryEvent> map = watchEvent.entryEvents().stream()
+ .collect(Collectors.toMap(evt -> new ByteArray(evt.entry().key()), identity()));
assertEquals(2, map.size());
// First update under revision.
- EntryEvent e2 = map.get(new Key(key2));
+ EntryEvent e2 = map.get(new ByteArray(key2));
assertNotNull(e2);
@@ -1186,7 +1739,7 @@ class SimpleInMemoryKeyValueStorageTest {
assertArrayEquals(val2_2, newEntry2.value());
// Second update under revision.
- EntryEvent e3 = map.get(new Key(key3));
+ EntryEvent e3 = map.get(new ByteArray(key3));
assertNotNull(e3);
@@ -1213,7 +1766,7 @@ class SimpleInMemoryKeyValueStorageTest {
watchEvent = it.next();
- assertFalse(watchEvent.batch());
+ assertTrue(watchEvent.single());
EntryEvent e1 = watchEvent.entryEvent();
@@ -1269,7 +1822,7 @@ class SimpleInMemoryKeyValueStorageTest {
WatchEvent watchEvent = it.next();
- assertFalse(watchEvent.batch());
+ assertTrue(watchEvent.single());
EntryEvent e1 = watchEvent.entryEvent();
@@ -1299,7 +1852,7 @@ class SimpleInMemoryKeyValueStorageTest {
watchEvent = it.next();
- assertFalse(watchEvent.batch());
+ assertTrue(watchEvent.single());
e1 = watchEvent.entryEvent();
@@ -1328,7 +1881,6 @@ class SimpleInMemoryKeyValueStorageTest {
public void watchCursorForKeys() {
byte[] key1 = k(1);
byte[] val1_1 = kv(1, 11);
- byte[] val1_2 = kv(1, 12);
byte[] key2 = k(2);
byte[] val2_1 = kv(2, 21);
@@ -1358,7 +1910,7 @@ class SimpleInMemoryKeyValueStorageTest {
WatchEvent watchEvent = it.next();
- assertTrue(watchEvent.batch());
+ assertFalse(watchEvent.single());
assertFalse(it.hasNext());
@@ -1368,7 +1920,7 @@ class SimpleInMemoryKeyValueStorageTest {
watchEvent = it.next();
- assertFalse(watchEvent.batch());
+ assertTrue(watchEvent.single());
assertFalse(it.hasNext());
diff --git a/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/ValueConditionTest.java b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/ValueConditionTest.java
new file mode 100644
index 0000000..654c67e
--- /dev/null
+++ b/modules/metastorage-server/src/test/java/org/apache/ignite/internal/metastorage/server/ValueConditionTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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 org.apache.ignite.internal.metastorage.server;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.ignite.internal.metastorage.server.ValueCondition.Type.EQUAL;
+import static org.apache.ignite.internal.metastorage.server.ValueCondition.Type.NOT_EQUAL;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for entry value condition.
+ *
+ * @see ValueCondition
+ */
+public class ValueConditionTest {
+ /** Entry key. */
+ private static final byte[] KEY = new byte[] {1};
+
+ /** Entry value. */
+ private static final byte[] VAL_1 = new byte[] {11};
+
+ /** Other entry value. */
+ private static final byte[] VAL_2 = new byte[] {22};
+
+ /**
+ * Tests values equality.
+ */
+ @Test
+ public void eq() {
+ Condition cond = new ValueCondition(EQUAL, KEY, VAL_1);
+
+ assertTrue(cond.test(new Entry(KEY, VAL_1, 1, 1)));
+ }
+
+ /**
+ * Tests values inequality.
+ */
+ @Test
+ public void ne() {
+ Condition cond = new ValueCondition(NOT_EQUAL, KEY, VAL_1);
+
+ assertTrue(cond.test(new Entry(KEY, VAL_2, 1, 1)));
+ }
+}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
index 669b937..945be3f 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.metastorage;
import java.util.Collection;
-import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -32,14 +31,13 @@ import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.metastorage.client.MetaStorageService;
-import org.apache.ignite.metastorage.common.CompactedException;
-import org.apache.ignite.metastorage.common.Condition;
-import org.apache.ignite.metastorage.common.Cursor;
-import org.apache.ignite.metastorage.common.Entry;
-import org.apache.ignite.metastorage.common.Key;
-import org.apache.ignite.metastorage.common.Operation;
-import org.apache.ignite.metastorage.common.OperationTimeoutException;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.metastorage.client.CompactedException;
+import org.apache.ignite.metastorage.client.Condition;
+import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.metastorage.client.Entry;
+import org.apache.ignite.metastorage.client.Operation;
+import org.apache.ignite.metastorage.client.OperationTimeoutException;
+import org.apache.ignite.metastorage.client.WatchListener;
import org.apache.ignite.network.ClusterService;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -155,7 +153,7 @@ import org.jetbrains.annotations.Nullable;
* subscription
*/
public synchronized CompletableFuture<Long> registerWatch(
- @Nullable Key key,
+ @Nullable ByteArray key,
@NotNull WatchListener lsnr
) {
return waitForReDeploy(watchAggregator.add(key, lsnr));
@@ -170,7 +168,7 @@ import org.jetbrains.annotations.Nullable;
* subscription
*/
public synchronized CompletableFuture<Long> registerWatchByPrefix(
- @Nullable Key key,
+ @Nullable ByteArray key,
@NotNull WatchListener lsnr
) {
return waitForReDeploy(watchAggregator.addPrefix(key, lsnr));
@@ -185,7 +183,7 @@ import org.jetbrains.annotations.Nullable;
* subscription
*/
public synchronized CompletableFuture<Long> registerWatch(
- @NotNull Collection<Key> keys,
+ @NotNull Collection<ByteArray> keys,
@NotNull WatchListener lsnr
) {
return waitForReDeploy(watchAggregator.add(keys, lsnr));
@@ -200,8 +198,8 @@ import org.jetbrains.annotations.Nullable;
* @return future with id of registered watch.
*/
public synchronized CompletableFuture<Long> registerWatch(
- @NotNull Key from,
- @NotNull Key to,
+ @NotNull ByteArray from,
+ @NotNull ByteArray to,
@NotNull WatchListener lsnr
) {
return waitForReDeploy(watchAggregator.add(from, to, lsnr));
@@ -222,129 +220,117 @@ import org.jetbrains.annotations.Nullable;
}
/**
- * @see MetaStorageService#get(Key)
+ * @see MetaStorageService#get(ByteArray)
*/
- public @NotNull CompletableFuture<Entry> get(@NotNull Key key) {
+ public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key) {
return metaStorageSvc.get(key);
}
/**
- * @see MetaStorageService#get(Key, long)
+ * @see MetaStorageService#get(ByteArray, long)
*/
- public @NotNull CompletableFuture<Entry> get(@NotNull Key key, long revUpperBound) {
+ public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key, long revUpperBound) {
return metaStorageSvc.get(key, revUpperBound);
}
/**
* @see MetaStorageService#getAll(Collection)
*/
- public @NotNull CompletableFuture<Map<Key, Entry>> getAll(Collection<Key> keys) {
+ public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAll(Collection<ByteArray> keys) {
return metaStorageSvc.getAll(keys);
}
/**
* @see MetaStorageService#getAll(Collection, long)
*/
- public @NotNull CompletableFuture<Map<Key, Entry>> getAll(Collection<Key> keys, long revUpperBound) {
+ public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAll(Collection<ByteArray> keys, long revUpperBound) {
return metaStorageSvc.getAll(keys, revUpperBound);
}
/**
- * @see MetaStorageService#put(Key, byte[])
+ * @see MetaStorageService#put(ByteArray, byte[])
*/
- public @NotNull CompletableFuture<Void> put(@NotNull Key key, byte[] val) {
+ public @NotNull CompletableFuture<Void> put(@NotNull ByteArray key, byte[] val) {
return metaStorageSvc.put(key, val);
}
/**
- * @see MetaStorageService#getAndPut(Key, byte[])
+ * @see MetaStorageService#getAndPut(ByteArray, byte[])
*/
- public @NotNull CompletableFuture<Entry> getAndPut(@NotNull Key key, byte[] val) {
+ public @NotNull CompletableFuture<Entry> getAndPut(@NotNull ByteArray key, byte[] val) {
return metaStorageSvc.getAndPut(key, val);
}
/**
* @see MetaStorageService#putAll(Map)
*/
- public @NotNull CompletableFuture<Void> putAll(@NotNull Map<Key, byte[]> vals) {
+ public @NotNull CompletableFuture<Void> putAll(@NotNull Map<ByteArray, byte[]> vals) {
return metaStorageSvc.putAll(vals);
}
/**
* @see MetaStorageService#getAndPutAll(Map)
*/
- public @NotNull CompletableFuture<Map<Key, Entry>> getAndPutAll(@NotNull Map<Key, byte[]> vals) {
+ public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(@NotNull Map<ByteArray, byte[]> vals) {
return metaStorageSvc.getAndPutAll(vals);
}
/**
- * @see MetaStorageService#remove(Key)
+ * @see MetaStorageService#remove(ByteArray)
*/
- public @NotNull CompletableFuture<Void> remove(@NotNull Key key) {
+ public @NotNull CompletableFuture<Void> remove(@NotNull ByteArray key) {
return metaStorageSvc.remove(key);
}
/**
- * @see MetaStorageService#getAndRemove(Key)
+ * @see MetaStorageService#getAndRemove(ByteArray)
*/
- public @NotNull CompletableFuture<Entry> getAndRemove(@NotNull Key key) {
+ public @NotNull CompletableFuture<Entry> getAndRemove(@NotNull ByteArray key) {
return metaStorageSvc.getAndRemove(key);
}
/**
* @see MetaStorageService#removeAll(Collection)
*/
- public @NotNull CompletableFuture<Void> removeAll(@NotNull Collection<Key> keys) {
+ public @NotNull CompletableFuture<Void> removeAll(@NotNull Collection<ByteArray> keys) {
return metaStorageSvc.removeAll(keys);
}
/**
* @see MetaStorageService#getAndRemoveAll(Collection)
*/
- public @NotNull CompletableFuture<Map<Key, Entry>> getAndRemoveAll(@NotNull Collection<Key> keys) {
+ public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(@NotNull Collection<ByteArray> keys) {
return metaStorageSvc.getAndRemoveAll(keys);
}
/**
* Invoke with single success/failure operation.
*
- * @see MetaStorageService#invoke(Condition, Collection, Collection)
+ * @see MetaStorageService#invoke(Condition, Operation, Operation)
*/
public @NotNull CompletableFuture<Boolean> invoke(
@NotNull Condition cond,
@NotNull Operation success,
@NotNull Operation failure
) {
- return metaStorageSvc.invoke(cond, Collections.singletonList(success), Collections.singletonList(failure));
+ return metaStorageSvc.invoke(cond, success, failure);
}
/**
* @see MetaStorageService#invoke(Condition, Collection, Collection)
*/
public @NotNull CompletableFuture<Boolean> invoke(
- @NotNull Condition cond,
- @NotNull Collection<Operation> success,
- @NotNull Collection<Operation> failure
+ @NotNull Condition cond,
+ @NotNull Collection<Operation> success,
+ @NotNull Collection<Operation> failure
) {
return metaStorageSvc.invoke(cond, success, failure);
}
/**
- * @see MetaStorageService#getAndInvoke(Key, Condition, Operation, Operation)
- */
- public @NotNull CompletableFuture<Entry> getAndInvoke(
- @NotNull Key key,
- @NotNull Condition cond,
- @NotNull Operation success,
- @NotNull Operation failure
- ) {
- return metaStorageSvc.getAndInvoke(key, cond, success, failure);
- }
-
- /**
- * @see MetaStorageService#range(Key, Key, long)
+ * @see MetaStorageService#range(ByteArray, ByteArray, long)
*/
- public @NotNull Cursor<Entry> range(@NotNull Key keyFrom, @Nullable Key keyTo, long revUpperBound) {
+ public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound) {
return metaStorageSvc.range(keyFrom, keyTo, revUpperBound);
}
@@ -358,10 +344,10 @@ import org.jetbrains.annotations.Nullable;
* @return Cursor built upon entries corresponding to the given range and applied revision.
* @throws OperationTimeoutException If the operation is timed out.
* @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
- * @see Key
+ * @see ByteArray
* @see Entry
*/
- public @NotNull Cursor<Entry> rangeWithAppliedRevision(@NotNull Key keyFrom, @Nullable Key keyTo) {
+ public @NotNull Cursor<Entry> rangeWithAppliedRevision(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo) {
try {
return metaStorageSvc.range(keyFrom, keyTo, vaultMgr.appliedRevision());
}
@@ -371,9 +357,9 @@ import org.jetbrains.annotations.Nullable;
}
/**
- * @see MetaStorageService#range(Key, Key)
+ * @see MetaStorageService#range(ByteArray, ByteArray)
*/
- public @NotNull Cursor<Entry> range(@NotNull Key keyFrom, @Nullable Key keyTo) {
+ public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo) {
return metaStorageSvc.range(keyFrom, keyTo);
}
@@ -425,13 +411,14 @@ import org.jetbrains.annotations.Nullable;
* @param revision associated revision.
* @return future, which will be completed when store action finished.
*/
- private CompletableFuture<Void> storeEntries(Collection<IgniteBiTuple<Key, byte[]>> entries, long revision) {
+ private CompletableFuture<Void> storeEntries(Collection<IgniteBiTuple<ByteArray, byte[]>> entries, long revision) {
try {
return vaultMgr.putAll(entries.stream().collect(
Collectors.toMap(
- e -> ByteArray.fromString(e.getKey().toString()),
- IgniteBiTuple::getValue)),
- revision);
+ e -> ByteArray.fromString(e.getKey().toString()),
+ IgniteBiTuple::getValue)
+ ),
+ revision);
}
catch (IgniteInternalCheckedException e) {
throw new IgniteInternalException("Couldn't put entries with considered revision.", e);
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/AggregatedWatch.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/AggregatedWatch.java
index 754bf76..cfedfbf 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/AggregatedWatch.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/AggregatedWatch.java
@@ -17,7 +17,7 @@
package org.apache.ignite.internal.metastorage.watch;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.metastorage.client.WatchListener;
/**
* Watch implementation with associated revision.
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java
index 654b872..622589b 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java
@@ -20,8 +20,8 @@ package org.apache.ignite.internal.metastorage.watch;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteBiTuple;
-import org.apache.ignite.metastorage.common.Key;
/**
* Filter for listen key's changes on metastore.
@@ -32,38 +32,38 @@ public interface KeyCriterion {
*
* @return Ignite tuple with first key as start of range and second as the end.
*/
- public IgniteBiTuple<Key, Key> toRange();
+ public IgniteBiTuple<ByteArray, ByteArray> toRange();
/**
* Check if this key criterion contains the key.
*
* @return true if criterion contains the key, false otherwise.
*/
- public boolean contains(Key key);
+ public boolean contains(ByteArray key);
/**
* Simple criterion which contains exactly one key.
*/
static class ExactCriterion implements KeyCriterion {
/** The key of criterion. */
- private final Key key;
+ private final ByteArray key;
/**
* Creates the instance of exact criterion.
*
* @param key Instance of the reference key.
*/
- public ExactCriterion(Key key) {
+ public ExactCriterion(ByteArray key) {
this.key = key;
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<Key, Key> toRange() {
+ @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
return new IgniteBiTuple<>(key, key);
}
/** {@inheritDoc} */
- @Override public boolean contains(Key key) {
+ @Override public boolean contains(ByteArray key) {
return this.key.equals(key);
}
@@ -74,10 +74,10 @@ public interface KeyCriterion {
*/
static class RangeCriterion implements KeyCriterion {
/** Start of the range. */
- private final Key from;
+ private final ByteArray from;
/** End of the range (exclusive). */
- private final Key to;
+ private final ByteArray to;
/**
* Creates the instance of range criterion.
@@ -85,18 +85,18 @@ public interface KeyCriterion {
* @param from Start of the range.
* @param to End of the range (exclusive).
*/
- public RangeCriterion(Key from, Key to) {
+ public RangeCriterion(ByteArray from, ByteArray to) {
this.from = from;
this.to = to;
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<Key, Key> toRange() {
+ @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
return new IgniteBiTuple<>(from, to);
}
/** {@inheritDoc} */
- @Override public boolean contains(Key key) {
+ @Override public boolean contains(ByteArray key) {
return key.compareTo(from) >= 0 && key.compareTo(to) < 0;
}
}
@@ -106,24 +106,24 @@ public interface KeyCriterion {
*/
static class CollectionCriterion implements KeyCriterion {
/** Collection of keys. */
- private final Collection<Key> keys;
+ private final Collection<ByteArray> keys;
/**
* Creates the instance of collection criterion.
*
* @param keys Collection of keys.
*/
- public CollectionCriterion(Collection<Key> keys) {
+ public CollectionCriterion(Collection<ByteArray> keys) {
this.keys = keys;
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<Key, Key> toRange() {
+ @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
return new IgniteBiTuple<>(Collections.min(keys), Collections.max(keys));
}
/** {@inheritDoc} */
- @Override public boolean contains(Key key) {
+ @Override public boolean contains(ByteArray key) {
return keys.contains(key);
}
}
@@ -133,29 +133,31 @@ public interface KeyCriterion {
*/
static class PrefixCriterion implements KeyCriterion {
/** Prefix of the key. */
- private final Key prefixKey;
+ private final ByteArray prefixKey;
/**
* Creates the instance of prefix key criterion.
*
* @param prefixKey Prefix of the key.
*/
- public PrefixCriterion(Key prefixKey) {
+ public PrefixCriterion(ByteArray prefixKey) {
this.prefixKey = prefixKey;
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<Key, Key> toRange() {
+ @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
var bytes = Arrays.copyOf(prefixKey.bytes(), prefixKey.bytes().length);
+
if (bytes[bytes.length - 1] != Byte.MAX_VALUE)
bytes[bytes.length - 1]++;
else
bytes = Arrays.copyOf(bytes, bytes.length + 1);
- return new IgniteBiTuple<>(prefixKey, new Key(bytes));
+
+ return new IgniteBiTuple<>(prefixKey, new ByteArray(bytes));
}
/** {@inheritDoc} */
- @Override public boolean contains(Key key) {
+ @Override public boolean contains(ByteArray key) {
return key.compareTo(prefixKey) >= 0 && key.compareTo(toRange().getValue()) < 0;
}
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
index 35c8881..6fb8023 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
@@ -25,10 +25,12 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
+
+import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteBiTuple;
-import org.apache.ignite.metastorage.common.Key;
-import org.apache.ignite.metastorage.common.WatchEvent;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.metastorage.client.EntryEvent;
+import org.apache.ignite.metastorage.client.WatchEvent;
+import org.apache.ignite.metastorage.client.WatchListener;
import org.jetbrains.annotations.NotNull;
/**
@@ -44,7 +46,7 @@ public class WatchAggregator {
private final Map<Long, Watch> watches = Collections.synchronizedMap(new LinkedHashMap<>());
/** Simple auto increment id for internal watches. */
- private AtomicLong idCntr = new AtomicLong(0);
+ private final AtomicLong idCntr = new AtomicLong(0);
/**
* Adds new watch with simple exact criterion.
@@ -53,7 +55,7 @@ public class WatchAggregator {
* @param lsnr Listener which will be executed on watch event.
* @return id of registered watch. Can be used for remove watch from later.
*/
- public long add(Key key, WatchListener lsnr) {
+ public long add(ByteArray key, WatchListener lsnr) {
var watch = new Watch(new KeyCriterion.ExactCriterion(key), lsnr);
var id = idCntr.incrementAndGet();
watches.put(id, watch);
@@ -67,7 +69,7 @@ public class WatchAggregator {
* @param lsnr Listener which will be executed on watch event.
* @return id of registered watch. Can be used for remove watch from later.
*/
- public long addPrefix(Key key, WatchListener lsnr) {
+ public long addPrefix(ByteArray key, WatchListener lsnr) {
var watch = new Watch(new KeyCriterion.PrefixCriterion(key), lsnr);
var id = idCntr.incrementAndGet();
watches.put(id, watch);
@@ -81,7 +83,7 @@ public class WatchAggregator {
* @param lsnr Listener which will be executed on watch event.
* @return id of registered watch. Can be used for remove watch from later.
*/
- public long add(Collection<Key> keys, WatchListener lsnr) {
+ public long add(Collection<ByteArray> keys, WatchListener lsnr) {
var watch = new Watch(new KeyCriterion.CollectionCriterion(keys), lsnr);
var id = idCntr.incrementAndGet();
watches.put(id, watch);
@@ -96,7 +98,7 @@ public class WatchAggregator {
* @param lsnr Listener which will be executed on watch event.
* @return id of registered watch. Can be used for remove watch from later.
*/
- public long add(Key from, Key to, WatchListener lsnr) {
+ public long add(ByteArray from, ByteArray to, WatchListener lsnr) {
var watch = new Watch(new KeyCriterion.RangeCriterion(from, to), lsnr);
var id = idCntr.incrementAndGet();
watches.put(id, watch);
@@ -128,7 +130,10 @@ public class WatchAggregator {
* @param saveRevisionAct action to commit keys-revision pair to persistent store for processed keys.
* @return result aggregated watch.
*/
- public Optional<AggregatedWatch> watch(long revision, BiConsumer<Collection<IgniteBiTuple<Key, byte[]>>, Long> saveRevisionAct) {
+ public Optional<AggregatedWatch> watch(
+ long revision,
+ BiConsumer<Collection<IgniteBiTuple<ByteArray, byte[]>>, Long> saveRevisionAct
+ ) {
synchronized (watches) {
if (watches.isEmpty())
return Optional.empty();
@@ -145,8 +150,8 @@ public class WatchAggregator {
// TODO: IGNITE-14667 We can do it better than infer range always
private KeyCriterion inferGeneralCriteria() {
return new KeyCriterion.RangeCriterion(
- watches.values().stream().map(w -> w.keyCriterion().toRange().getKey()).min(Key::compareTo).get(),
- watches.values().stream().map(w -> w.keyCriterion().toRange().getValue()).max(Key::compareTo).get()
+ watches.values().stream().map(w -> w.keyCriterion().toRange().getKey()).min(ByteArray::compareTo).get(),
+ watches.values().stream().map(w -> w.keyCriterion().toRange().getValue()).max(ByteArray::compareTo).get()
);
}
@@ -156,7 +161,7 @@ public class WatchAggregator {
* @param storeRevision action to commit keys-revision pair to persistent store for processed keys.
* @return watch listener, which will dispatch events to appropriate watches.
*/
- private WatchListener watchListener(BiConsumer<Collection<IgniteBiTuple<Key, byte[]>>, Long> storeRevision) {
+ private WatchListener watchListener(BiConsumer<Collection<IgniteBiTuple<ByteArray, byte[]>>, Long> storeRevision) {
// Copy watches to separate collection, because all changes on the WatchAggregator watches
// shouldn't be propagated to listener watches immediately.
// WatchAggregator will be redeployed with new watches if needed instead.
@@ -164,22 +169,22 @@ public class WatchAggregator {
return new WatchListener() {
- @Override public boolean onUpdate(@NotNull Iterable<WatchEvent> evts) {
+ @Override public boolean onUpdate(@NotNull WatchEvent evt) {
var watchIt = cpWatches.entrySet().iterator();
Collection<Long> toCancel = new ArrayList<>();
while (watchIt.hasNext()) {
Map.Entry<Long, WatchAggregator.Watch> entry = watchIt.next();
WatchAggregator.Watch watch = entry.getValue();
- var filteredEvts = new ArrayList<WatchEvent>();
+ var filteredEvts = new ArrayList<EntryEvent>();
- for (WatchEvent evt : evts) {
- if (watch.keyCriterion().contains(evt.oldEntry().key()))
- filteredEvts.add(evt);
+ for (EntryEvent entryEvt : evt.entryEvents()) {
+ if (watch.keyCriterion().contains(entryEvt.oldEntry().key()))
+ filteredEvts.add(entryEvt);
}
if (!filteredEvts.isEmpty()) {
- if (!watch.listener().onUpdate(filteredEvts)) {
+ if (!watch.listener().onUpdate(new WatchEvent(filteredEvts))) {
watchIt.remove();
toCancel.add(entry.getKey());
@@ -193,11 +198,11 @@ public class WatchAggregator {
cancelAll(toCancel);
var revision = 0L;
- var entries = new ArrayList<IgniteBiTuple<Key, byte[]>>();
- for (WatchEvent evt: evts) {
- revision = evt.newEntry().revision();
+ var entries = new ArrayList<IgniteBiTuple<ByteArray, byte[]>>();
+ for (EntryEvent entryEvt: evt.entryEvents()) {
+ revision = entryEvt.newEntry().revision();
- entries.add(new IgniteBiTuple<>(evt.newEntry().key(), evt.newEntry().value()));
+ entries.add(new IgniteBiTuple<>(entryEvt.newEntry().key(), entryEvt.newEntry().value()));
}
storeRevision.accept(entries, revision);
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java
index 4eb1901..038b774 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java
@@ -18,17 +18,16 @@
package org.apache.ignite.internal.metastorage;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.Collections;
+import java.util.List;
import org.apache.ignite.internal.metastorage.watch.WatchAggregator;
-import org.apache.ignite.metastorage.common.Entry;
-import org.apache.ignite.metastorage.common.Key;
-import org.apache.ignite.metastorage.common.WatchEvent;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.lang.ByteArray;
+import org.apache.ignite.metastorage.client.Entry;
+import org.apache.ignite.metastorage.client.EntryEvent;
+import org.apache.ignite.metastorage.client.WatchEvent;
+import org.apache.ignite.metastorage.client.WatchListener;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.Test;
-
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -42,19 +41,27 @@ public class WatchAggregatorTest {
var watchAggregator = new WatchAggregator();
var lsnr1 = mock(WatchListener.class);
var lsnr2 = mock(WatchListener.class);
- watchAggregator.add(new Key("1"), lsnr1);
- watchAggregator.add(new Key("2"), lsnr2);
+ watchAggregator.add(new ByteArray("1"), lsnr1);
+ watchAggregator.add(new ByteArray("2"), lsnr2);
- var watchEvent1 = new WatchEvent(
- entry("1", "value1", 1),
- entry("1", "value1n", 1));
- var watchEvent2 = new WatchEvent(
- entry("2", "value2", 1),
- entry("2", "value2n", 1));
- watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(Arrays.asList(watchEvent1, watchEvent2));
+ var entryEvt1 = new EntryEvent(
+ entry("1", "value1", 1),
+ entry("1", "value1n", 1)
+ );
+
+ var watchEvent1 = new WatchEvent(entryEvt1);
+
+ var entryEvt2 = new EntryEvent(
+ entry("2", "value2", 1),
+ entry("2", "value2n", 1)
+ );
+
+ var watchEvent2 = new WatchEvent(entryEvt2);
+
+ watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(new WatchEvent(List.of(entryEvt1, entryEvt2)));
- verify(lsnr1).onUpdate(Collections.singletonList(watchEvent1));
- verify(lsnr2).onUpdate(Collections.singletonList(watchEvent2));
+ verify(lsnr1).onUpdate(watchEvent1);
+ verify(lsnr2).onUpdate(watchEvent2);
}
@Test
@@ -64,22 +71,26 @@ public class WatchAggregatorTest {
when(lsnr1.onUpdate(any())).thenReturn(true);
var lsnr2 = mock(WatchListener.class);
when(lsnr2.onUpdate(any())).thenReturn(true);
- var id1 = watchAggregator.add(new Key("1"), lsnr1);
- var id2 = watchAggregator.add(new Key("2"), lsnr2);
+ var id1 = watchAggregator.add(new ByteArray("1"), lsnr1);
+ var id2 = watchAggregator.add(new ByteArray("2"), lsnr2);
- var watchEvent1 = new WatchEvent(
- entry("1", "value1", 1),
- entry("1", "value1n", 1));
- var watchEvent2 = new WatchEvent(
- entry("2", "value2", 1),
- entry("2", "value2n", 1));
- watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(Arrays.asList(watchEvent1, watchEvent2));
+ var entryEvt1 = new EntryEvent(
+ entry("1", "value1", 1),
+ entry("1", "value1n", 1)
+ );
+
+ var entryEvt2 = new EntryEvent(
+ entry("2", "value2", 1),
+ entry("2", "value2n", 1)
+ );
+
+ watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(new WatchEvent(List.of(entryEvt1, entryEvt2)));
verify(lsnr1, times(1)).onUpdate(any());
verify(lsnr2, times(1)).onUpdate(any());
watchAggregator.cancel(id1);
- watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(Arrays.asList(watchEvent1, watchEvent2));
+ watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(new WatchEvent(List.of(entryEvt1, entryEvt2)));
verify(lsnr1, times(1)).onUpdate(any());
verify(lsnr2, times(2)).onUpdate(any());
@@ -92,21 +103,25 @@ public class WatchAggregatorTest {
when(lsnr1.onUpdate(any())).thenReturn(false);
var lsnr2 = mock(WatchListener.class);
when(lsnr2.onUpdate(any())).thenReturn(true);
- var id1 = watchAggregator.add(new Key("1"), lsnr1);
- var id2 = watchAggregator.add(new Key("2"), lsnr2);
+ var id1 = watchAggregator.add(new ByteArray("1"), lsnr1);
+ var id2 = watchAggregator.add(new ByteArray("2"), lsnr2);
- var watchEvent1 = new WatchEvent(
+ var entryEvt1 = new EntryEvent(
entry("1", "value1", 1),
- entry("1", "value1n", 1));
- var watchEvent2 = new WatchEvent(
+ entry("1", "value1n", 1)
+ );
+
+ var entryEvt2 = new EntryEvent(
entry("2", "value2", 1),
- entry("2", "value2n", 1));
- watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(Arrays.asList(watchEvent1, watchEvent2));
+ entry("2", "value2n", 1)
+ );
+
+ watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(new WatchEvent(List.of(entryEvt1, entryEvt2)));
verify(lsnr1, times(1)).onUpdate(any());
verify(lsnr2, times(1)).onUpdate(any());
- watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(Arrays.asList(watchEvent1, watchEvent2));
+ watchAggregator.watch(1, (v1, v2) -> {}).get().listener().onUpdate(new WatchEvent(List.of(entryEvt1, entryEvt2)));
verify(lsnr1, times(1)).onUpdate(any());
verify(lsnr2, times(2)).onUpdate(any());
@@ -115,8 +130,8 @@ public class WatchAggregatorTest {
private Entry entry(String key, String value, long revision) {
return new Entry() {
- @Override public @NotNull Key key() {
- return new Key(key);
+ @Override public @NotNull ByteArray key() {
+ return new ByteArray(key);
}
@Override public @Nullable byte[] value() {
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java
index 4e608eb..8effc42 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java
@@ -26,7 +26,6 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
import org.apache.ignite.app.Ignite;
import org.apache.ignite.app.Ignition;
import org.apache.ignite.configuration.RootKey;
@@ -48,15 +47,15 @@ import org.apache.ignite.internal.storage.LocalConfigurationStorage;
import org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.impl.VaultServiceImpl;
+import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.metastorage.client.MetaStorageService;
-import org.apache.ignite.metastorage.common.Condition;
-import org.apache.ignite.metastorage.common.Cursor;
-import org.apache.ignite.metastorage.common.Entry;
-import org.apache.ignite.metastorage.common.Key;
-import org.apache.ignite.metastorage.common.Operation;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.metastorage.client.Condition;
+import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.metastorage.client.Entry;
+import org.apache.ignite.metastorage.client.Operation;
+import org.apache.ignite.metastorage.client.WatchListener;
import org.apache.ignite.network.ClusterLocalConfiguration;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.message.MessageSerializationRegistry;
@@ -202,9 +201,7 @@ public class IgnitionImpl implements Ignition {
private static void ackBanner() {
String ver = IgniteProperties.get(VER_KEY);
- String banner = Arrays
- .stream(BANNER)
- .collect(Collectors.joining("\n"));
+ String banner = String.join("\n", BANNER);
LOG.info(banner + '\n' + " ".repeat(22) + "Apache Ignite ver. " + ver + '\n');
}
@@ -212,94 +209,96 @@ public class IgnitionImpl implements Ignition {
// TODO: remove when metastorage service will be ready.
private static MetaStorageService metaStorageServiceMock() {
return new MetaStorageService() {
- @Override public @NotNull CompletableFuture<Entry> get(@NotNull Key key) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Entry> get(@NotNull Key key, long revUpperBound) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key, long revUpperBound) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Map<Key, Entry>> getAll(Collection<Key> keys) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAll(Collection<ByteArray> keys) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Map<Key, Entry>> getAll(Collection<Key> keys, long revUpperBound) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAll(Collection<ByteArray> keys, long revUpperBound) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Void> put(@NotNull Key key, @NotNull byte[] value) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Void> put(@NotNull ByteArray key, @NotNull byte[] value) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Entry> getAndPut(@NotNull Key key, @NotNull byte[] value) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Entry> getAndPut(@NotNull ByteArray key, @NotNull byte[] value) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Void> putAll(@NotNull Map<Key, byte[]> vals) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Void> putAll(@NotNull Map<ByteArray, byte[]> vals) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Map<Key, Entry>> getAndPutAll(@NotNull Map<Key, byte[]> vals) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(@NotNull Map<ByteArray, byte[]> vals) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Void> remove(@NotNull Key key) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Void> remove(@NotNull ByteArray key) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Entry> getAndRemove(@NotNull Key key) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Entry> getAndRemove(@NotNull ByteArray key) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Void> removeAll(@NotNull Collection<Key> keys) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Void> removeAll(@NotNull Collection<ByteArray> keys) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
@Override
- public @NotNull CompletableFuture<Map<Key, Entry>> getAndRemoveAll(@NotNull Collection<Key> keys) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(@NotNull Collection<ByteArray> keys) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
@Override public @NotNull CompletableFuture<Boolean> invoke(@NotNull Condition condition,
- @NotNull Collection<Operation> success, @NotNull Collection<Operation> failure) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @NotNull Operation success, @NotNull Operation failure) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<Entry> getAndInvoke(@NotNull Key key, @NotNull Condition condition,
- @NotNull Operation success, @NotNull Operation failure) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull CompletableFuture<Boolean> invoke(@NotNull Condition condition,
+ @NotNull Collection<Operation> success,
+ @NotNull Collection<Operation> failure
+ ) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull Cursor<Entry> range(@NotNull Key keyFrom, @Nullable Key keyTo, long revUpperBound) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull Cursor<Entry> range(@NotNull Key keyFrom, @Nullable Key keyTo) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ @Override public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo) {
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
- @Override public @NotNull CompletableFuture<IgniteUuid> watch(@Nullable Key keyFrom, @Nullable Key keyTo,
+ @Override public @NotNull CompletableFuture<IgniteUuid> watch(@Nullable ByteArray keyFrom, @Nullable ByteArray keyTo,
long revision, @NotNull WatchListener lsnr) {
return CompletableFuture.completedFuture(new IgniteUuid(UUID.randomUUID(), 0L));
}
- @Override public @NotNull CompletableFuture<IgniteUuid> watch(@NotNull Key key, long revision,
+ @Override public @NotNull CompletableFuture<IgniteUuid> watch(@NotNull ByteArray key, long revision,
@NotNull WatchListener lsnr) {
return CompletableFuture.completedFuture(new IgniteUuid(UUID.randomUUID(), 0L));
}
- @Override public @NotNull CompletableFuture<IgniteUuid> watch(@NotNull Collection<Key> keys, long revision,
+ @Override public @NotNull CompletableFuture<IgniteUuid> watch(@NotNull Collection<ByteArray> keys, long revision,
@NotNull WatchListener lsnr) {
return CompletableFuture.completedFuture(new IgniteUuid(UUID.randomUUID(), 0L));
}
@Override public @NotNull CompletableFuture<Void> stopWatch(@NotNull IgniteUuid id) {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
@Override public @NotNull CompletableFuture<Void> compact() {
- throw new UnsupportedOperationException("Metastorage service is not implemented yet");
+ throw new UnsupportedOperationException("Meta storage service is not implemented yet");
}
};
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 012d57b..197777d 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -50,11 +50,7 @@ import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
-import org.apache.ignite.metastorage.common.Conditions;
-import org.apache.ignite.metastorage.common.Key;
-import org.apache.ignite.metastorage.common.Operations;
-import org.apache.ignite.metastorage.common.WatchEvent;
-import org.apache.ignite.metastorage.common.WatchListener;
+import org.apache.ignite.metastorage.client.*;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.table.Table;
@@ -81,9 +77,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
private CompletableFuture<Long> tableCreationSubscriptionFut;
/** Tables. */
- private Map<String, TableImpl> tables = new ConcurrentHashMap<>();
+ private final Map<String, TableImpl> tables = new ConcurrentHashMap<>();
- /*
+ /**
* @param configurationMgr Configuration manager.
* @param metaStorageMgr Meta storage manager.
* @param schemaManager Schema manager.
@@ -123,9 +119,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
String tableInternalPrefix = INTERNAL_PREFIX + "assignment.";
- tableCreationSubscriptionFut = metaStorageMgr.registerWatchByPrefix(new Key(tableInternalPrefix), new WatchListener() {
- @Override public boolean onUpdate(@NotNull Iterable<WatchEvent> events) {
- for (WatchEvent evt : events) {
+ tableCreationSubscriptionFut = metaStorageMgr.registerWatchByPrefix(new ByteArray(tableInternalPrefix), new WatchListener() {
+ @Override public boolean onUpdate(@NotNull WatchEvent watchEvt) {
+ for (EntryEvent evt : watchEvt.entryEvents()) {
String placeholderValue = evt.newEntry().key().toString().substring(tableInternalPrefix.length() - 1);
String name = new String(vaultManager.get(ByteArray.fromString(INTERNAL_PREFIX + placeholderValue))
@@ -228,7 +224,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
configurationMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY)
.tables().listen(ctx -> {
Set<String> tablesToStart = ctx.newValue().namedListKeys() == null ?
- Collections.EMPTY_SET : ctx.newValue().namedListKeys();
+ Collections.emptySet() : ctx.newValue().namedListKeys();
tablesToStart.removeAll(ctx.oldValue().namedListKeys());
@@ -242,18 +238,19 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
UUID tblId = new UUID(revision, update);
- var key = new Key(INTERNAL_PREFIX + tblId.toString());
+ var key = new ByteArray(INTERNAL_PREFIX + tblId);
+
futs.add(metaStorageMgr.invoke(
- Conditions.key(key).value().eq(null),
+ Conditions.notExists(key),
Operations.put(key, tableView.name().getBytes(StandardCharsets.UTF_8)),
Operations.noop()).thenCompose(res ->
- res ? metaStorageMgr.put(new Key(INTERNAL_PREFIX + "assignment." + tblId.toString()), new byte[0])
+ res ? metaStorageMgr.put(new ByteArray(INTERNAL_PREFIX + "assignment." + tblId), new byte[0])
.thenApply(v -> true)
: CompletableFuture.completedFuture(false)));
}
Set<String> tablesToStop = ctx.oldValue().namedListKeys() == null ?
- Collections.EMPTY_SET : ctx.oldValue().namedListKeys();
+ Collections.emptySet() : ctx.oldValue().namedListKeys();
tablesToStop.removeAll(ctx.newValue().namedListKeys());
@@ -262,12 +259,13 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
UUID tblId = t.internalTable().tableId();
- var key = new Key(INTERNAL_PREFIX + "assignment." + tblId.toString());
+ var key = new ByteArray(INTERNAL_PREFIX + "assignment." + tblId);
+
futs.add(metaStorageMgr.invoke(
- Conditions.key(key).value().ne(null),
+ Conditions.value(key).ne(null),
Operations.remove(key),
Operations.noop()).thenCompose(res ->
- res ? metaStorageMgr.remove(new Key(INTERNAL_PREFIX + tblId.toString()))
+ res ? metaStorageMgr.remove(new ByteArray(INTERNAL_PREFIX + tblId))
.thenApply(v -> true)
: CompletableFuture.completedFuture(false)));
}