You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2020/07/22 09:47:31 UTC
[ignite] branch master updated: IGNITE-13148 .NET: Add Thin Client
Continuous Query
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 5fe4b38 IGNITE-13148 .NET: Add Thin Client Continuous Query
5fe4b38 is described below
commit 5fe4b38e335f3f07f7cae4b5877b8726eb5f81df
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Wed Jul 22 12:47:08 2020 +0300
IGNITE-13148 .NET: Add Thin Client Continuous Query
* Add Continuous Query feature to the thin client protocol
* Implement Continuous Queries in the .NET thin client
IEP: https://cwiki.apache.org/confluence/display/IGNITE/IEP-50%3A+Thin+Client+Continuous+Queries
---
.../platform/client/ClientMessageParser.java | 10 +
.../processors/platform/client/ClientPlatform.java | 35 +
.../cache/ClientCacheEntryEventNotification.java | 84 +++
.../cache/ClientCacheQueryContinuousHandle.java | 94 +++
.../cache/ClientCacheQueryContinuousRequest.java | 147 ++++
.../cache/ClientCacheQueryContinuousResponse.java | 59 ++
.../client/cache/ClientCacheScanQueryRequest.java | 27 +-
.../PlatformCacheEntryEvenKeyEventFilter.java | 32 +
.../Apache.Ignite.Core.Tests.DotNetCore.csproj | 3 +
.../Apache.Ignite.Core.Tests.csproj | 1 +
.../Serializable/AdvancedSerializationTest.cs | 2 +-
.../Continuous/ContinuousQueryJavaFilterTest.cs | 2 +-
.../Client/Cache/CacheClientAsyncWrapper.cs | 7 +
.../Client/Cache/ContinuousQueryTest.cs | 802 +++++++++++++++++++++
.../Client/ClientConnectionTest.cs | 1 -
.../Client/ClientServerCacheAdapter.cs | 8 +-
.../Client/IgniteClientConfigurationTest.cs | 8 +-
.../Apache.Ignite.Core.Tests/IgniteLockTests.cs | 1 -
.../Services/ServicesTest.cs | 2 +-
.../Apache.Ignite.Core.Tests/TestUtils.Common.cs | 18 +-
.../Apache.Ignite.Core/Apache.Ignite.Core.csproj | 4 +
.../Cache/Query/Continuous/ContinuousQuery.cs | 32 +-
.../Client/Cache/ICacheClient.cs | 8 +
.../Query/Continuous/ContinuousQueryClient.cs | 101 +++
.../ContinuousQueryDisconnectedEventArgs.cs | 47 ++
.../Continuous/IContinuousQueryHandleClient.cs | 33 +
.../Impl/Cache/Query/QueryCursorBase.cs | 7 +-
.../Impl/Client/Cache/CacheClient.cs | 143 +++-
.../Cache/Query/ClientContinuousQueryHandle.cs | 96 +++
.../Apache.Ignite.Core/Impl/Client/ClientOp.cs | 6 +-
.../Apache.Ignite.Core/Impl/Client/ClientSocket.cs | 16 +-
.../Apache.Ignite.Core/Impl/Client/IgniteClient.cs | 9 +-
32 files changed, 1782 insertions(+), 63 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java
index 94d32bc..5664f15 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java
@@ -56,6 +56,7 @@ import org.apache.ignite.internal.processors.platform.client.cache.ClientCachePa
import org.apache.ignite.internal.processors.platform.client.cache.ClientCachePutAllRequest;
import org.apache.ignite.internal.processors.platform.client.cache.ClientCachePutIfAbsentRequest;
import org.apache.ignite.internal.processors.platform.client.cache.ClientCachePutRequest;
+import org.apache.ignite.internal.processors.platform.client.cache.ClientCacheQueryContinuousRequest;
import org.apache.ignite.internal.processors.platform.client.cache.ClientCacheQueryNextPageRequest;
import org.apache.ignite.internal.processors.platform.client.cache.ClientCacheRemoveAllRequest;
import org.apache.ignite.internal.processors.platform.client.cache.ClientCacheRemoveIfEqualsRequest;
@@ -202,6 +203,12 @@ public class ClientMessageParser implements ClientListenerMessageParser {
/** */
private static final short OP_QUERY_SQL_FIELDS_CURSOR_GET_PAGE = 2005;
+ /** */
+ private static final short OP_QUERY_CONTINUOUS = 2006;
+
+ /** */
+ public static final short OP_QUERY_CONTINUOUS_EVENT_NOTIFICATION = 2007;
+
/* Binary metadata operations. */
/** */
private static final short OP_BINARY_TYPE_NAME_GET = 3000;
@@ -424,6 +431,9 @@ public class ClientMessageParser implements ClientListenerMessageParser {
case OP_QUERY_SQL_FIELDS_CURSOR_GET_PAGE:
return new ClientCacheQueryNextPageRequest(reader);
+ case OP_QUERY_CONTINUOUS:
+ return new ClientCacheQueryContinuousRequest(reader);
+
case OP_TX_START:
return new ClientTxStartRequest(reader);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientPlatform.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientPlatform.java
new file mode 100644
index 0000000..ca27268
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientPlatform.java
@@ -0,0 +1,35 @@
+/*
+ * 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.processors.platform.client;
+
+/**
+ * Client platform codes.
+ */
+public class ClientPlatform {
+ /** */
+ public static final byte UNKNOWN = 0;
+
+ /** */
+ public static final byte JAVA = 1;
+
+ /** */
+ public static final byte DOTNET = 2;
+
+ /** */
+ public static final byte CPP = 3;
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheEntryEventNotification.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheEntryEventNotification.java
new file mode 100644
index 0000000..4f2db8b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheEntryEventNotification.java
@@ -0,0 +1,84 @@
+/*
+ * 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.processors.platform.client.cache;
+
+import javax.cache.event.CacheEntryEvent;
+import org.apache.ignite.internal.binary.BinaryRawWriterEx;
+import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
+import org.apache.ignite.internal.processors.platform.client.ClientNotification;
+
+/**
+ * Continuous query notification.
+ */
+@SuppressWarnings("rawtypes")
+public class ClientCacheEntryEventNotification extends ClientNotification {
+ /** */
+ private final Iterable<CacheEntryEvent<?, ?>> evts;
+
+ /**
+ * Ctor.
+ * @param opCode Operation code.
+ * @param rsrcId Resource ID.
+ * @param evts Events.
+ */
+ public ClientCacheEntryEventNotification(short opCode, long rsrcId, Iterable<CacheEntryEvent<?, ?>> evts) {
+ super(opCode, rsrcId);
+
+ assert evts != null;
+ this.evts = evts;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void encode(ClientConnectionContext ctx, BinaryRawWriterEx writer) {
+ super.encode(ctx, writer);
+
+ int pos = writer.reserveInt();
+ int cnt = 0;
+
+ for (CacheEntryEvent evt : evts) {
+ writer.writeObjectDetached(evt.getKey());
+ writer.writeObjectDetached(evt.getOldValue());
+ writer.writeObjectDetached(evt.getValue());
+
+ switch (evt.getEventType()) {
+ case CREATED:
+ writer.writeByte((byte) 0);
+ break;
+
+ case UPDATED:
+ writer.writeByte((byte) 1);
+ break;
+
+ case REMOVED:
+ writer.writeByte((byte) 2);
+ break;
+
+ case EXPIRED:
+ writer.writeByte((byte) 3);
+ break;
+
+ default:
+ throw new IllegalArgumentException("Unknown event type: " + evt.getEventType());
+ }
+
+ cnt++;
+ }
+
+ writer.writeInt(pos, cnt);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousHandle.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousHandle.java
new file mode 100644
index 0000000..127a367
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousHandle.java
@@ -0,0 +1,94 @@
+/*
+ * 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.processors.platform.client.cache;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.internal.processors.platform.client.ClientCloseableResource;
+import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
+import org.apache.ignite.internal.processors.platform.client.ClientMessageParser;
+
+/**
+ * Continuous query handle.
+ * NOTE: Do not mark with {@link org.apache.ignite.lang.IgniteAsyncCallback} - it disables batching and sends
+ * events one by one.
+ */
+public class ClientCacheQueryContinuousHandle implements CacheEntryUpdatedListener<Object, Object>, ClientCloseableResource {
+ /** */
+ private final ClientConnectionContext ctx;
+
+ /** */
+ private final AtomicBoolean closeGuard = new AtomicBoolean();
+
+ /** */
+ private volatile Long id;
+
+ /** */
+ private volatile QueryCursor<?> cur;
+
+ /**
+ * Ctor.
+ * @param ctx Context.
+ */
+ public ClientCacheQueryContinuousHandle(ClientConnectionContext ctx) {
+ assert ctx != null;
+
+ this.ctx = ctx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<?, ?>> iterable) throws CacheEntryListenerException {
+ // Client is not yet ready to receive notifications - skip them.
+ if (id == null)
+ return;
+
+ ClientCacheEntryEventNotification notification = new ClientCacheEntryEventNotification(
+ ClientMessageParser.OP_QUERY_CONTINUOUS_EVENT_NOTIFICATION, id, iterable);
+
+ ctx.notifyClient(notification);
+ }
+
+ /**
+ * Sets the cursor.
+ * @param cur Cursor.
+ */
+ public void setCursor(QueryCursor<?> cur) {
+ this.cur = cur;
+ }
+
+ /**
+ * Sets the cursor id.
+ * @param id Cursor id.
+ */
+ public void startNotifications(long id) {
+ this.id = id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ if (closeGuard.compareAndSet(false, true)) {
+ assert cur != null;
+ cur.close();
+
+ ctx.decrementCursors();
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousRequest.java
new file mode 100644
index 0000000..5075ca7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousRequest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.processors.platform.client.cache;
+
+import javax.cache.configuration.Factory;
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.event.CacheEntryEventFilter;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.internal.binary.BinaryObjectImpl;
+import org.apache.ignite.internal.binary.BinaryRawReaderEx;
+import org.apache.ignite.internal.binary.GridBinaryMarshaller;
+import org.apache.ignite.internal.processors.platform.PlatformContext;
+import org.apache.ignite.internal.processors.platform.PlatformJavaObjectFactoryProxy;
+import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
+import org.apache.ignite.internal.processors.platform.client.ClientPlatform;
+import org.apache.ignite.internal.processors.platform.client.ClientResponse;
+import org.apache.ignite.internal.processors.platform.client.ClientStatus;
+import org.apache.ignite.internal.processors.platform.client.IgniteClientException;
+import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
+
+/**
+ * Continuous query request.
+ */
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class ClientCacheQueryContinuousRequest extends ClientCacheRequest {
+ /** Query. */
+ private final ContinuousQuery qry;
+
+ /** */
+ private final Object filter;
+
+ /** */
+ private final byte filterPlatform;
+
+ /**
+ * Ctor.
+ *
+ * @param reader Reader.
+ */
+ public ClientCacheQueryContinuousRequest(BinaryRawReaderEx reader) {
+ super(reader);
+
+ int pageSize = reader.readInt();
+ long timeInterval = reader.readLong();
+ boolean includeExpired = reader.readBoolean();
+
+ filter = reader.readObjectDetached();
+ filterPlatform = filter == null ? 0 : reader.readByte();
+
+ qry = new ContinuousQuery();
+
+ qry.setPageSize(pageSize)
+ .setTimeInterval(timeInterval)
+ .setIncludeExpired(includeExpired);
+ }
+
+ /** {@inheritDoc} */
+ @Override public ClientResponse process(ClientConnectionContext ctx) {
+ qry.setRemoteFilterFactory(getFilterFactory(ctx));
+
+ ctx.incrementCursors();
+
+ try {
+ IgniteCache cache = filterPlatform == ClientPlatform.JAVA && !isKeepBinary()
+ ? rawCache(ctx)
+ : cache(ctx);
+
+ ClientCacheQueryContinuousHandle handle = new ClientCacheQueryContinuousHandle(ctx);
+ qry.setLocalListener(handle);
+
+ QueryCursor cur = cache.query(qry);
+ long cursorId = ctx.resources().put(handle);
+ handle.setCursor(cur);
+
+ return new ClientCacheQueryContinuousResponse(requestId(), handle, cursorId);
+ }
+ catch (Exception e) {
+ ctx.decrementCursors();
+ throw e;
+ }
+ }
+
+ /**
+ * Gets the filter factory.
+ *
+ * @param ctx Connection context.
+ * @return Filter factory or null.
+ */
+ private Factory<? extends CacheEntryEventFilter> getFilterFactory(ClientConnectionContext ctx) {
+ if (filter == null)
+ return null;
+
+ if (!(filter instanceof BinaryObject))
+ throw new IgniteClientException(ClientStatus.FAILED,
+ "Filter must be a BinaryObject: " + filter.getClass());
+
+ BinaryObjectImpl bo = (BinaryObjectImpl) filter;
+
+ switch (filterPlatform) {
+ case ClientPlatform.JAVA: {
+ if (bo.typeId() == GridBinaryMarshaller.PLATFORM_JAVA_OBJECT_FACTORY_PROXY) {
+ PlatformJavaObjectFactoryProxy prx = bo.deserialize();
+
+ CacheEntryEventSerializableFilter rmtFilter =
+ (CacheEntryEventSerializableFilter) prx.factory(ctx.kernalContext()).create();
+
+ return FactoryBuilder.factoryOf(rmtFilter);
+ }
+
+ return bo.deserialize();
+ }
+
+ case ClientPlatform.DOTNET: {
+ PlatformContext platformCtx = ctx.kernalContext().platform().context();
+
+ String curPlatform = platformCtx.platform();
+
+ if (!PlatformUtils.PLATFORM_DOTNET.equals(curPlatform))
+ throw new IgniteClientException(ClientStatus.FAILED, "ContinuousQuery filter platform is " +
+ PlatformUtils.PLATFORM_DOTNET + ", current platform is " + curPlatform);
+
+ return FactoryBuilder.factoryOf(platformCtx.createContinuousQueryFilter(filter));
+ }
+ default:
+ throw new IgniteClientException(ClientStatus.FAILED, "Unsupported filter platform: " + filterPlatform);
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousResponse.java
new file mode 100644
index 0000000..054fb6e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheQueryContinuousResponse.java
@@ -0,0 +1,59 @@
+/*
+ * 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.processors.platform.client.cache;
+
+import org.apache.ignite.internal.binary.BinaryRawWriterEx;
+import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
+import org.apache.ignite.internal.processors.platform.client.ClientResponse;
+
+/**
+ * Continuous query response.
+ */
+class ClientCacheQueryContinuousResponse extends ClientResponse {
+ /** */
+ private final ClientCacheQueryContinuousHandle handle;
+
+ /** */
+ private final long qryId;
+
+ /**
+ * Ctor.
+ * @param reqId Request id.
+ * @param handle Handle.
+ * @param qryId Continuous query handle id.
+ */
+ public ClientCacheQueryContinuousResponse(long reqId, ClientCacheQueryContinuousHandle handle, long qryId) {
+ super(reqId);
+ this.handle = handle;
+ this.qryId = qryId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void encode(ClientConnectionContext ctx, BinaryRawWriterEx writer) {
+ super.encode(ctx, writer);
+
+ writer.writeLong(qryId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onSent() {
+ super.onSent();
+
+ handle.startNotifications(qryId);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheScanQueryRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheScanQueryRequest.java
index b863560..dcb3030b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheScanQueryRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheScanQueryRequest.java
@@ -22,9 +22,11 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.processors.platform.PlatformContext;
import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
+import org.apache.ignite.internal.processors.platform.client.ClientPlatform;
import org.apache.ignite.internal.processors.platform.client.ClientResponse;
import org.apache.ignite.internal.processors.platform.client.tx.ClientTxAwareRequest;
import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
@@ -33,17 +35,8 @@ import org.apache.ignite.lang.IgniteBiPredicate;
/**
* Scan query request.
*/
-@SuppressWarnings("unchecked")
+@SuppressWarnings({"unchecked", "rawtypes"})
public class ClientCacheScanQueryRequest extends ClientCacheDataRequest implements ClientTxAwareRequest {
- /** Java filter. */
- private static final byte FILTER_PLATFORM_JAVA = 1;
-
- /** .NET filter. */
- private static final byte FILTER_PLATFORM_DOTNET = 2;
-
- /** C++ filter. */
- private static final byte FILTER_PLATFORM_CPP = 3;
-
/** Local flag. */
private final boolean loc;
@@ -81,13 +74,13 @@ public class ClientCacheScanQueryRequest extends ClientCacheDataRequest implemen
/** {@inheritDoc} */
@Override public ClientResponse process(ClientConnectionContext ctx) {
- IgniteCache cache = filterPlatform == FILTER_PLATFORM_JAVA && !isKeepBinary() ? rawCache(ctx) : cache(ctx);
+ IgniteCache cache = filterPlatform == ClientPlatform.JAVA && !isKeepBinary() ? rawCache(ctx) : cache(ctx);
ScanQuery qry = new ScanQuery()
.setLocal(loc)
.setPageSize(pageSize)
.setPartition(part)
- .setFilter(createFilter(ctx));
+ .setFilter(createFilter(ctx.kernalContext(), filterObj, filterPlatform));
ctx.incrementCursors();
@@ -115,16 +108,16 @@ public class ClientCacheScanQueryRequest extends ClientCacheDataRequest implemen
* @return Filter.
* @param ctx Context.
*/
- private IgniteBiPredicate createFilter(ClientConnectionContext ctx) {
+ public static IgniteBiPredicate createFilter(GridKernalContext ctx, Object filterObj, byte filterPlatform) {
if (filterObj == null)
return null;
switch (filterPlatform) {
- case FILTER_PLATFORM_JAVA:
+ case ClientPlatform.JAVA:
return ((BinaryObject)filterObj).deserialize();
- case FILTER_PLATFORM_DOTNET:
- PlatformContext platformCtx = ctx.kernalContext().platform().context();
+ case ClientPlatform.DOTNET:
+ PlatformContext platformCtx = ctx.platform().context();
String curPlatform = platformCtx.platform();
@@ -135,7 +128,7 @@ public class ClientCacheScanQueryRequest extends ClientCacheDataRequest implemen
return platformCtx.createCacheEntryFilter(filterObj, 0);
- case FILTER_PLATFORM_CPP:
+ case ClientPlatform.CPP:
default:
throw new UnsupportedOperationException("Invalid client ScanQuery filter code: " + filterPlatform);
}
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/PlatformCacheEntryEvenKeyEventFilter.java b/modules/core/src/test/java/org/apache/ignite/platform/PlatformCacheEntryEvenKeyEventFilter.java
new file mode 100644
index 0000000..a8711b4
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/platform/PlatformCacheEntryEvenKeyEventFilter.java
@@ -0,0 +1,32 @@
+/*
+ * 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.platform;
+
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+
+/**
+ * Test filter.
+ */
+public class PlatformCacheEntryEvenKeyEventFilter implements CacheEntryEventSerializableFilter<Integer, Object> {
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends Integer, ?> cacheEntryEvent) throws CacheEntryListenerException {
+ return cacheEntryEvent.getKey() % 2 == 0;
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj b/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj
index e745e90..5a6deb1 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj
@@ -211,6 +211,9 @@
<Link>Cache\Store\CacheTestStore.cs</Link>
</Compile>
<Compile Include="..\Apache.Ignite.Core.Tests\Cache\TestReferenceObject.cs" Link="Cache\TestReferenceObject.cs" />
+ <Compile Include="..\Apache.Ignite.Core.Tests\Client\Cache\ContinuousQueryTest.cs">
+ <Link>ThinClient\Cache\ContinuousQueryTest.cs</Link>
+ </Compile>
<Compile Include="..\Apache.Ignite.Core.Tests\Client\Cache\DynamicFieldSetTest.cs">
<Link>ThinClient\Cache\DynamicFieldSetTest.cs</Link>
</Compile>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
index 9779dda..3184d4c 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
@@ -144,6 +144,7 @@
<Compile Include="Cache\Query\Linq\CacheLinqTest.Contains.cs" />
<Compile Include="Cache\Store\CacheStoreSessionTestCodeConfig.cs" />
<Compile Include="Cache\Store\CacheStoreSessionTestSharedFactory.cs" />
+ <Compile Include="Client\Cache\ContinuousQueryTest.cs" />
<Compile Include="Client\Cache\DynamicFieldSetTest.cs" />
<Compile Include="Client\Cache\SerializableObjectsTest.cs" />
<Compile Include="Client\Cache\PartitionAwarenessTest.cs" />
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/Serializable/AdvancedSerializationTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/Serializable/AdvancedSerializationTest.cs
index e65effb..5665d71 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/Serializable/AdvancedSerializationTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/Serializable/AdvancedSerializationTest.cs
@@ -96,7 +96,7 @@ namespace Apache.Ignite.Core.Tests.Binary.Serializable
Assert.AreEqual(expectedRes, jobResult.InnerXml);
}
-#if !NETCOREAPP2_0 && !NETCOREAPP2_1 && !NETCOREAPP3_0 // AppDomains are not supported in .NET Core
+#if !NETCOREAPP// AppDomains are not supported in .NET Core
/// <summary>
/// Tests custom serialization binder.
/// </summary>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Query/Continuous/ContinuousQueryJavaFilterTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Query/Continuous/ContinuousQueryJavaFilterTest.cs
index b70f61d..a4f7e42 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Query/Continuous/ContinuousQueryJavaFilterTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Query/Continuous/ContinuousQueryJavaFilterTest.cs
@@ -55,7 +55,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Query.Continuous
/** */
private IIgnite _ignite;
-
+
/** */
private string _javaNodeName;
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheClientAsyncWrapper.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheClientAsyncWrapper.cs
index b4e01c7..c56986b 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheClientAsyncWrapper.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheClientAsyncWrapper.cs
@@ -23,6 +23,7 @@ namespace Apache.Ignite.Core.Tests.Client.Cache
using Apache.Ignite.Core.Cache.Expiry;
using Apache.Ignite.Core.Cache.Query;
using Apache.Ignite.Core.Client.Cache;
+ using Apache.Ignite.Core.Client.Cache.Query.Continuous;
/// <summary>
/// Cache client async wrapper.
@@ -357,5 +358,11 @@ namespace Apache.Ignite.Core.Tests.Client.Cache
{
return _cache.WithExpiryPolicy(plc);
}
+
+ /** <inheritDoc /> */
+ public IContinuousQueryHandleClient QueryContinuous(ContinuousQueryClient<TK, TV> continuousQuery)
+ {
+ throw new System.NotImplementedException();
+ }
}
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ContinuousQueryTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ContinuousQueryTest.cs
new file mode 100644
index 0000000..fad4191
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ContinuousQueryTest.cs
@@ -0,0 +1,802 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Core.Tests.Client.Cache
+{
+ using System;
+ using System.Collections.Concurrent;
+ using System.Collections.Generic;
+ using System.Linq;
+ using System.Threading;
+ using System.Threading.Tasks;
+ using Apache.Ignite.Core.Binary;
+ using Apache.Ignite.Core.Cache;
+ using Apache.Ignite.Core.Cache.Event;
+ using Apache.Ignite.Core.Cache.Query.Continuous;
+ using Apache.Ignite.Core.Client;
+ using Apache.Ignite.Core.Client.Cache;
+ using Apache.Ignite.Core.Client.Cache.Query.Continuous;
+ using Apache.Ignite.Core.Configuration;
+ using Apache.Ignite.Core.Interop;
+ using Apache.Ignite.Core.Log;
+ using Apache.Ignite.Core.Tests.Compute;
+ using NUnit.Framework;
+
+ /// <summary>
+ /// Tests for thin client continuous queries.
+ /// </summary>
+ public class ContinuousQueryTest : ClientTestBase
+ {
+ /** */
+ private const int MaxCursors = 20;
+
+ /// <summary>
+ /// Initializes a new instance of <see cref="ContinuousQueryTest"/>.
+ /// </summary>
+ public ContinuousQueryTest() : base(2)
+ {
+ // No-op.
+ }
+
+ /// <summary>
+ /// Executes after every test.
+ /// </summary>
+ [TearDown]
+ public void TestTearDown()
+ {
+ Assert.IsEmpty(Client.GetActiveNotificationListeners());
+ }
+
+ /// <summary>
+ /// Basic continuous query test.
+ ///
+ /// - Start the query
+ /// - Add cache entry, verify query event
+ /// - Update cache entry, verify query event
+ /// - Remove cache entry, verify query event
+ /// </summary>
+ [Test]
+ public void TestContinuousQueryCallsLocalListenerWithCorrectEvent()
+ {
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ var events = new List<ICacheEntryEvent<int, int>>();
+ var qry = new ContinuousQueryClient<int, int>(new DelegateListener<int, int>(events.Add));
+ using (cache.QueryContinuous(qry))
+ {
+ // Create.
+ cache.Put(1, 1);
+ TestUtils.WaitForTrueCondition(() => events.Count == 1);
+
+ var evt = events.Single();
+ Assert.AreEqual(CacheEntryEventType.Created, evt.EventType);
+ Assert.IsFalse(evt.HasOldValue);
+ Assert.IsTrue(evt.HasValue);
+ Assert.AreEqual(1, evt.Key);
+ Assert.AreEqual(1, evt.Value);
+
+ // Update.
+ cache.Put(1, 2);
+ TestUtils.WaitForTrueCondition(() => events.Count == 2);
+
+ evt = events.Last();
+ Assert.AreEqual(CacheEntryEventType.Updated, evt.EventType);
+ Assert.IsTrue(evt.HasOldValue);
+ Assert.IsTrue(evt.HasValue);
+ Assert.AreEqual(1, evt.Key);
+ Assert.AreEqual(2, evt.Value);
+ Assert.AreEqual(1, evt.OldValue);
+
+ // Remove.
+ cache.Remove(1);
+ TestUtils.WaitForTrueCondition(() => events.Count == 3);
+
+ evt = events.Last();
+ Assert.AreEqual(CacheEntryEventType.Removed, evt.EventType);
+ Assert.IsTrue(evt.HasOldValue);
+ Assert.IsTrue(evt.HasValue);
+ Assert.AreEqual(1, evt.Key);
+ Assert.AreEqual(2, evt.Value);
+ Assert.AreEqual(2, evt.OldValue);
+ }
+ }
+
+ /// <summary>
+ /// Tests that default query settings have correct values.
+ /// </summary>
+ [Test]
+ public void TestDefaultSettings()
+ {
+ var qry = new ContinuousQueryClient<int, int>();
+
+ Assert.AreEqual(ContinuousQueryClient.DefaultBufferSize, qry.BufferSize);
+ Assert.AreEqual(TimeSpan.Zero, qry.TimeInterval);
+ }
+
+ /// <summary>
+ /// Tests that Compute notifications and Continuous Query notifications work together correctly.
+ /// Compute and Continuous Queries use the same server -> client notification mechanism,
+ /// we ensure that there are no conflicts when both are used in parallel.
+ ///
+ /// - Start a new thread that runs Compute tasks in background
+ /// - Start two continuous queries, with and without filter, using same thin client connection
+ /// - Update cache and verify that continuous query listeners receive correct events
+ /// </summary>
+ [Test]
+ public void TestComputeWorksWhenContinuousQueryIsActive()
+ {
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ var receivedKeysAll = new ConcurrentBag<int>();
+ var receivedKeysOdd = new ConcurrentBag<int>();
+
+ var qry1 = new ContinuousQueryClient<int, int>
+ {
+ Listener = new DelegateListener<int, int>(e => receivedKeysOdd.Add(e.Key)),
+ Filter = new OddKeyFilter()
+ };
+
+ var qry2 = new ContinuousQueryClient<int, int>
+ {
+ Listener = new DelegateListener<int, int>(e => receivedKeysAll.Add(e.Key)),
+ };
+
+ var cts = new CancellationTokenSource();
+
+ var computeRunnerTask = Task.Factory.StartNew(() =>
+ {
+ while (!cts.IsCancellationRequested)
+ {
+ var res = Client.GetCompute().ExecuteJavaTask<int>(
+ ComputeApiTest.EchoTask, ComputeApiTest.EchoTypeInt);
+
+ Assert.AreEqual(1, res);
+ }
+ }, cts.Token);
+
+ var keys = Enumerable.Range(1, 10000).ToArray();
+
+ using (cache.QueryContinuous(qry1))
+ using (cache.QueryContinuous(qry2))
+ {
+ foreach (var key in keys)
+ {
+ cache[key] = key;
+ }
+ }
+
+ cts.Cancel();
+ computeRunnerTask.Wait();
+
+ TestUtils.WaitForTrueCondition(() => receivedKeysAll.Count == keys.Length);
+ Assert.AreEqual(keys.Length / 2, receivedKeysOdd.Count);
+
+ CollectionAssert.AreEquivalent(keys, receivedKeysAll);
+ CollectionAssert.AreEquivalent(keys.Where(k => k % 2 == 1), receivedKeysOdd);
+ }
+
+ /// <summary>
+ /// Tests that continuous query with filter receives only matching events.
+ ///
+ /// - Start a continuous query with filter
+ /// - Verify that filter receives all events
+ /// - Verify that listener receives filtered events
+ /// </summary>
+ [Test]
+ public void TestContinuousQueryWithFilterReceivesOnlyMatchingEvents()
+ {
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ ICacheEntryEvent<int, int> lastEvt = null;
+ var listener = new DelegateListener<int, int>(e => lastEvt = e);
+
+ var qry = new ContinuousQueryClient<int,int>(listener)
+ {
+ Filter = new OddKeyFilter()
+ };
+
+ using (cache.QueryContinuous(qry))
+ {
+ cache.Put(0, 0);
+ TestUtils.WaitForTrueCondition(() => OddKeyFilter.LastKey == 0);
+ Assert.IsNull(lastEvt);
+
+ cache.Put(5, 5);
+ TestUtils.WaitForTrueCondition(() => OddKeyFilter.LastKey == 5);
+ TestUtils.WaitForTrueCondition(() => lastEvt != null);
+ Assert.IsNotNull(lastEvt);
+ Assert.AreEqual(5, lastEvt.Key);
+
+ cache.Put(8, 8);
+ TestUtils.WaitForTrueCondition(() => OddKeyFilter.LastKey == 8);
+ Assert.AreEqual(5, lastEvt.Key);
+ }
+ }
+
+ /// <summary>
+ /// Tests that continuous query with a Java filter receives only matching events.
+ ///
+ /// - Start a continuous query with a Java filter
+ /// - Check that .NET listener receives filtered events
+ /// </summary>
+ [Test]
+ public void TestContinuousQueryWithJavaFilterReceivesOnlyMatchingEvents()
+ {
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ var evts = new ConcurrentBag<int>();
+ var listener = new DelegateListener<int, int>(e => evts.Add(e.Key));
+
+ var qry = new ContinuousQueryClient<int, int>(listener)
+ {
+ Filter = new JavaObject("org.apache.ignite.platform.PlatformCacheEntryEvenKeyEventFilter", null)
+ .ToCacheEntryEventFilter<int, int>()
+ };
+
+ using (cache.QueryContinuous(qry))
+ {
+ cache.PutAll(Enumerable.Range(1, 5).ToDictionary(x => x, x => x));
+ }
+
+ TestUtils.WaitForTrueCondition(() => evts.Count == 2);
+ CollectionAssert.AreEquivalent(new[] {2, 4}, evts);
+ }
+
+ /// <summary>
+ /// Tests that server-side updates are sent to the client.
+ ///
+ /// - Start a thin client continuous query
+ /// - Update cache from server node
+ /// - Check that thin client receives events
+ /// </summary>
+ [Test]
+ public void TestClientContinuousQueryReceivesEventsFromServerCache()
+ {
+ const int count = 10000;
+
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ var receiveCount = 0;
+
+ var listener = new DelegateListener<int, int>(e =>
+ {
+ Interlocked.Increment(ref receiveCount);
+ });
+
+ using (cache.QueryContinuous(new ContinuousQueryClient<int, int>(listener)))
+ {
+ var serverCache = Ignition.GetIgnite("1").GetCache<int, int>(cache.Name);
+
+ for (var i = 0; i < count; i++)
+ {
+ serverCache.Put(i, i);
+ }
+
+ TestUtils.WaitForTrueCondition(() => receiveCount == count);
+ }
+ }
+
+ /// <summary>
+ /// Tests that when cache is in binary mode (<see cref="ICacheClient{TK,TV}.WithKeepBinary{K1,V1}"/>,
+ /// continuous query listener and filter receive binary objects.
+ ///
+ /// - Get a cache in binary mode (WithKeepBinary)
+ /// - Start a continuous query
+ /// - Check that both filter and listener receive objects in binary form
+ /// </summary>
+ [Test]
+ public void TestContinuousQueryWithKeepBinaryPassesBinaryObjectsToListenerAndFilter()
+ {
+ var cache = Client.GetOrCreateCache<int, Person>(TestUtils.TestName);
+ var binCache = cache.WithKeepBinary<int, IBinaryObject>();
+
+ var evts = new ConcurrentBag<IBinaryObject>();
+
+ var qry = new ContinuousQueryClient<int, IBinaryObject>
+ {
+ Listener = new DelegateListener<int, IBinaryObject>(e => evts.Add(e.Value)),
+ Filter = new EvenIdBinaryFilter()
+ };
+
+ using (binCache.QueryContinuous(qry))
+ {
+ cache[2] = new Person(2);
+ cache[3] = new Person(3);
+
+ TestUtils.WaitForTrueCondition(() => !evts.IsEmpty);
+ }
+
+ // Listener should receive one event with an even id.
+ var binObj = evts.Single();
+ Assert.IsNotNull(binObj);
+ Assert.AreEqual(2, binObj.GetField<int>("Id"));
+ Assert.AreEqual("Person 2", binObj.GetField<string>("Name"));
+
+ // Filter has received both events, last one was filtered out.
+ binObj = EvenIdBinaryFilter.LastValue;
+ Assert.IsNotNull(binObj);
+ Assert.AreEqual(3, binObj.GetField<int>("Id"));
+ Assert.AreEqual("Person 3", binObj.GetField<string>("Name"));
+ }
+
+ /// <summary>
+ /// Tests that custom key / value objects can be used in Continuous Query filter and listener.
+ ///
+ /// - Create a cache with a custom class value (Person)
+ /// - Run a continuous query with filter and listener
+ /// </summary>
+ [Test]
+ public void TestContinuousQueryWithCustomObjects()
+ {
+ var cache = Client.GetOrCreateCache<int, Person>(TestUtils.TestName);
+
+ var evts = new ConcurrentBag<Person>();
+
+ var qry = new ContinuousQueryClient<int, Person>
+ {
+ Listener = new DelegateListener<int, Person>(e => evts.Add(e.Value)),
+ Filter = new EvenIdFilter()
+ };
+
+ using (cache.QueryContinuous(qry))
+ {
+ cache[2] = new Person(2);
+ cache[3] = new Person(3);
+
+ TestUtils.WaitForTrueCondition(() => !evts.IsEmpty);
+ }
+
+ // Listener should receive one event with an even id.
+ var obj = evts.Single();
+ Assert.IsNotNull(obj);
+ Assert.AreEqual(2, obj.Id);
+ Assert.AreEqual("Person 2", obj.Name);
+
+ // Filter has received both events, last one was filtered out.
+ obj = EvenIdFilter.LastValue;
+ Assert.IsNotNull(obj);
+ Assert.AreEqual(3, obj.Id);
+ Assert.AreEqual("Person 3", obj.Name);
+ }
+
+ /// <summary>
+ /// Tests that exception in continuous query remote filter is logged and event is delivered anyway.
+ ///
+ /// - Run a continuous query with a filter that throws an exception
+ /// - Verify that exception is logged
+ /// - Verify that the client receives an event
+ /// </summary>
+ [Test]
+ public void TestExceptionInFilterIsLoggedAndFilterIsIgnored()
+ {
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ var evts = new ConcurrentBag<int>();
+ var qry = new ContinuousQueryClient<int, int>(new DelegateListener<int, int>(e => evts.Add(e.Key)))
+ {
+ Filter = new ExceptionalFilter()
+ };
+
+ using (cache.QueryContinuous(qry))
+ {
+ cache[1] = 1;
+ }
+
+ Assert.AreEqual(1, cache[1]);
+
+ // Assert: error is logged.
+ var error = GetLoggers()
+ .SelectMany(x => x.Entries)
+ .Where(e => e.Level >= LogLevel.Warn)
+ .Select(e => e.Message)
+ .LastOrDefault();
+
+ Assert.AreEqual(
+ "CacheEntryEventFilter failed: javax.cache.event.CacheEntryListenerException: " +
+ ExceptionalFilter.Error,
+ error);
+
+ // Assert: continuous query event is delivered.
+ Assert.AreEqual(new[] {1}, evts);
+ }
+
+ /// <summary>
+ /// Tests that continuous query disconnected event (<see cref="IContinuousQueryHandleClient.Disconnected"/>)
+ /// is raised when thin client connection is lost.
+ ///
+ /// - Start a continuous query
+ /// - Disconnect the client
+ /// - Verify that the Disconnected event is raised
+ /// </summary>
+ [Test]
+ public void TestClientDisconnectRaisesDisconnectedEventOnQueryHandle()
+ {
+ ICacheEntryEvent<int, int> lastEvt = null;
+ var qry = new ContinuousQueryClient<int,int>(
+ new DelegateListener<int, int>(e => lastEvt = e));
+
+ var client = GetClient();
+ var cache = client.GetOrCreateCache<int, int>(TestUtils.TestName);
+ var handle = cache.QueryContinuous(qry);
+
+ ContinuousQueryClientDisconnectedEventArgs disconnectedEventArgs = null;
+
+ handle.Disconnected += (sender, args) =>
+ {
+ disconnectedEventArgs = args;
+
+ // ReSharper disable once AccessToDisposedClosure (disposed state does not matter)
+ Assert.AreEqual(handle, sender);
+ };
+
+ cache[1] = 1;
+ TestUtils.WaitForTrueCondition(() => lastEvt != null);
+
+ client.Dispose();
+
+ // Assert: disconnected event has been raised.
+ TestUtils.WaitForTrueCondition(() => disconnectedEventArgs != null);
+ Assert.IsNotNull(disconnectedEventArgs.Exception);
+
+ StringAssert.StartsWith("Cannot access a disposed object", disconnectedEventArgs.Exception.Message);
+
+ // Multiple dispose is allowed.
+ handle.Dispose();
+ handle.Dispose();
+ }
+
+ /// <summary>
+ /// Tests that continuous query disconnected event is not raised on a disposed handle.
+ ///
+ /// - Start a continuous query
+ /// - Subscribe to the Disconnected event
+ /// - Stop the continuous query
+ /// - Stop the client
+ /// - Check that the Disconnected event did not occur
+ /// </summary>
+ [Test]
+ public void TestClientDisconnectDoesNotRaiseDisconnectedEventOnDisposedQueryHandle()
+ {
+ var qry = new ContinuousQueryClient<int,int>(new DelegateListener<int, int>());
+
+ var client = GetClient();
+ var cache = client.GetOrCreateCache<int, int>(TestUtils.TestName);
+ ContinuousQueryClientDisconnectedEventArgs disconnectedEventArgs = null;
+
+ using (var handle = cache.QueryContinuous(qry))
+ {
+ handle.Disconnected += (sender, args) => disconnectedEventArgs = args;
+ }
+
+ client.Dispose();
+
+ // Assert: disconnected event has NOT been raised.
+ Assert.IsNull(disconnectedEventArgs);
+ }
+
+ /// <summary>
+ /// Tests that client does not receive updates for a stopped continuous query.
+ ///
+ /// - Start a continuous query with the TimeInterval set to 1 second and the BufferSize to 10
+ /// - Put 1 entry to the cache
+ /// - Stop the continuous query
+ /// - Check that the client does not receive any continuous query events after the query has been stopped
+ /// </summary>
+ [Test]
+ public void TestDisposedQueryHandleDoesNotReceiveUpdates()
+ {
+ // Stop the query before the batch is sent out by time interval.
+ var interval = TimeSpan.FromSeconds(1);
+ TestBatches(1, 10, interval, (keys, res) => { });
+ Thread.Sleep(interval);
+
+ // Check that socket has no dangling notifications.
+ Assert.IsEmpty(Client.GetActiveNotificationListeners());
+ }
+
+ /// <summary>
+ /// Tests that <see cref="ClientConnectorConfiguration.MaxOpenCursorsPerConnection"/> controls
+ /// maximum continuous query count.
+ ///
+ /// - Set MaxOpenCursorsPerConnection
+ /// - Try to start more queries than that
+ /// - Check that correct exception is returned
+ /// </summary>
+ [Test]
+ public void TestContinuousQueryCountIsLimitedByMaxCursors()
+ {
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ var qry = new ContinuousQueryClient<int, int>( new DelegateListener<int, int>());
+
+ var queries = Enumerable.Range(1, MaxCursors).Select(_ => cache.QueryContinuous(qry)).ToList();
+
+ var ex = Assert.Throws<IgniteClientException>(() => cache.QueryContinuous(qry));
+
+ queries.ForEach(q => q.Dispose());
+
+ StringAssert.StartsWith("Too many open cursors", ex.Message);
+ Assert.AreEqual(ClientStatusCode.TooManyCursors, ex.StatusCode);
+ }
+
+ /// <summary>
+ /// Tests that multiple queries can be started from multiple threads.
+ ///
+ /// - Run multiple continuous queries from different threads using the same thin client connection
+ /// - Versify that results are correct
+ /// </summary>
+ [Test]
+ public void TestMultipleQueriesMultithreaded()
+ {
+ var cache = Ignition.GetIgnite().GetOrCreateCache<int, int>(TestUtils.TestName);
+ cache.Put(1, 0);
+
+ var cts = new CancellationTokenSource();
+ var updaterThread = Task.Factory.StartNew(() =>
+ {
+ for (long i = 0; i < long.MaxValue && !cts.IsCancellationRequested; i++)
+ {
+ cache.Put(1, (int) i);
+ }
+ });
+
+ var clientCache = Client.GetCache<int, int>(cache.Name);
+
+ TestUtils.RunMultiThreaded(() =>
+ {
+ var evt = new ManualResetEventSlim();
+
+ var qry = new ContinuousQueryClient<int, int>
+ {
+ Listener = new DelegateListener<int, int>(e =>
+ {
+ Assert.AreEqual(1, e.Key);
+ Assert.AreEqual(CacheEntryEventType.Updated, e.EventType);
+ evt.Set();
+ })
+ };
+
+ for (var i = 0; i < 200; i++)
+ {
+ evt.Reset();
+ using (clientCache.QueryContinuous(qry))
+ {
+ evt.Wait();
+ }
+
+ }
+ }, 16);
+
+ cts.Cancel();
+ updaterThread.Wait();
+ }
+
+ /// <summary>
+ /// Tests that Listener presence is validated before starting the continuous query.
+ /// </summary>
+ [Test]
+ public void TestContinuousQueryWithoutListenerThrowsException()
+ {
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+
+ var ex = Assert.Throws<ArgumentNullException>(() => cache.QueryContinuous(new ContinuousQueryClient<int, int>()));
+ Assert.AreEqual("continuousQuery.Listener", ex.ParamName);
+ }
+
+ /// <summary>
+ /// Tests that when custom <see cref="ContinuousQueryClient{TK,TV}.BufferSize"/> is set,
+ /// events are sent in batches, not 1 by 1.
+ ///
+ /// - Start a continuous query with the BufferSize set to 3
+ /// - Put 8 entries to the cache
+ /// - Check that 2 batches of 3 entries has been received, and last 2 keys has not been received.
+ /// </summary>
+ [Test]
+ public void TestCustomBufferSizeResultsInBatchedUpdates()
+ {
+ TestBatches(8, 3, TimeSpan.Zero, (keys, res) =>
+ {
+ TestUtils.WaitForTrueCondition(() => res.Count == 2, () => res.Count.ToString());
+
+ var resOrdered = res.OrderBy(x => x.FirstOrDefault()).ToList();
+ CollectionAssert.AreEquivalent(keys.Take(3), resOrdered.First());
+ CollectionAssert.AreEquivalent(keys.Skip(3).Take(3), resOrdered.Last());
+ });
+ }
+
+ /// <summary>
+ /// Tests that when custom <see cref="ContinuousQueryClient{TK,TV}.TimeInterval"/> is set,
+ /// and <see cref="ContinuousQueryClient{TK,TV}.BufferSize"/> is greater than 1,
+ /// batches are sent out before buffer is full when the time interval passes.
+ ///
+ /// - Start a continuous query with the BufferSize set to 4 and TimeInterval set to 1 second
+ /// - Put 2 entries to the cache
+ /// - Wait and check that clint has received one batch of 2 entries
+ /// </summary>
+ [Test]
+ public void TestCustomTimeIntervalCausesIncompleteBatches()
+ {
+ TestBatches(2, 4, TimeSpan.FromSeconds(1), (keys, res) =>
+ {
+ TestUtils.WaitForTrueCondition(() => res.Count == 1, () => res.Count.ToString(), 2000);
+
+ CollectionAssert.AreEquivalent(keys.Take(2), res.Single());
+ });
+ }
+
+ /// <summary>
+ /// Tests batching behavior.
+ /// </summary>
+ private void TestBatches(int keyCount, int bufferSize, TimeSpan interval,
+ Action<List<int>, ConcurrentQueue<List<int>>> assert)
+ {
+ var res = new ConcurrentQueue<List<int>>();
+
+ var qry = new ContinuousQueryClient<int, int>
+ {
+ Listener = new DelegateBatchListener<int, int>(evts =>
+ res.Enqueue(evts.Select(e => e.Key).ToList())),
+ BufferSize = bufferSize,
+ TimeInterval = interval
+ };
+
+ var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
+ var server = Ignition.GetIgnite("1");
+ var serverCache = server.GetCache<int, int>(cache.Name);
+
+ // Use primary keys for "remote" node to ensure batching.
+ // Client is connected to another server node, so it will receive batches as expected.
+ var keys = TestUtils.GetPrimaryKeys(server, cache.Name).Take(keyCount).ToList();
+
+ using (cache.QueryContinuous(qry))
+ {
+ keys.ForEach(k => serverCache.Put(k, k));
+
+ assert(keys, res);
+ }
+ }
+
+ /** <inheritdoc /> */
+ protected override IgniteConfiguration GetIgniteConfiguration()
+ {
+ return new IgniteConfiguration(base.GetIgniteConfiguration())
+ {
+ ClientConnectorConfiguration = new ClientConnectorConfiguration
+ {
+ MaxOpenCursorsPerConnection = MaxCursors,
+ ThinClientConfiguration = new ThinClientConfiguration
+ {
+ MaxActiveComputeTasksPerConnection = 100,
+ }
+ }
+ };
+ }
+
+ /** */
+ private class DelegateListener<TK, TV> : ICacheEntryEventListener<TK, TV>
+ {
+ /** */
+ private readonly Action<ICacheEntryEvent<TK, TV>> _action;
+
+ /** */
+ public DelegateListener(Action<ICacheEntryEvent<TK, TV>> action = null)
+ {
+ _action = action ?? (_ => {});
+ }
+
+ /** <inheritdoc /> */
+ public void OnEvent(IEnumerable<ICacheEntryEvent<TK, TV>> evts)
+ {
+ foreach (var evt in evts)
+ {
+ _action(evt);
+ }
+ }
+ }
+
+ /** */
+ private class DelegateBatchListener<TK, TV> : ICacheEntryEventListener<TK, TV>
+ {
+ /** */
+ private readonly Action<IEnumerable<ICacheEntryEvent<TK, TV>>> _action;
+
+ /** */
+ public DelegateBatchListener(Action<IEnumerable<ICacheEntryEvent<TK, TV>>> action = null)
+ {
+ _action = action ?? (_ => {});
+ }
+
+ /** <inheritdoc /> */
+ public void OnEvent(IEnumerable<ICacheEntryEvent<TK, TV>> evts)
+ {
+ _action(evts);
+ }
+ }
+
+ /** */
+ private class OddKeyFilter : ICacheEntryEventFilter<int, int>, ICacheEntryFilter<int, int>
+ {
+ /** */
+ public static int LastKey;
+
+ /** <inheritdoc /> */
+ public bool Evaluate(ICacheEntryEvent<int, int> evt)
+ {
+ LastKey = evt.Key;
+
+ return evt.Key % 2 == 1;
+ }
+
+ /** <inheritdoc /> */
+ public bool Invoke(ICacheEntry<int, int> entry)
+ {
+ LastKey = entry.Key;
+
+ return entry.Key % 2 == 1;
+ }
+ }
+
+ /** */
+ private class ExceptionalFilter : ICacheEntryEventFilter<int, int>, ICacheEntryFilter<int, int>
+ {
+ /** */
+ public const string Error = "Foo failed because of bar";
+
+ /** <inheritdoc /> */
+ public bool Evaluate(ICacheEntryEvent<int, int> evt)
+ {
+ throw new Exception(Error);
+ }
+
+ /** <inheritdoc /> */
+ public bool Invoke(ICacheEntry<int, int> entry)
+ {
+ throw new Exception(Error);
+ }
+ }
+
+ /** */
+ private class EvenIdFilter : ICacheEntryEventFilter<int, Person>
+ {
+ /** */
+ public static Person LastValue { get; set; }
+
+ /** <inheritdoc /> */
+ public bool Evaluate(ICacheEntryEvent<int, Person> evt)
+ {
+ LastValue = evt.Value;
+
+ return evt.Value.Id % 2 == 0;
+ }
+ }
+
+ /** */
+ private class EvenIdBinaryFilter : ICacheEntryEventFilter<int, IBinaryObject>
+ {
+ /** */
+ public static IBinaryObject LastValue { get; set; }
+
+ /** <inheritdoc /> */
+ public bool Evaluate(ICacheEntryEvent<int, IBinaryObject> evt)
+ {
+ LastValue = evt.Value;
+
+ return evt.Value.GetField<int>("Id") % 2 == 0;
+ }
+ }
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
index 46f9770..f4e479e 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
@@ -18,7 +18,6 @@
namespace Apache.Ignite.Core.Tests.Client
{
using System;
- using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientServerCacheAdapter.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientServerCacheAdapter.cs
index 7e101dd..ae5c1fa 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientServerCacheAdapter.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientServerCacheAdapter.cs
@@ -25,6 +25,7 @@ namespace Apache.Ignite.Core.Tests.Client
using Apache.Ignite.Core.Cache.Expiry;
using Apache.Ignite.Core.Cache.Query;
using Apache.Ignite.Core.Client.Cache;
+ using Apache.Ignite.Core.Client.Cache.Query.Continuous;
/// <summary>
/// Adapts <see cref="ICache{TK,TV}"/> to <see cref="ICacheClient{TK,TV}"/>
@@ -298,5 +299,10 @@ namespace Apache.Ignite.Core.Tests.Client
{
throw new NotSupportedException();
}
+
+ public IContinuousQueryHandleClient QueryContinuous(ContinuousQueryClient<TK, TV> continuousQuery)
+ {
+ throw new NotImplementedException();
+ }
}
-}
\ No newline at end of file
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/IgniteClientConfigurationTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/IgniteClientConfigurationTest.cs
index 17830d8..001b075 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/IgniteClientConfigurationTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/IgniteClientConfigurationTest.cs
@@ -138,7 +138,7 @@ namespace Apache.Ignite.Core.Tests.Client
Assert.AreEqual("<?xml version=\"1.0\" encoding=\"utf-16\"?>" + Environment.NewLine +
"<igniteClientConfiguration " +
"xmlns=\"http://ignite.apache.org/schema/dotnet/IgniteClientConfigurationSection\">" +
- Environment.NewLine + " <logger type=\"null\" />" + Environment.NewLine +
+ Environment.NewLine + " <logger type=\"null\" />" + Environment.NewLine +
"</igniteClientConfiguration>",
emptyConfig.ToXml());
@@ -153,7 +153,7 @@ namespace Apache.Ignite.Core.Tests.Client
Assert.AreEqual("<?xml version=\"1.0\" encoding=\"utf-16\"?>" + Environment.NewLine +
"<igniteClientConfiguration host=\"myHost\" port=\"123\" " +
"xmlns=\"http://ignite.apache.org/schema/dotnet/IgniteClientConfigurationSection\">" +
- Environment.NewLine + " <logger type=\"null\" />" + Environment.NewLine +
+ Environment.NewLine + " <logger type=\"null\" />" + Environment.NewLine +
"</igniteClientConfiguration>",
cfg.ToXml());
@@ -188,7 +188,7 @@ namespace Apache.Ignite.Core.Tests.Client
public void TestDefaultLoggerWritesToConsole()
{
IgniteClientConfiguration cfg = null;
-
+
TestConsoleLogging(c => { cfg = c;}, (client, log) =>
{
Assert.AreSame(cfg.Logger, client.GetConfiguration().Logger);
@@ -291,7 +291,7 @@ namespace Apache.Ignite.Core.Tests.Client
}
}
-#if !NETCOREAPP2_0 && !NETCOREAPP2_1 && !NETCOREAPP3_0
+#if !NETCOREAPP
/// <summary>
/// Tests the schema validation.
/// </summary>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteLockTests.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteLockTests.cs
index 0436952..f66e543 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteLockTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteLockTests.cs
@@ -194,7 +194,6 @@ namespace Apache.Ignite.Core.Tests
/// Tests that removed lock throws correct exception.
/// </summary>
[Test]
- [Ignore("IGNITE-13128")]
public void TestRemovedBeforeUseLockThrowsIgniteException()
{
var lock1 = Ignite.GetOrCreateLock(TestUtils.TestName);
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs
index aa8f8ea..b055617 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs
@@ -1123,7 +1123,7 @@ namespace Apache.Ignite.Core.Tests.Services
/// </summary>
private IgniteConfiguration GetConfiguration(string springConfigUrl)
{
-#if !NETCOREAPP2_0 && !NETCOREAPP2_1 && !NETCOREAPP3_0
+#if !NETCOREAPP
if (!CompactFooter)
{
springConfigUrl = Compute.ComputeApiTestFullFooter.ReplaceFooterSetting(springConfigUrl);
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs
index f8b14d2..e244000 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs
@@ -322,8 +322,24 @@ namespace Apache.Ignite.Core.Tests
/// <param name="message">Assertion message.</param>
public static void WaitForTrueCondition(Func<bool> cond, int timeout = 1000, string message = null)
{
+ WaitForTrueCondition(cond, message == null ? (Func<string>) null : () => message, timeout);
+ }
+
+ /// <summary>
+ /// Waits for condition, polling in a busy wait loop, then asserts that condition is true.
+ /// </summary>
+ /// <param name="cond">Condition.</param>
+ /// <param name="messageFunc">Assertion message func.</param>
+ /// <param name="timeout">Timeout, in milliseconds.</param>
+ public static void WaitForTrueCondition(Func<bool> cond, Func<string> messageFunc, int timeout = 1000)
+ {
var res = WaitForCondition(cond, timeout);
- message = message ?? string.Format("Condition not reached within {0} ms", timeout);
+ var message = string.Format("Condition not reached within {0} ms", timeout);
+
+ if (messageFunc != null)
+ {
+ message += string.Format(" ({0})", messageFunc());
+ }
Assert.IsTrue(res, message);
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
index 96d17d7..c45a7f8 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
@@ -59,6 +59,9 @@
<Compile Include="Cache\Query\Continuous\IContinuousQueryHandleFields.cs" />
<Compile Include="Cache\Query\IQueryCursorField.cs" />
<Compile Include="Client\Cache\CacheClientConfiguration.cs" />
+ <Compile Include="Client\Cache\Query\Continuous\ContinuousQueryClient.cs" />
+ <Compile Include="Client\Cache\Query\Continuous\ContinuousQueryDisconnectedEventArgs.cs" />
+ <Compile Include="Client\Cache\Query\Continuous\IContinuousQueryHandleClient.cs" />
<Compile Include="Client\Compute\IComputeClient.cs" />
<Compile Include="Client\IClientCluster.cs" />
<Compile Include="Client\IClientClusterGroup.cs" />
@@ -100,6 +103,7 @@
<Compile Include="Impl\Cache\Query\IQueryBaseInternal.cs" />
<Compile Include="Impl\Cache\Query\PlatformCacheQueryCursor.cs" />
<Compile Include="Impl\Cache\Query\QueryCursorField.cs" />
+ <Compile Include="Impl\Client\Cache\Query\ClientContinuousQueryHandle.cs" />
<Compile Include="Impl\Client\ClientBitmaskFeature.cs" />
<Compile Include="Impl\Client\ClientConnection.cs" />
<Compile Include="Impl\Client\ClientContextBase.cs" />
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Cache/Query/Continuous/ContinuousQuery.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Cache/Query/Continuous/ContinuousQuery.cs
index 81913c8..4c471dc 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Cache/Query/Continuous/ContinuousQuery.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Cache/Query/Continuous/ContinuousQuery.cs
@@ -52,16 +52,16 @@ namespace Apache.Ignite.Core.Cache.Query.Continuous
/// <summary>
/// API for configuring continuous cache queries.
/// <para />
- /// Continuous queries allow to register a remote and a listener for cache update events.
- /// If an update event passes the filter, it will be sent to the node that executed the
+ /// Continuous queries allow to register a remote and a listener for cache update events.
+ /// If an update event passes the filter, it will be sent to the node that executed the
/// query and listener will be notified on that node.
/// <para />
/// Continuous query can either be executed on the whole topology or only on local node.
/// <para />
- /// In case query is distributed and a new node joins, it will get the filter for the query
+ /// In case query is distributed and a new node joins, it will get the filter for the query
/// during discovery process before it actually joins topology, so no updates will be missed.
/// <para />
- /// To execute the query use method
+ /// To execute the query use method
/// <see cref="ICache{K,V}.QueryContinuous(ContinuousQuery{K,V})"/>.
/// </summary>
public class ContinuousQuery<TK, TV> : ContinuousQuery
@@ -114,7 +114,7 @@ namespace Apache.Ignite.Core.Cache.Query.Continuous
}
/// <summary>
- /// Cache entry event listener. Invoked on the node where continuous query execution
+ /// Cache entry event listener. Invoked on the node where continuous query execution
/// has been started.
/// </summary>
public ICacheEntryEventListener<TK, TV> Listener { get; set; }
@@ -129,8 +129,8 @@ namespace Apache.Ignite.Core.Cache.Query.Continuous
public ICacheEntryEventFilter<TK, TV> Filter { get; set; }
/// <summary>
- /// Buffer size. When a cache update happens, entry is first put into a buffer.
- /// Entries from buffer will be sent to the master node only if the buffer is
+ /// Buffer size. When a cache update happens, entry is first put into a buffer.
+ /// Entries from buffer will be sent to the master node only if the buffer is
/// full or time provided via <see cref="TimeInterval"/> is exceeded.
/// <para />
/// Defaults to <see cref="ContinuousQuery.DefaultBufferSize"/>
@@ -138,21 +138,21 @@ namespace Apache.Ignite.Core.Cache.Query.Continuous
public int BufferSize { get; set; }
/// <summary>
- /// Time interval. When a cache update happens, entry is first put into a buffer.
- /// Entries from buffer will be sent to the master node only if the buffer is full
- /// (its size can be provided via <see cref="BufferSize"/> property) or time provided
+ /// Time interval. When a cache update happens, entry is first put into a buffer.
+ /// Entries from buffer will be sent to the master node only if the buffer is full
+ /// (its size can be provided via <see cref="BufferSize"/> property) or time provided
/// via this method is exceeded.
/// <para />
- /// Defaults to <c>0</c> which means that time check is disabled and entries will be
+ /// Defaults to <c>0</c> which means that time check is disabled and entries will be
/// sent only when buffer is full.
/// </summary>
public TimeSpan TimeInterval { get; set; }
/// <summary>
- /// Automatic unsubscribe flag. This flag indicates that query filters on remote nodes
- /// should be automatically unregistered if master node (node that initiated the query)
- /// leaves topology. If this flag is <c>false</c>, filters will be unregistered only
- /// when the query is cancelled from master node, and won't ever be unregistered if
+ /// Automatic unsubscribe flag. This flag indicates that query filters on remote nodes
+ /// should be automatically unregistered if master node (node that initiated the query)
+ /// leaves topology. If this flag is <c>false</c>, filters will be unregistered only
+ /// when the query is cancelled from master node, and won't ever be unregistered if
/// master node leaves grid.
/// <para />
/// Defaults to <c>true</c>.
@@ -160,7 +160,7 @@ namespace Apache.Ignite.Core.Cache.Query.Continuous
public bool AutoUnsubscribe { get; set; }
/// <summary>
- /// Local flag. When set query will be executed only on local node, so only local
+ /// Local flag. When set query will be executed only on local node, so only local
/// entries will be returned as query result.
/// <para />
/// Defaults to <c>false</c>.
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/ICacheClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/ICacheClient.cs
index 65b192b..e818a31 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/ICacheClient.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/ICacheClient.cs
@@ -24,6 +24,7 @@ namespace Apache.Ignite.Core.Client.Cache
using Apache.Ignite.Core.Cache;
using Apache.Ignite.Core.Cache.Expiry;
using Apache.Ignite.Core.Cache.Query;
+ using Apache.Ignite.Core.Client.Cache.Query.Continuous;
/// <summary>
/// Client cache API. See <see cref="IIgniteClient.GetCache{K, V}"/>.
@@ -435,5 +436,12 @@ namespace Apache.Ignite.Core.Client.Cache
/// <param name="plc">Expiry policy to use.</param>
/// <returns>Cache instance with the specified expiry policy set.</returns>
ICacheClient<TK, TV> WithExpiryPolicy(IExpiryPolicy plc);
+
+ /// <summary>
+ /// Starts the continuous query execution.
+ /// </summary>
+ /// <param name="continuousQuery">Continuous query.</param>
+ /// <returns>Handle to stop the query execution.</returns>
+ IContinuousQueryHandleClient QueryContinuous(ContinuousQueryClient<TK, TV> continuousQuery);
}
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/ContinuousQueryClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/ContinuousQueryClient.cs
new file mode 100644
index 0000000..6d5d6f3
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/ContinuousQueryClient.cs
@@ -0,0 +1,101 @@
+/*
+ * 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.
+ */
+namespace Apache.Ignite.Core.Client.Cache.Query.Continuous
+{
+ using System;
+ using Apache.Ignite.Core.Cache.Event;
+ using Apache.Ignite.Core.Cache.Query.Continuous;
+ using Apache.Ignite.Core.Impl.Common;
+ using Apache.Ignite.Core.Interop;
+
+ /// <summary>
+ /// Thin client continuous query base class, see <see cref="ContinuousQueryClient{K, V}"/>.
+ /// </summary>
+ public abstract class ContinuousQueryClient
+ {
+ /// <summary>
+ /// Default buffer size.
+ /// </summary>
+ public const int DefaultBufferSize = 1;
+ }
+
+ /// <summary>
+ /// API for configuring continuous cache queries in thin client.
+ /// <para />
+ /// Continuous queries allow to register a remote filter and a listener for cache update events.
+ /// If an update event passes the filter, it will be sent to the client.
+ /// <para />
+ /// To execute the query use method
+ /// <see cref="ICacheClient{K,V}.QueryContinuous"/>.
+ /// </summary>
+ public class ContinuousQueryClient<TK, TV> : ContinuousQueryClient
+ {
+ /// <summary>
+ /// Initializes a new instance of <see cref="ContinuousQueryClient{TK,TV}"/> class.
+ /// </summary>
+ public ContinuousQueryClient()
+ {
+ BufferSize = DefaultBufferSize;
+ }
+
+ /// <summary>
+ /// Initializes a new instance of <see cref="ContinuousQueryClient{TK,TV}"/> class with a specified listener.
+ /// </summary>
+ public ContinuousQueryClient(ICacheEntryEventListener<TK, TV> listener) : this()
+ {
+ IgniteArgumentCheck.NotNull(listener, "listener");
+
+ Listener = listener;
+ }
+
+ /// <summary>
+ /// Cache entry event listener. Invoked locally.
+ /// </summary>
+ public ICacheEntryEventListener<TK, TV> Listener { get; set; }
+
+ /// <summary>
+ /// Optional cache entry filter. Invoked on a node where cache event occurred. If the filter
+ /// returns <c>false</c>, then cache entry event will not be sent to a node where the
+ /// continuous query has been started.
+ /// <para />
+ /// This filter will be serialized and sent to the server nodes. .NET filters require all
+ /// server nodes to be .NET-based. Java filters can be used with <see cref="JavaObject"/>
+ /// and <see cref="ContinuousQueryExtensions.ToCacheEntryEventFilter{K, V}"/>.
+ /// </summary>
+ public ICacheEntryEventFilter<TK, TV> Filter { get; set; }
+
+ /// <summary>
+ /// Buffer size. When a cache update happens, entry is first put into a buffer.
+ /// Entries from buffer will be sent to the master node only if the buffer is
+ /// full or time provided via <see cref="TimeInterval"/> is exceeded.
+ /// <para />
+ /// Defaults to <see cref="ContinuousQuery.DefaultBufferSize"/>
+ /// </summary>
+ public int BufferSize { get; set; }
+
+ /// <summary>
+ /// Time interval. When a cache update happens, entry is first put into a buffer.
+ /// Entries from buffer will be sent to the master node only if the buffer is full
+ /// (its size can be provided via <see cref="BufferSize"/> property) or time provided
+ /// via this method is exceeded.
+ /// <para />
+ /// Defaults to <c>0</c> which means that time check is disabled and entries will be
+ /// sent only when buffer is full.
+ /// </summary>
+ public TimeSpan TimeInterval { get; set; }
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/ContinuousQueryDisconnectedEventArgs.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/ContinuousQueryDisconnectedEventArgs.cs
new file mode 100644
index 0000000..ff55b1b
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/ContinuousQueryDisconnectedEventArgs.cs
@@ -0,0 +1,47 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Core.Client.Cache.Query.Continuous
+{
+ using System;
+
+ /// <summary>
+ /// Provides data for the <see cref="IContinuousQueryHandleClient.Disconnected"/> event.
+ /// </summary>
+ public class ContinuousQueryClientDisconnectedEventArgs : EventArgs
+ {
+ /** */
+ private readonly Exception _exception;
+
+ /// <summary>
+ /// Initializes a new instance of <see cref="ContinuousQueryClientDisconnectedEventArgs"/> class.
+ /// </summary>
+ /// <param name="exception">Exception that caused the disconnect.</param>
+ public ContinuousQueryClientDisconnectedEventArgs(Exception exception)
+ {
+ _exception = exception;
+ }
+
+ /// <summary>
+ /// Gets the exception that caused the disconnect.
+ /// </summary>
+ public Exception Exception
+ {
+ get { return _exception; }
+ }
+ }
+}
\ No newline at end of file
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/IContinuousQueryHandleClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/IContinuousQueryHandleClient.cs
new file mode 100644
index 0000000..8ccc9bd
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Client/Cache/Query/Continuous/IContinuousQueryHandleClient.cs
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Core.Client.Cache.Query.Continuous
+{
+ using System;
+
+ /// <summary>
+ /// Represents a continuous query handle. Call <see cref="IDisposable.Dispose"/> to stop the continuous query.
+ /// </summary>
+ public interface IContinuousQueryHandleClient : IDisposable
+ {
+ /// <summary>
+ /// Occurs when continuous query gets disconnected.
+ /// Disconnect can be caused by a server node failure or a network failure.
+ /// </summary>
+ event EventHandler<ContinuousQueryClientDisconnectedEventArgs> Disconnected;
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursorBase.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursorBase.cs
index b65ae11..058027c 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursorBase.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursorBase.cs
@@ -41,7 +41,7 @@ namespace Apache.Ignite.Core.Impl.Cache.Query
/** Read func. */
private readonly Func<BinaryReader, T> _readFunc;
-
+
/** Lock object. */
private readonly object _syncRoot = new object();
@@ -70,7 +70,7 @@ namespace Apache.Ignite.Core.Impl.Cache.Query
/// <param name="keepBinary">Keep binary flag.</param>
/// <param name="readFunc">The read function.</param>
/// <param name="initialBatchStream">Optional stream with initial batch.</param>
- protected QueryCursorBase(Marshaller marsh, bool keepBinary, Func<BinaryReader, T> readFunc,
+ protected QueryCursorBase(Marshaller marsh, bool keepBinary, Func<BinaryReader, T> readFunc,
IBinaryStream initialBatchStream = null)
{
Debug.Assert(marsh != null);
@@ -220,7 +220,7 @@ namespace Apache.Ignite.Core.Impl.Cache.Query
lock (_syncRoot)
{
ThrowIfDisposed();
-
+
_batch = _hasNext ? GetBatch() : null;
_batchPos = 0;
@@ -293,6 +293,7 @@ namespace Apache.Ignite.Core.Impl.Cache.Query
return;
}
+ // When _hasNext is false, cursor is already disposed by the server.
if (_hasNext)
{
Dispose(true);
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
index b741afe..f240922 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
@@ -26,17 +26,22 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
using Apache.Ignite.Core.Binary;
using Apache.Ignite.Core.Cache;
using Apache.Ignite.Core.Cache.Configuration;
+ using Apache.Ignite.Core.Cache.Event;
using Apache.Ignite.Core.Cache.Expiry;
using Apache.Ignite.Core.Cache.Query;
using Apache.Ignite.Core.Client;
using Apache.Ignite.Core.Client.Cache;
+ using Apache.Ignite.Core.Client.Cache.Query.Continuous;
using Apache.Ignite.Core.Impl.Binary;
using Apache.Ignite.Core.Impl.Binary.IO;
using Apache.Ignite.Core.Impl.Cache;
using Apache.Ignite.Core.Impl.Cache.Expiry;
+ using Apache.Ignite.Core.Impl.Cache.Query.Continuous;
using Apache.Ignite.Core.Impl.Client;
using Apache.Ignite.Core.Impl.Client.Cache.Query;
using Apache.Ignite.Core.Impl.Common;
+ using Apache.Ignite.Core.Impl.Log;
+ using Apache.Ignite.Core.Log;
using BinaryWriter = Apache.Ignite.Core.Impl.Binary.BinaryWriter;
/// <summary>
@@ -75,7 +80,10 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
WithExpiryPolicy = 1 << 2
}
- /** Scan query filter platform code: .NET filter. */
+ /** Query filter platform code: Java filter. */
+ private const byte FilterPlatformJava = 1;
+
+ /** Query filter platform code: .NET filter. */
private const byte FilterPlatformDotnet = 2;
/** Cache name. */
@@ -96,6 +104,9 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
/** Expiry policy. */
private readonly IExpiryPolicy _expiryPolicy;
+ /** Logger. Lazily initialized. */
+ private ILogger _logger;
+
/// <summary>
/// Initializes a new instance of the <see cref="CacheClient{TK, TV}" /> class.
/// </summary>
@@ -602,6 +613,15 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
}
/** <inheritDoc /> */
+ public IContinuousQueryHandleClient QueryContinuous(ContinuousQueryClient<TK, TV> continuousQuery)
+ {
+ IgniteArgumentCheck.NotNull(continuousQuery, "continuousQuery");
+ IgniteArgumentCheck.NotNull(continuousQuery.Listener, "continuousQuery.Listener");
+
+ return QueryContinuousInternal(continuousQuery);
+ }
+
+ /** <inheritDoc /> */
[ExcludeFromCodeCoverage]
public T DoOutInOpExtension<T>(int extensionId, int opCode, Action<IBinaryRawWriter> writeAction,
Func<IBinaryRawReader, T> readFunc)
@@ -881,7 +901,6 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
writer.WriteBoolean(qry.Lazy);
writer.WriteTimeSpanAsLong(qry.Timeout);
writer.WriteBoolean(includeColumns);
-
}
/// <summary>
@@ -978,5 +997,125 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
return res;
}
+
+ /// <summary>
+ /// Starts the continuous query.
+ /// </summary>
+ private ClientContinuousQueryHandle QueryContinuousInternal(
+ ContinuousQueryClient<TK, TV> continuousQuery)
+ {
+ Debug.Assert(continuousQuery != null);
+ Debug.Assert(continuousQuery.Listener != null);
+
+ var listener = continuousQuery.Listener;
+
+ return DoOutInOp(
+ ClientOp.QueryContinuous,
+ ctx => WriteContinuousQuery(ctx, continuousQuery),
+ ctx =>
+ {
+ var queryId = ctx.Stream.ReadLong();
+
+ var qryHandle = new ClientContinuousQueryHandle(ctx.Socket, queryId);
+
+ ctx.Socket.AddNotificationHandler(queryId,
+ (stream, err) => HandleContinuousQueryEvents(stream, err, listener, qryHandle));
+
+ return qryHandle;
+ });
+ }
+
+ /// <summary>
+ /// Writes the continuous query.
+ /// </summary>
+ /// <param name="ctx">Request context.</param>
+ /// <param name="continuousQuery">Query.</param>
+ private void WriteContinuousQuery(ClientRequestContext ctx, ContinuousQueryClient<TK, TV> continuousQuery)
+ {
+ var w = ctx.Writer;
+ w.WriteInt(continuousQuery.BufferSize);
+ w.WriteLong((long) continuousQuery.TimeInterval.TotalMilliseconds);
+ w.WriteBoolean(false); // Include expired.
+
+ if (continuousQuery.Filter == null)
+ {
+ w.WriteObject<object>(null);
+ }
+ else
+ {
+ var javaFilter = continuousQuery.Filter as PlatformJavaObjectFactoryProxy;
+
+ if (javaFilter != null)
+ {
+ w.WriteObject(javaFilter.GetRawProxy());
+ w.WriteByte(FilterPlatformJava);
+ }
+ else
+ {
+ var filterHolder = new ContinuousQueryFilterHolder(continuousQuery.Filter, _keepBinary);
+
+ w.WriteObject(filterHolder);
+ w.WriteByte(FilterPlatformDotnet);
+ }
+ }
+
+ ctx.Socket.ExpectNotifications();
+ }
+
+ /// <summary>
+ /// Handles continuous query events.
+ /// </summary>
+ private void HandleContinuousQueryEvents(IBinaryStream stream, Exception err,
+ ICacheEntryEventListener<TK, TV> listener, ClientContinuousQueryHandle qryHandle)
+ {
+ if (err != null)
+ {
+ qryHandle.OnError(err);
+ return;
+ }
+
+ var flags = (ClientFlags) stream.ReadShort();
+ var opCode = (ClientOp) stream.ReadShort();
+
+ if ((flags & ClientFlags.Error) == ClientFlags.Error)
+ {
+ var status = (ClientStatusCode) stream.ReadInt();
+ var msg = _marsh.Unmarshal<string>(stream);
+
+ GetLogger().Error("Error while handling Continuous Query notification ({0}): {1}", status, msg);
+
+ qryHandle.OnError(new IgniteClientException(msg, null, status));
+
+ return;
+ }
+
+ if (opCode == ClientOp.QueryContinuousEventNotification)
+ {
+ var evts = ContinuousQueryUtils.ReadEvents<TK, TV>(stream, _marsh, _keepBinary);
+
+ listener.OnEvent(evts);
+
+ return;
+ }
+
+ GetLogger().Error("Error while handling Continuous Query notification: unexpected op '{0}'", opCode);
+ }
+
+ /// <summary>
+ /// Gets the logger.
+ /// </summary>
+ private ILogger GetLogger()
+ {
+ // Don't care about thread safety here, it is ok to initialize multiple times.
+ // ReSharper disable once ConvertIfStatementToNullCoalescingExpression (readability).
+ if (_logger == null)
+ {
+ _logger = _ignite.Configuration.Logger != null
+ ? _ignite.Configuration.Logger.GetLogger(GetType())
+ : NoopLogger.Instance;
+ }
+
+ return _logger;
+ }
}
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientContinuousQueryHandle.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientContinuousQueryHandle.cs
new file mode 100644
index 0000000..67917c1
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientContinuousQueryHandle.cs
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Core.Impl.Client.Cache.Query
+{
+ using System;
+ using Apache.Ignite.Core.Client.Cache.Query.Continuous;
+
+ /// <summary>
+ /// Thin client continuous query handle.
+ /// </summary>
+ internal sealed class ClientContinuousQueryHandle : IContinuousQueryHandleClient
+ {
+ /** Socket. */
+ private readonly ClientSocket _socket;
+
+ /** Cursor ID. */
+ private readonly long _queryId;
+
+ /** */
+ private readonly object _disposeSyncRoot = new object();
+
+ /** */
+ private bool _disposed;
+
+ /// <summary>
+ /// Initializes a new instance of <see cref="ClientContinuousQueryHandle"/>.
+ /// </summary>
+ public ClientContinuousQueryHandle(ClientSocket socket, long queryId)
+ {
+ _socket = socket;
+ _queryId = queryId;
+ }
+
+ /** <inheritdoc /> */
+ public event EventHandler<ContinuousQueryClientDisconnectedEventArgs> Disconnected;
+
+ /** <inheritdoc /> */
+ public void Dispose()
+ {
+ lock (_disposeSyncRoot)
+ {
+ if (_disposed)
+ {
+ return;
+ }
+
+ if (!_socket.IsDisposed)
+ {
+ _socket.DoOutInOp<object>(ClientOp.ResourceClose,
+ ctx => ctx.Writer.WriteLong(_queryId), null);
+
+ _socket.RemoveNotificationHandler(_queryId);
+ }
+
+ _disposed = true;
+ }
+ }
+
+ /// <summary>
+ /// Called when error occurs during the continuous query execution.
+ /// </summary>
+ internal void OnError(Exception exception)
+ {
+ lock (_disposeSyncRoot)
+ {
+ if (_disposed)
+ {
+ return;
+ }
+
+ var disconnected = Disconnected;
+ if (disconnected != null)
+ {
+ disconnected.Invoke(this, new ContinuousQueryClientDisconnectedEventArgs(exception));
+ }
+
+ _disposed = true;
+ }
+ }
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientOp.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientOp.cs
index a038cc0..486a9eb0 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientOp.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientOp.cs
@@ -55,7 +55,7 @@ namespace Apache.Ignite.Core.Impl.Client
CacheGetConfiguration = 1055,
CacheDestroy = 1056,
CachePartitions = 1101,
-
+
// Queries.
QueryScan = 2000,
QueryScanCursorGetPage = 2001,
@@ -63,6 +63,8 @@ namespace Apache.Ignite.Core.Impl.Client
QuerySqlCursorGetPage = 2003,
QuerySqlFields = 2004,
QuerySqlFieldsCursorGetPage = 2005,
+ QueryContinuous = 2006,
+ QueryContinuousEventNotification = 2007,
// Metadata.
BinaryTypeNameGet = 3000,
@@ -78,7 +80,7 @@ namespace Apache.Ignite.Core.Impl.Client
ClusterGroupGetNodeIds = 5100,
ClusterGroupGetNodesInfo = 5101,
ClusterGroupGetNodesEndpoints = 5102,
-
+
// Compute.
ComputeTaskExecute = 6000,
ComputeTaskFinished = 6001
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
index bec7c10..de4c55f 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
@@ -283,6 +283,12 @@ namespace Apache.Ignite.Core.Impl.Client
ClientNotificationHandler unused;
var removed = _notificationListeners.TryRemove(notificationId, out unused);
Debug.Assert(removed);
+
+ var count = Interlocked.Decrement(ref _expectedNotifications);
+ if (count < 0)
+ {
+ throw new IgniteClientException("Negative thin client expected notification count.");
+ }
}
/// <summary>
@@ -445,14 +451,8 @@ namespace Apache.Ignite.Core.Impl.Client
return false;
}
- var count = Interlocked.Decrement(ref _expectedNotifications);
- if (count < 0)
- {
- throw new IgniteClientException("Unexpected thin client notification: " + requestId);
- }
-
- var handler = _notificationListeners.GetOrAdd(requestId, _ => new ClientNotificationHandler(_logger));
- handler.Handle(stream, null);
+ _notificationListeners.GetOrAdd(requestId, _ => new ClientNotificationHandler(_logger))
+ .Handle(stream, null);
return true;
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/IgniteClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/IgniteClient.cs
index 0815b65..183f8f9 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/IgniteClient.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/IgniteClient.cs
@@ -64,7 +64,7 @@ namespace Apache.Ignite.Core.Impl.Client
/** Node info cache. */
private readonly ConcurrentDictionary<Guid, IClientClusterNode> _nodes =
new ConcurrentDictionary<Guid, IClientClusterNode>();
-
+
/** Cluster. */
private readonly ClientCluster _cluster;
@@ -91,16 +91,16 @@ namespace Apache.Ignite.Core.Impl.Client
_binProc = _configuration.BinaryProcessor ?? new BinaryProcessorClient(_socket);
_binary = new Binary(_marsh);
-
+
_cluster = new ClientCluster(this);
-
+
_compute = new ComputeClient(this, ComputeClientFlags.None, TimeSpan.Zero, null);
}
/// <summary>
/// Gets the socket.
/// </summary>
- public ClientFailoverSocket Socket
+ internal ClientFailoverSocket Socket
{
get { return _socket; }
}
@@ -208,6 +208,7 @@ namespace Apache.Ignite.Core.Impl.Client
throw GetClientNotSupportedException();
}
+ /** <inheritDoc /> */
public object GetJavaThreadLocal()
{
throw GetClientNotSupportedException();