You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2015/09/21 16:27:40 UTC
[44/52] [partial] ignite git commit: IGNITE-1513: Moved .Net.
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheMetricsImpl.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheMetricsImpl.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheMetricsImpl.cs
new file mode 100644
index 0000000..b5982f6
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheMetricsImpl.cs
@@ -0,0 +1,248 @@
+/*
+ * 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.Cache
+{
+ using Apache.Ignite.Core.Cache;
+ using Apache.Ignite.Core.Portable;
+
+ /// <summary>
+ /// Cache metrics used to obtain statistics on cache.
+ /// </summary>
+ internal class CacheMetricsImpl : ICacheMetrics
+ {
+ /// <summary>
+ /// Initializes a new instance of the <see cref="CacheMetricsImpl"/> class.
+ /// </summary>
+ /// <param name="reader">The reader.</param>
+ public CacheMetricsImpl(IPortableRawReader reader)
+ {
+ CacheGets = reader.ReadLong();
+ CachePuts = reader.ReadLong();
+ CacheHits = reader.ReadLong();
+ CacheMisses = reader.ReadLong();
+ CacheTxCommits = reader.ReadLong();
+ CacheTxRollbacks = reader.ReadLong();
+ CacheEvictions = reader.ReadLong();
+ CacheRemovals = reader.ReadLong();
+ AveragePutTime = reader.ReadFloat();
+ AverageGetTime = reader.ReadFloat();
+ AverageRemoveTime = reader.ReadFloat();
+ AverageTxCommitTime = reader.ReadFloat();
+ AverageTxRollbackTime = reader.ReadFloat();
+ CacheName = reader.ReadString();
+ OverflowSize = reader.ReadLong();
+ OffHeapEntriesCount = reader.ReadLong();
+ OffHeapAllocatedSize = reader.ReadLong();
+ Size = reader.ReadInt();
+ KeySize = reader.ReadInt();
+ IsEmpty = reader.ReadBoolean();
+ DhtEvictQueueCurrentSize = reader.ReadInt();
+ TxThreadMapSize = reader.ReadInt();
+ TxXidMapSize = reader.ReadInt();
+ TxCommitQueueSize = reader.ReadInt();
+ TxPrepareQueueSize = reader.ReadInt();
+ TxStartVersionCountsSize = reader.ReadInt();
+ TxCommittedVersionsSize = reader.ReadInt();
+ TxRolledbackVersionsSize = reader.ReadInt();
+ TxDhtThreadMapSize = reader.ReadInt();
+ TxDhtXidMapSize = reader.ReadInt();
+ TxDhtCommitQueueSize = reader.ReadInt();
+ TxDhtPrepareQueueSize = reader.ReadInt();
+ TxDhtStartVersionCountsSize = reader.ReadInt();
+ TxDhtCommittedVersionsSize = reader.ReadInt();
+ TxDhtRolledbackVersionsSize = reader.ReadInt();
+ IsWriteBehindEnabled = reader.ReadBoolean();
+ WriteBehindFlushSize = reader.ReadInt();
+ WriteBehindFlushThreadCount = reader.ReadInt();
+ WriteBehindFlushFrequency = reader.ReadLong();
+ WriteBehindStoreBatchSize = reader.ReadInt();
+ WriteBehindTotalCriticalOverflowCount = reader.ReadInt();
+ WriteBehindCriticalOverflowCount = reader.ReadInt();
+ WriteBehindErrorRetryCount = reader.ReadInt();
+ WriteBehindBufferSize = reader.ReadInt();
+ KeyType = reader.ReadString();
+ ValueType = reader.ReadString();
+ IsStoreByValue = reader.ReadBoolean();
+ IsStatisticsEnabled = reader.ReadBoolean();
+ IsManagementEnabled = reader.ReadBoolean();
+ IsReadThrough = reader.ReadBoolean();
+ IsWriteThrough = reader.ReadBoolean();
+ CacheHitPercentage = reader.ReadFloat();
+ CacheMissPercentage = reader.ReadFloat();
+ }
+
+ /** <inheritdoc /> */
+ public long CacheHits { get; private set; }
+
+ /** <inheritdoc /> */
+ public float CacheHitPercentage { get; private set; }
+
+ /** <inheritdoc /> */
+ public long CacheMisses { get; private set; }
+
+ /** <inheritdoc /> */
+ public float CacheMissPercentage { get; private set; }
+
+ /** <inheritdoc /> */
+ public long CacheGets { get; private set; }
+
+ /** <inheritdoc /> */
+ public long CachePuts { get; private set; }
+
+ /** <inheritdoc /> */
+ public long CacheRemovals { get; private set; }
+
+ /** <inheritdoc /> */
+ public long CacheEvictions { get; private set; }
+
+ /** <inheritdoc /> */
+ public float AverageGetTime { get; private set; }
+
+ /** <inheritdoc /> */
+ public float AveragePutTime { get; private set; }
+
+ /** <inheritdoc /> */
+ public float AverageRemoveTime { get; private set; }
+
+ /** <inheritdoc /> */
+ public float AverageTxCommitTime { get; private set; }
+
+ /** <inheritdoc /> */
+ public float AverageTxRollbackTime { get; private set; }
+
+ /** <inheritdoc /> */
+ public long CacheTxCommits { get; private set; }
+
+ /** <inheritdoc /> */
+ public long CacheTxRollbacks { get; private set; }
+
+ /** <inheritdoc /> */
+ public string CacheName { get; private set; }
+
+ /** <inheritdoc /> */
+ public long OverflowSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public long OffHeapEntriesCount { get; private set; }
+
+ /** <inheritdoc /> */
+ public long OffHeapAllocatedSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int Size { get; private set; }
+
+ /** <inheritdoc /> */
+ public int KeySize { get; private set; }
+
+ /** <inheritdoc /> */
+ public bool IsEmpty { get; private set; }
+
+ /** <inheritdoc /> */
+ public int DhtEvictQueueCurrentSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxThreadMapSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxXidMapSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxCommitQueueSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxPrepareQueueSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxStartVersionCountsSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxCommittedVersionsSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxRolledbackVersionsSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxDhtThreadMapSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxDhtXidMapSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxDhtCommitQueueSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxDhtPrepareQueueSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxDhtStartVersionCountsSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxDhtCommittedVersionsSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int TxDhtRolledbackVersionsSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public bool IsWriteBehindEnabled { get; private set; }
+
+ /** <inheritdoc /> */
+ public int WriteBehindFlushSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int WriteBehindFlushThreadCount { get; private set; }
+
+ /** <inheritdoc /> */
+ public long WriteBehindFlushFrequency { get; private set; }
+
+ /** <inheritdoc /> */
+ public int WriteBehindStoreBatchSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public int WriteBehindTotalCriticalOverflowCount { get; private set; }
+
+ /** <inheritdoc /> */
+ public int WriteBehindCriticalOverflowCount { get; private set; }
+
+ /** <inheritdoc /> */
+ public int WriteBehindErrorRetryCount { get; private set; }
+
+ /** <inheritdoc /> */
+ public int WriteBehindBufferSize { get; private set; }
+
+ /** <inheritdoc /> */
+ public string KeyType { get; private set; }
+
+ /** <inheritdoc /> */
+ public string ValueType { get; private set; }
+
+ /** <inheritdoc /> */
+ public bool IsStoreByValue { get; private set; }
+
+ /** <inheritdoc /> */
+ public bool IsStatisticsEnabled { get; private set; }
+
+ /** <inheritdoc /> */
+ public bool IsManagementEnabled { get; private set; }
+
+ /** <inheritdoc /> */
+ public bool IsReadThrough { get; private set; }
+
+ /** <inheritdoc /> */
+ public bool IsWriteThrough { get; private set; }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheOp.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheOp.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheOp.cs
new file mode 100644
index 0000000..3eb63ca
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheOp.cs
@@ -0,0 +1,63 @@
+/*
+ * 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.Cache
+{
+ /// <summary>
+ /// Cache opcodes.
+ /// </summary>
+ internal enum CacheOp
+ {
+ Clear = 1,
+ ClearAll = 2,
+ ContainsKey = 3,
+ ContainsKeys = 4,
+ Get = 5,
+ GetAll = 6,
+ GetAndPut = 7,
+ GetAndPutIfAbsent = 8,
+ GetAndRemove = 9,
+ GetAndReplace = 10,
+ GetName = 11,
+ Invoke = 12,
+ InvokeAll = 13,
+ IsLocalLocked = 14,
+ LoadCache = 15,
+ LocEvict = 16,
+ LocLoadCache = 17,
+ LocPromote = 18,
+ LocalClear = 20,
+ LocalClearAll = 21,
+ Lock = 22,
+ LockAll = 23,
+ Metrics = 24,
+ Peek = 25,
+ Put = 26,
+ PutAll = 27,
+ PutIfAbsent = 28,
+ QryContinuous = 29,
+ QryScan = 30,
+ QrySql = 31,
+ QrySqlFields = 32,
+ QryTxt = 33,
+ RemoveAll = 34,
+ RemoveBool = 35,
+ RemoveObj = 36,
+ Replace2 = 37,
+ Replace3 = 38
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheProxyImpl.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheProxyImpl.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheProxyImpl.cs
new file mode 100644
index 0000000..bfd7866
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/CacheProxyImpl.cs
@@ -0,0 +1,500 @@
+/*
+ * 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.Cache
+{
+ using System.Collections;
+ using System.Collections.Generic;
+ using System.Diagnostics;
+ using System.Diagnostics.CodeAnalysis;
+ using System.Threading;
+ using Apache.Ignite.Core.Cache;
+ using Apache.Ignite.Core.Cache.Expiry;
+ using Apache.Ignite.Core.Cache.Query;
+ using Apache.Ignite.Core.Cache.Query.Continuous;
+ using Apache.Ignite.Core.Common;
+
+ /// <summary>
+ /// Cache proxy.
+ /// </summary>
+ [SuppressMessage("Microsoft.Design", "CA1001:TypesThatOwnDisposableFieldsShouldBeDisposable")]
+ internal class CacheProxyImpl<TK, TV> : ICache<TK, TV>
+ {
+ /** wrapped cache instance */
+ private readonly CacheImpl<TK, TV> _cache;
+
+ /** */
+ private readonly ThreadLocal<int> _lastAsyncOp = new ThreadLocal<int>(() => PlatformTarget.OpNone);
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="CacheProxyImpl{K, V}"/> class.
+ /// </summary>
+ /// <param name="cache">The cache to wrap.</param>
+ public CacheProxyImpl(CacheImpl<TK, TV> cache)
+ {
+ Debug.Assert(cache != null);
+
+ _cache = cache;
+ }
+
+ /** <inheritDoc /> */
+ public ICache<TK, TV> WithSkipStore()
+ {
+ return _cache.IsSkipStore ? this : new CacheProxyImpl<TK, TV>((CacheImpl<TK, TV>)_cache.WithSkipStore());
+ }
+
+ /** <inheritDoc /> */
+ public ICache<TK, TV> WithExpiryPolicy(IExpiryPolicy plc)
+ {
+ return new CacheProxyImpl<TK, TV>((CacheImpl<TK, TV>)_cache.WithExpiryPolicy(plc));
+ }
+
+ /** <inheritDoc /> */
+ public ICache<TK, TV> WithAsync()
+ {
+ return IsAsync ? this : new CacheProxyImpl<TK, TV>((CacheImpl<TK, TV>) _cache.WithAsync());
+ }
+
+ /** <inheritDoc /> */
+ public bool IsAsync
+ {
+ get { return _cache.IsAsync; }
+ }
+
+ /** <inheritDoc /> */
+ public IFuture GetFuture()
+ {
+ return GetFuture<object>();
+ }
+
+ /** <inheritDoc /> */
+ public IFuture<TResult> GetFuture<TResult>()
+ {
+ var fut = _cache.GetFuture<TResult>(_lastAsyncOp.Value);
+
+ ClearLastAsyncOp();
+
+ return fut;
+ }
+
+ /** <inheritDoc /> */
+ public IEnumerator<ICacheEntry<TK, TV>> GetEnumerator()
+ {
+ return _cache.GetEnumerator();
+ }
+
+ /** <inheritDoc /> */
+ IEnumerator IEnumerable.GetEnumerator()
+ {
+ return ((IEnumerable) _cache).GetEnumerator();
+ }
+
+ /** <inheritDoc /> */
+ public string Name
+ {
+ get { return _cache.Name; }
+ }
+
+ /** <inheritDoc /> */
+ public IIgnite Ignite
+ {
+ get { return _cache.Ignite; }
+ }
+
+ /** <inheritDoc /> */
+
+ public bool IsEmpty()
+ {
+ return _cache.IsEmpty();
+ }
+
+ /** <inheritDoc /> */
+ public bool IsKeepPortable
+ {
+ get { return _cache.IsKeepPortable; }
+ }
+
+ /// <summary>
+ /// Skip store flag.
+ /// </summary>
+ internal bool SkipStore
+ {
+ get { return _cache.IsSkipStore; }
+ }
+
+ /** <inheritDoc /> */
+ public ICache<TK1, TV1> WithKeepPortable<TK1, TV1>()
+ {
+ return new CacheProxyImpl<TK1, TV1>((CacheImpl<TK1, TV1>) _cache.WithKeepPortable<TK1, TV1>());
+ }
+
+ /** <inheritDoc /> */
+ public void LoadCache(ICacheEntryFilter<TK, TV> p, params object[] args)
+ {
+ _cache.LoadCache(p, args);
+
+ SetLastAsyncOp(CacheOp.LoadCache);
+ }
+
+ /** <inheritDoc /> */
+ public void LocalLoadCache(ICacheEntryFilter<TK, TV> p, params object[] args)
+ {
+ _cache.LocalLoadCache(p, args);
+
+ SetLastAsyncOp(CacheOp.LocLoadCache);
+ }
+
+ /** <inheritDoc /> */
+ public bool ContainsKey(TK key)
+ {
+ var result = _cache.ContainsKey(key);
+
+ SetLastAsyncOp(CacheOp.ContainsKey);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public bool ContainsKeys(IEnumerable<TK> keys)
+ {
+ var result = _cache.ContainsKeys(keys);
+
+ SetLastAsyncOp(CacheOp.ContainsKeys);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public TV LocalPeek(TK key, params CachePeekMode[] modes)
+ {
+ return _cache.LocalPeek(key, modes);
+ }
+
+ /** <inheritDoc /> */
+ public TV Get(TK key)
+ {
+ var result = _cache.Get(key);
+
+ SetLastAsyncOp(CacheOp.Get);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public IDictionary<TK, TV> GetAll(IEnumerable<TK> keys)
+ {
+ var result = _cache.GetAll(keys);
+
+ SetLastAsyncOp(CacheOp.GetAll);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public void Put(TK key, TV val)
+ {
+ _cache.Put(key, val);
+
+ SetLastAsyncOp(CacheOp.Put);
+ }
+
+ /** <inheritDoc /> */
+ public TV GetAndPut(TK key, TV val)
+ {
+ var result = _cache.GetAndPut(key, val);
+
+ SetLastAsyncOp(CacheOp.GetAndPut);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public TV GetAndReplace(TK key, TV val)
+ {
+ var result = _cache.GetAndReplace(key, val);
+
+ SetLastAsyncOp(CacheOp.GetAndReplace);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public TV GetAndRemove(TK key)
+ {
+ var result = _cache.GetAndRemove(key);
+
+ SetLastAsyncOp(CacheOp.GetAndRemove);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public bool PutIfAbsent(TK key, TV val)
+ {
+ var result = _cache.PutIfAbsent(key, val);
+
+ SetLastAsyncOp(CacheOp.PutIfAbsent);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public TV GetAndPutIfAbsent(TK key, TV val)
+ {
+ var result = _cache.GetAndPutIfAbsent(key, val);
+
+ SetLastAsyncOp(CacheOp.GetAndPutIfAbsent);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public bool Replace(TK key, TV val)
+ {
+ var result = _cache.Replace(key, val);
+
+ SetLastAsyncOp(CacheOp.Replace2);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public bool Replace(TK key, TV oldVal, TV newVal)
+ {
+ var result = _cache.Replace(key, oldVal, newVal);
+
+ SetLastAsyncOp(CacheOp.Replace3);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public void PutAll(IDictionary<TK, TV> vals)
+ {
+ _cache.PutAll(vals);
+
+ SetLastAsyncOp(CacheOp.PutAll);
+ }
+
+ /** <inheritDoc /> */
+ public void LocalEvict(IEnumerable<TK> keys)
+ {
+ _cache.LocalEvict(keys);
+ }
+
+ /** <inheritDoc /> */
+ public void Clear()
+ {
+ _cache.Clear();
+
+ ClearLastAsyncOp();
+ }
+
+ /** <inheritDoc /> */
+ public void Clear(TK key)
+ {
+ _cache.Clear(key);
+
+ SetLastAsyncOp(CacheOp.Clear);
+ }
+
+ /** <inheritDoc /> */
+ public void ClearAll(IEnumerable<TK> keys)
+ {
+ _cache.ClearAll(keys);
+
+ SetLastAsyncOp(CacheOp.ClearAll);
+ }
+
+ /** <inheritDoc /> */
+ public void LocalClear(TK key)
+ {
+ _cache.LocalClear(key);
+ }
+
+ /** <inheritDoc /> */
+ public void LocalClearAll(IEnumerable<TK> keys)
+ {
+ _cache.LocalClearAll(keys);
+ }
+
+ /** <inheritDoc /> */
+ public bool Remove(TK key)
+ {
+ var result = _cache.Remove(key);
+
+ SetLastAsyncOp(CacheOp.RemoveObj);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public bool Remove(TK key, TV val)
+ {
+ var result = _cache.Remove(key, val);
+
+ SetLastAsyncOp(CacheOp.RemoveBool);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public void RemoveAll(IEnumerable<TK> keys)
+ {
+ _cache.RemoveAll(keys);
+
+ SetLastAsyncOp(CacheOp.RemoveAll);
+ }
+
+ /** <inheritDoc /> */
+ public void RemoveAll()
+ {
+ _cache.RemoveAll();
+
+ ClearLastAsyncOp();
+ }
+
+ /** <inheritDoc /> */
+ public int GetLocalSize(params CachePeekMode[] modes)
+ {
+ return _cache.GetLocalSize(modes);
+ }
+
+ /** <inheritDoc /> */
+ public int GetSize(params CachePeekMode[] modes)
+ {
+ var result = _cache.GetSize(modes);
+
+ ClearLastAsyncOp();
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public void LocalPromote(IEnumerable<TK> keys)
+ {
+ _cache.LocalPromote(keys);
+ }
+
+ /** <inheritDoc /> */
+ public IQueryCursor<ICacheEntry<TK, TV>> Query(QueryBase qry)
+ {
+ return _cache.Query(qry);
+ }
+
+ /** <inheritDoc /> */
+ public IQueryCursor<IList> QueryFields(SqlFieldsQuery qry)
+ {
+ return _cache.QueryFields(qry);
+ }
+
+ /** <inheritDoc /> */
+ public IContinuousQueryHandle QueryContinuous(ContinuousQuery<TK, TV> qry)
+ {
+ return _cache.QueryContinuous(qry);
+ }
+
+ /** <inheritDoc /> */
+ public IContinuousQueryHandle<ICacheEntry<TK, TV>> QueryContinuous(ContinuousQuery<TK, TV> qry, QueryBase initialQry)
+ {
+ return _cache.QueryContinuous(qry, initialQry);
+ }
+
+ /** <inheritDoc /> */
+ public IEnumerable<ICacheEntry<TK, TV>> GetLocalEntries(params CachePeekMode[] peekModes)
+ {
+ return _cache.GetLocalEntries(peekModes);
+ }
+
+ /** <inheritDoc /> */
+ public TR Invoke<TR, TA>(TK key, ICacheEntryProcessor<TK, TV, TA, TR> processor, TA arg)
+ {
+ var result = _cache.Invoke(key, processor, arg);
+
+ SetLastAsyncOp(CacheOp.Invoke);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public IDictionary<TK, ICacheEntryProcessorResult<TR>> InvokeAll<TR, TA>(IEnumerable<TK> keys,
+ ICacheEntryProcessor<TK, TV, TA, TR> processor, TA arg)
+ {
+ var result = _cache.InvokeAll(keys, processor, arg);
+
+ SetLastAsyncOp(CacheOp.InvokeAll);
+
+ return result;
+ }
+
+ /** <inheritDoc /> */
+ public ICacheLock Lock(TK key)
+ {
+ return _cache.Lock(key);
+ }
+
+ /** <inheritDoc /> */
+ public ICacheLock LockAll(IEnumerable<TK> keys)
+ {
+ return _cache.LockAll(keys);
+ }
+
+ /** <inheritDoc /> */
+ public bool IsLocalLocked(TK key, bool byCurrentThread)
+ {
+ return _cache.IsLocalLocked(key, byCurrentThread);
+ }
+
+ /** <inheritDoc /> */
+ public ICacheMetrics GetMetrics()
+ {
+ return _cache.GetMetrics();
+ }
+
+ /** <inheritDoc /> */
+ public IFuture Rebalance()
+ {
+ return _cache.Rebalance();
+ }
+
+ /** <inheritDoc /> */
+ public ICache<TK, TV> WithNoRetries()
+ {
+ return _cache.IsNoRetries ? this : new CacheProxyImpl<TK, TV>((CacheImpl<TK, TV>) _cache.WithNoRetries());
+ }
+
+ /// <summary>
+ /// Sets the last asynchronous op id.
+ /// </summary>
+ /// <param name="opId">The op identifier.</param>
+ private void SetLastAsyncOp(CacheOp opId)
+ {
+ if (IsAsync)
+ _lastAsyncOp.Value = (int) opId;
+ }
+
+ /// <summary>
+ /// Clears the last asynchronous op id.
+ /// This should be called in the end of each method that supports async and does not call SetLastAsyncOp.
+ /// </summary>
+ private void ClearLastAsyncOp()
+ {
+ if (IsAsync)
+ _lastAsyncOp.Value = PlatformTarget.OpNone;
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryCreateEvent.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryCreateEvent.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryCreateEvent.cs
new file mode 100644
index 0000000..8d9dfef
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryCreateEvent.cs
@@ -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.
+ */
+
+namespace Apache.Ignite.Core.Impl.Cache.Event
+{
+ using Apache.Ignite.Core.Cache.Event;
+
+ /// <summary>
+ /// Cache entry create event.
+ /// </summary>
+ internal class CacheEntryCreateEvent<TK, TV> : ICacheEntryEvent<TK, TV>
+ {
+ /** Key.*/
+ private readonly TK _key;
+
+ /** Value.*/
+ private readonly TV _val;
+
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="key">Key.</param>
+ /// <param name="val">Value.</param>
+ public CacheEntryCreateEvent(TK key, TV val)
+ {
+ _key = key;
+ _val = val;
+ }
+
+ /** <inheritdoc /> */
+ public TK Key
+ {
+ get { return _key; }
+ }
+
+ /** <inheritdoc /> */
+ public TV Value
+ {
+ get { return _val; }
+ }
+
+ /** <inheritdoc /> */
+ public TV OldValue
+ {
+ get { return default(TV); }
+ }
+
+ /** <inheritdoc /> */
+ public bool HasOldValue
+ {
+ get { return false; }
+ }
+
+ /** <inheritdoc /> */
+ public CacheEntryEventType EventType
+ {
+ get { return CacheEntryEventType.Created; }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryRemoveEvent.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryRemoveEvent.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryRemoveEvent.cs
new file mode 100644
index 0000000..a44a800
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryRemoveEvent.cs
@@ -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.
+ */
+
+namespace Apache.Ignite.Core.Impl.Cache.Event
+{
+ using Apache.Ignite.Core.Cache.Event;
+
+ /// <summary>
+ /// Cache entry remove event.
+ /// </summary>
+ internal class CacheEntryRemoveEvent<TK, TV> : ICacheEntryEvent<TK, TV>
+ {
+ /** Key.*/
+ private readonly TK _key;
+
+ /** Old value.*/
+ private readonly TV _oldVal;
+
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="key">Key.</param>
+ /// <param name="oldVal">Old value.</param>
+ public CacheEntryRemoveEvent(TK key, TV oldVal)
+ {
+ _key = key;
+ _oldVal = oldVal;
+ }
+
+ /** <inheritdoc /> */
+ public TK Key
+ {
+ get { return _key; }
+ }
+
+ /** <inheritdoc /> */
+ public TV Value
+ {
+ get { return default(TV); }
+ }
+
+ /** <inheritdoc /> */
+ public TV OldValue
+ {
+ get { return _oldVal; }
+ }
+
+ /** <inheritdoc /> */
+ public bool HasOldValue
+ {
+ get { return true; }
+ }
+
+ /** <inheritdoc /> */
+ public CacheEntryEventType EventType
+ {
+ get { return CacheEntryEventType.Removed; }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryUpdateEvent.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryUpdateEvent.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryUpdateEvent.cs
new file mode 100644
index 0000000..e6fb927
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Event/CacheEntryUpdateEvent.cs
@@ -0,0 +1,79 @@
+/*
+ * 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.Cache.Event
+{
+ using Apache.Ignite.Core.Cache.Event;
+
+ /// <summary>
+ /// Cache entry update event.
+ /// </summary>
+ internal class CacheEntryUpdateEvent<TK, TV> : ICacheEntryEvent<TK, TV>
+ {
+ /** Key.*/
+ private readonly TK _key;
+
+ /** Value.*/
+ private readonly TV _val;
+
+ /** Old value.*/
+ private readonly TV _oldVal;
+
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="key">Key.</param>
+ /// <param name="oldVal">Old value.</param>
+ /// <param name="val">Value.</param>
+ public CacheEntryUpdateEvent(TK key, TV oldVal, TV val)
+ {
+ _key = key;
+ _oldVal = oldVal;
+ _val = val;
+ }
+
+ /** <inheritdoc /> */
+ public TK Key
+ {
+ get { return _key; }
+ }
+
+ /** <inheritdoc /> */
+ public TV Value
+ {
+ get { return _val; }
+ }
+
+ /** <inheritdoc /> */
+ public TV OldValue
+ {
+ get { return _oldVal; }
+ }
+
+ /** <inheritdoc /> */
+ public bool HasOldValue
+ {
+ get { return true; }
+ }
+
+ /** <inheritdoc /> */
+ public CacheEntryEventType EventType
+ {
+ get { return CacheEntryEventType.Updated; }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/MutableCacheEntry.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/MutableCacheEntry.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/MutableCacheEntry.cs
new file mode 100644
index 0000000..2c69043
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/MutableCacheEntry.cs
@@ -0,0 +1,163 @@
+/*
+ * 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.Cache
+{
+ using System;
+ using Apache.Ignite.Core.Cache;
+ using Apache.Ignite.Core.Impl.Common;
+
+ /// <summary>
+ /// Represents a cache entry.
+ /// </summary>
+ internal class MutableCacheEntry<TK, TV> : IMutableCacheEntry<TK, TV>, IMutableCacheEntryInternal
+ {
+ // Entry value
+ private TV _value;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="MutableCacheEntry{K, V}"/> class.
+ /// </summary>
+ /// <param name="key">The key.</param>
+ public MutableCacheEntry(TK key)
+ {
+ Key = key;
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="MutableCacheEntry{K, V}"/> class.
+ /// </summary>
+ /// <param name="key">The key.</param>
+ /// <param name="value">The value.</param>
+ public MutableCacheEntry(TK key, TV value)
+ {
+ Key = key;
+ _value = value;
+ Exists = true;
+ }
+
+ /** <inheritdoc /> */
+ public TK Key { get; private set; }
+
+ /** <inheritdoc /> */
+ object IMutableCacheEntryInternal.Key
+ {
+ get { return Key; }
+ }
+
+ /** <inheritdoc /> */
+ public TV Value
+ {
+ get { return _value; }
+ set
+ {
+ _value = value;
+ Exists = true;
+ State = MutableCacheEntryState.ValueSet;
+ }
+ }
+
+ /** <inheritdoc /> */
+ object IMutableCacheEntryInternal.Value
+ {
+ get { return Value; }
+ }
+
+ /** <inheritdoc /> */
+ public bool Exists { get; private set; }
+
+ /** <inheritdoc /> */
+ public void Remove()
+ {
+ Value = default(TV);
+ Exists = false;
+ State = MutableCacheEntryState.Removed;
+ }
+
+ /** <inheritdoc /> */
+ public MutableCacheEntryState State { get; private set; }
+ }
+
+ /// <summary>
+ /// Internal non-generic representation of a mutable cache entry.
+ /// </summary>
+ internal interface IMutableCacheEntryInternal
+ {
+ /// <summary>
+ /// Gets the key.
+ /// </summary>
+ object Key { get; }
+
+ /// <summary>
+ /// Gets the value.
+ /// </summary>
+ object Value { get; }
+
+ /// <summary>
+ /// Gets a value indicating whether cache entry exists.
+ /// </summary>
+ bool Exists { get; }
+
+ /// <summary>
+ /// Gets the state indicating user operation on this instance.
+ /// </summary>
+ MutableCacheEntryState State { get; }
+ }
+
+ /// <summary>
+ /// Mutable cache entry factory.
+ /// </summary>
+ internal static class MutableCacheEntry
+ {
+ private static readonly CopyOnWriteConcurrentDictionary<Tuple<Type, Type>, Func<object, object, bool, IMutableCacheEntryInternal>>
+ Ctors = new CopyOnWriteConcurrentDictionary<Tuple<Type, Type>, Func<object, object, bool, IMutableCacheEntryInternal>>();
+
+ public static Func<object, object, bool, IMutableCacheEntryInternal> GetCtor(Type keyType, Type valType)
+ {
+ Func<object, object, bool, IMutableCacheEntryInternal> result;
+ var funcKey = new Tuple<Type, Type>(keyType, valType);
+
+ return Ctors.TryGetValue(funcKey, out result)
+ ? result
+ : Ctors.GetOrAdd(funcKey, x =>
+ {
+ var entryType = typeof (MutableCacheEntry<,>).MakeGenericType(keyType, valType);
+
+ var oneArg = DelegateConverter.CompileCtor<Func<object, IMutableCacheEntryInternal>>(entryType,
+ new[] {keyType}, false);
+
+ var twoArg =
+ DelegateConverter.CompileCtor<Func<object, object, IMutableCacheEntryInternal>>(entryType,
+ new[] {keyType, valType}, false);
+
+ return (k, v, exists) => exists ? twoArg(k, v) : oneArg(k);
+ });
+ }
+ }
+
+ /// <summary>
+ /// Represents result of user operation on a mutable cache entry.
+ /// </summary>
+ internal enum MutableCacheEntryState : byte
+ {
+ Intact = 0,
+ ValueSet = 1,
+ Removed = 2,
+ ErrPortable = 3,
+ ErrString = 4
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/AbstractQueryCursor.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/AbstractQueryCursor.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/AbstractQueryCursor.cs
new file mode 100644
index 0000000..0f4b5a3
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/AbstractQueryCursor.cs
@@ -0,0 +1,264 @@
+/*
+ * 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.Cache.Query
+{
+ using System;
+ using System.Collections;
+ using System.Collections.Generic;
+ using System.Diagnostics.CodeAnalysis;
+ using Apache.Ignite.Core.Cache.Query;
+ using Apache.Ignite.Core.Impl.Portable;
+ using Apache.Ignite.Core.Impl.Portable.IO;
+ using Apache.Ignite.Core.Impl.Unmanaged;
+ using UU = Apache.Ignite.Core.Impl.Unmanaged.UnmanagedUtils;
+
+ /// <summary>
+ /// Abstract query cursor implementation.
+ /// </summary>
+ internal abstract class AbstractQueryCursor<T> : PlatformDisposableTarget, IQueryCursor<T>, IEnumerator<T>
+ {
+ /** */
+ private const int OpGetAll = 1;
+
+ /** */
+ private const int OpGetBatch = 2;
+
+ /** Position before head. */
+ private const int BatchPosBeforeHead = -1;
+
+ /** Keep portable flag. */
+ private readonly bool _keepPortable;
+
+ /** Wherther "GetAll" was called. */
+ private bool _getAllCalled;
+
+ /** Whether "GetEnumerator" was called. */
+ private bool _iterCalled;
+
+ /** Batch with entries. */
+ private T[] _batch;
+
+ /** Current position in batch. */
+ private int _batchPos = BatchPosBeforeHead;
+
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="target">Target.</param>
+ /// <param name="marsh">Marshaller.</param>
+ /// <param name="keepPortable">Keep portable flag.</param>
+ protected AbstractQueryCursor(IUnmanagedTarget target, PortableMarshaller marsh, bool keepPortable) :
+ base(target, marsh)
+ {
+ _keepPortable = keepPortable;
+ }
+
+ #region Public methods
+
+ /** <inheritdoc /> */
+ public IList<T> GetAll()
+ {
+ ThrowIfDisposed();
+
+ if (_iterCalled)
+ throw new InvalidOperationException("Failed to get all entries because GetEnumerator() " +
+ "method has already been called.");
+
+ if (_getAllCalled)
+ throw new InvalidOperationException("Failed to get all entries because GetAll() " +
+ "method has already been called.");
+
+ var res = DoInOp<IList<T>>(OpGetAll, ConvertGetAll);
+
+ _getAllCalled = true;
+
+ return res;
+ }
+
+ /** <inheritdoc /> */
+ protected override void Dispose(bool disposing)
+ {
+ try
+ {
+ UU.QueryCursorClose(Target);
+ }
+ finally
+ {
+ base.Dispose(disposing);
+ }
+ }
+
+ #endregion
+
+ #region Public IEnumerable methods
+
+ /** <inheritdoc /> */
+ [SuppressMessage("ReSharper", "PossibleNullReferenceException")]
+ public IEnumerator<T> GetEnumerator()
+ {
+ ThrowIfDisposed();
+
+ if (_iterCalled)
+ throw new InvalidOperationException("Failed to get enumerator entries because " +
+ "GetEnumeartor() method has already been called.");
+
+ if (_getAllCalled)
+ throw new InvalidOperationException("Failed to get enumerator entries because " +
+ "GetAll() method has already been called.");
+
+ UU.QueryCursorIterator(Target);
+
+ _iterCalled = true;
+
+ return this;
+ }
+
+ /** <inheritdoc /> */
+ IEnumerator IEnumerable.GetEnumerator()
+ {
+ return GetEnumerator();
+ }
+
+ #endregion
+
+ #region Public IEnumerator methods
+
+ /** <inheritdoc /> */
+ public T Current
+ {
+ get
+ {
+ ThrowIfDisposed();
+
+ if (_batchPos == BatchPosBeforeHead)
+ throw new InvalidOperationException("MoveNext has not been called.");
+
+ if (_batch == null)
+ throw new InvalidOperationException("Previous call to MoveNext returned false.");
+
+ return _batch[_batchPos];
+ }
+ }
+
+ /** <inheritdoc /> */
+ object IEnumerator.Current
+ {
+ get { return Current; }
+ }
+
+ /** <inheritdoc /> */
+ public bool MoveNext()
+ {
+ ThrowIfDisposed();
+
+ if (_batch == null)
+ {
+ if (_batchPos == BatchPosBeforeHead)
+ // Standing before head, let's get batch and advance position.
+ RequestBatch();
+ }
+ else
+ {
+ _batchPos++;
+
+ if (_batch.Length == _batchPos)
+ // Reached batch end => request another.
+ RequestBatch();
+ }
+
+ return _batch != null;
+ }
+
+ /** <inheritdoc /> */
+ public void Reset()
+ {
+ throw new NotSupportedException("Reset is not supported.");
+ }
+
+ #endregion
+
+ #region Non-public methods
+
+ /// <summary>
+ /// Read entry from the reader.
+ /// </summary>
+ /// <param name="reader">Reader.</param>
+ /// <returns>Entry.</returns>
+ protected abstract T Read(PortableReaderImpl reader);
+
+ /** <inheritdoc /> */
+ protected override T1 Unmarshal<T1>(IPortableStream stream)
+ {
+ return Marshaller.Unmarshal<T1>(stream, _keepPortable);
+ }
+
+ /// <summary>
+ /// Request next batch.
+ /// </summary>
+ private void RequestBatch()
+ {
+ _batch = DoInOp<T[]>(OpGetBatch, ConvertGetBatch);
+
+ _batchPos = 0;
+ }
+
+ /// <summary>
+ /// Converter for GET_ALL operation.
+ /// </summary>
+ /// <param name="stream">Portable stream.</param>
+ /// <returns>Result.</returns>
+ private IList<T> ConvertGetAll(IPortableStream stream)
+ {
+ var reader = Marshaller.StartUnmarshal(stream, _keepPortable);
+
+ var size = reader.ReadInt();
+
+ var res = new List<T>(size);
+
+ for (var i = 0; i < size; i++)
+ res.Add(Read(reader));
+
+ return res;
+ }
+
+ /// <summary>
+ /// Converter for GET_BATCH operation.
+ /// </summary>
+ /// <param name="stream">Portable stream.</param>
+ /// <returns>Result.</returns>
+ private T[] ConvertGetBatch(IPortableStream stream)
+ {
+ var reader = Marshaller.StartUnmarshal(stream, _keepPortable);
+
+ var size = reader.ReadInt();
+
+ if (size == 0)
+ return null;
+
+ var res = new T[size];
+
+ for (var i = 0; i < size; i++)
+ res[i] = Read(reader);
+
+ return res;
+ }
+
+ #endregion
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilter.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilter.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilter.cs
new file mode 100644
index 0000000..5738ed9
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilter.cs
@@ -0,0 +1,125 @@
+/*
+ * 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.Cache.Query.Continuous
+{
+ using Apache.Ignite.Core.Cache.Event;
+ using Apache.Ignite.Core.Impl.Portable.IO;
+ using Apache.Ignite.Core.Impl.Resource;
+ using CQU = ContinuousQueryUtils;
+
+ /// <summary>
+ /// Continuous query filter interface. Required to hide generic nature of underliyng real filter.
+ /// </summary>
+ internal interface IContinuousQueryFilter
+ {
+ /// <summary>
+ /// Evaluate filter.
+ /// </summary>
+ /// <param name="stream">Stream.</param>
+ /// <returns>Result.</returns>
+ bool Evaluate(IPortableStream stream);
+
+ /// <summary>
+ /// Inject grid.
+ /// </summary>
+ /// <param name="grid"></param>
+ void Inject(Ignite grid);
+
+ /// <summary>
+ /// Allocate handle for the filter.
+ /// </summary>
+ /// <returns></returns>
+ long Allocate();
+
+ /// <summary>
+ /// Release filter.
+ /// </summary>
+ void Release();
+ }
+
+ /// <summary>
+ /// Continuous query filter generic implementation.
+ /// </summary>
+ internal class ContinuousQueryFilter<TK, TV> : IContinuousQueryFilter
+ {
+ /** Actual filter. */
+ private readonly ICacheEntryEventFilter<TK, TV> _filter;
+
+ /** Keep portable flag. */
+ private readonly bool _keepPortable;
+
+ /** Ignite hosting the filter. */
+ private volatile Ignite _ignite;
+
+ /** GC handle. */
+ private long? _hnd;
+
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="filter">Actual filter.</param>
+ /// <param name="keepPortable">Keep portable flag.</param>
+ public ContinuousQueryFilter(ICacheEntryEventFilter<TK, TV> filter, bool keepPortable)
+ {
+ _filter = filter;
+ _keepPortable = keepPortable;
+ }
+
+ /** <inheritDoc /> */
+ public bool Evaluate(IPortableStream stream)
+ {
+ ICacheEntryEvent<TK, TV> evt = CQU.ReadEvent<TK, TV>(stream, _ignite.Marshaller, _keepPortable);
+
+ return _filter.Evaluate(evt);
+ }
+
+ /** <inheritDoc /> */
+ public void Inject(Ignite grid)
+ {
+ _ignite = grid;
+
+ ResourceProcessor.Inject(_filter, grid);
+ }
+
+ /** <inheritDoc /> */
+ public long Allocate()
+ {
+ lock (this)
+ {
+ if (!_hnd.HasValue)
+ _hnd = _ignite.HandleRegistry.Allocate(this);
+
+ return _hnd.Value;
+ }
+ }
+
+ /** <inheritDoc /> */
+ public void Release()
+ {
+ lock (this)
+ {
+ if (_hnd.HasValue)
+ {
+ _ignite.HandleRegistry.Release(_hnd.Value);
+
+ _hnd = null;
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilterHolder.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilterHolder.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilterHolder.cs
new file mode 100644
index 0000000..65da674
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryFilterHolder.cs
@@ -0,0 +1,118 @@
+/*
+ * 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.Cache.Query.Continuous
+{
+ using System;
+ using Apache.Ignite.Core.Impl.Portable;
+ using Apache.Ignite.Core.Portable;
+
+ /// <summary>
+ /// Continuous query remote filter holder. Wraps real filter into portable object,
+ /// so that it can be passed over wire to another node.
+ /// </summary>
+ public class ContinuousQueryFilterHolder : IPortableWriteAware
+ {
+ /** Key type. */
+ private readonly Type _keyTyp;
+
+ /** Value type. */
+ private readonly Type _valTyp;
+
+ /** Filter object. */
+ private readonly object _filter;
+
+ /** Keep portable flag. */
+ private readonly bool _keepPortable;
+
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="keyTyp">Key type.</param>
+ /// <param name="valTyp">Value type.</param>
+ /// <param name="filter">Filter.</param>
+ /// <param name="keepPortable">Keep portable flag.</param>
+ public ContinuousQueryFilterHolder(Type keyTyp, Type valTyp, object filter, bool keepPortable)
+ {
+ _keyTyp = keyTyp;
+ _valTyp = valTyp;
+ _filter = filter;
+ _keepPortable = keepPortable;
+ }
+
+ /// <summary>
+ /// Key type.
+ /// </summary>
+ internal Type KeyType
+ {
+ get { return _keyTyp; }
+ }
+
+ /// <summary>
+ /// Value type.
+ /// </summary>
+ internal Type ValueType
+ {
+ get { return _valTyp; }
+ }
+
+ /// <summary>
+ /// Filter.
+ /// </summary>
+ internal object Filter
+ {
+ get { return _filter; }
+ }
+
+ /// <summary>
+ /// Keep portable flag.
+ /// </summary>
+ internal bool KeepPortable
+ {
+ get { return _keepPortable; }
+ }
+
+ /// <summary>
+ /// Writes this object to the given writer.
+ /// </summary>
+ /// <param name="writer">Writer.</param>
+ public void WritePortable(IPortableWriter writer)
+ {
+ PortableWriterImpl rawWriter = (PortableWriterImpl) writer.RawWriter();
+
+ PortableUtils.WritePortableOrSerializable(rawWriter, _keyTyp);
+ PortableUtils.WritePortableOrSerializable(rawWriter, _valTyp);
+ PortableUtils.WritePortableOrSerializable(rawWriter, _filter);
+
+ rawWriter.WriteBoolean(_keepPortable);
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="ContinuousQueryFilterHolder"/> class.
+ /// </summary>
+ /// <param name="reader">The reader.</param>
+ public ContinuousQueryFilterHolder(IPortableReader reader)
+ {
+ PortableReaderImpl rawReader = (PortableReaderImpl) reader.RawReader();
+
+ _keyTyp = PortableUtils.ReadPortableOrSerializable<Type>(rawReader);
+ _valTyp = PortableUtils.ReadPortableOrSerializable<Type>(rawReader);
+ _filter = PortableUtils.ReadPortableOrSerializable<object>(rawReader);
+ _keepPortable = rawReader.ReadBoolean();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryHandleImpl.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryHandleImpl.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryHandleImpl.cs
new file mode 100644
index 0000000..d8d014b
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryHandleImpl.cs
@@ -0,0 +1,210 @@
+/*
+ * 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.Cache.Query.Continuous
+{
+ using System;
+ using System.Diagnostics;
+ using Apache.Ignite.Core.Cache;
+ using Apache.Ignite.Core.Cache.Event;
+ using Apache.Ignite.Core.Cache.Query;
+ using Apache.Ignite.Core.Cache.Query.Continuous;
+ using Apache.Ignite.Core.Impl.Portable;
+ using Apache.Ignite.Core.Impl.Portable.IO;
+ using Apache.Ignite.Core.Impl.Resource;
+ using Apache.Ignite.Core.Impl.Unmanaged;
+ using UU = Apache.Ignite.Core.Impl.Unmanaged.UnmanagedUtils;
+ using CQU = ContinuousQueryUtils;
+
+ /// <summary>
+ /// Continuous query handle interface.
+ /// </summary>
+ internal interface IContinuousQueryHandleImpl : IDisposable
+ {
+ /// <summary>
+ /// Process callback.
+ /// </summary>
+ /// <param name="stream">Stream.</param>
+ /// <returns>Result.</returns>
+ void Apply(IPortableStream stream);
+ }
+
+ /// <summary>
+ /// Continuous query handle.
+ /// </summary>
+ internal class ContinuousQueryHandleImpl<TK, TV> : IContinuousQueryHandleImpl, IContinuousQueryFilter,
+ IContinuousQueryHandle<ICacheEntry<TK, TV>>
+ {
+ /** Marshaller. */
+ private readonly PortableMarshaller _marsh;
+
+ /** Keep portable flag. */
+ private readonly bool _keepPortable;
+
+ /** Real listener. */
+ private readonly ICacheEntryEventListener<TK, TV> _lsnr;
+
+ /** Real filter. */
+ private readonly ICacheEntryEventFilter<TK, TV> _filter;
+
+ /** GC handle. */
+ private long _hnd;
+
+ /** Native query. */
+ private volatile IUnmanagedTarget _nativeQry;
+
+ /** Initial query cursor. */
+ private volatile IQueryCursor<ICacheEntry<TK, TV>> _initialQueryCursor;
+
+ /** Disposed flag. */
+ private bool _disposed;
+
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="qry">Query.</param>
+ /// <param name="marsh">Marshaller.</param>
+ /// <param name="keepPortable">Keep portable flag.</param>
+ public ContinuousQueryHandleImpl(ContinuousQuery<TK, TV> qry, PortableMarshaller marsh, bool keepPortable)
+ {
+ _marsh = marsh;
+ _keepPortable = keepPortable;
+
+ _lsnr = qry.Listener;
+ _filter = qry.Filter;
+ }
+
+ /// <summary>
+ /// Start execution.
+ /// </summary>
+ /// <param name="grid">Ignite instance.</param>
+ /// <param name="writer">Writer.</param>
+ /// <param name="cb">Callback invoked when all necessary data is written to stream.</param>
+ /// <param name="qry">Query.</param>
+ public void Start(Ignite grid, PortableWriterImpl writer, Func<IUnmanagedTarget> cb,
+ ContinuousQuery<TK, TV> qry)
+ {
+ // 1. Inject resources.
+ ResourceProcessor.Inject(_lsnr, grid);
+ ResourceProcessor.Inject(_filter, grid);
+
+ // 2. Allocate handle.
+ _hnd = grid.HandleRegistry.Allocate(this);
+
+ // 3. Write data to stream.
+ writer.WriteLong(_hnd);
+ writer.WriteBoolean(qry.Local);
+ writer.WriteBoolean(_filter != null);
+
+ ContinuousQueryFilterHolder filterHolder = _filter == null || qry.Local ? null :
+ new ContinuousQueryFilterHolder(typeof (TK), typeof (TV), _filter, _keepPortable);
+
+ writer.WriteObject(filterHolder);
+
+ writer.WriteInt(qry.BufferSize);
+ writer.WriteLong((long)qry.TimeInterval.TotalMilliseconds);
+ writer.WriteBoolean(qry.AutoUnsubscribe);
+
+ // 4. Call Java.
+ _nativeQry = cb();
+
+ // 5. Initial query.
+ var nativeInitialQryCur = UU.ContinuousQueryGetInitialQueryCursor(_nativeQry);
+ _initialQueryCursor = nativeInitialQryCur == null
+ ? null
+ : new QueryCursor<TK, TV>(nativeInitialQryCur, _marsh, _keepPortable);
+ }
+
+ /** <inheritdoc /> */
+ public void Apply(IPortableStream stream)
+ {
+ ICacheEntryEvent<TK, TV>[] evts = CQU.ReadEvents<TK, TV>(stream, _marsh, _keepPortable);
+
+ _lsnr.OnEvent(evts);
+ }
+
+ /** <inheritdoc /> */
+ public bool Evaluate(IPortableStream stream)
+ {
+ Debug.Assert(_filter != null, "Evaluate should not be called if filter is not set.");
+
+ ICacheEntryEvent<TK, TV> evt = CQU.ReadEvent<TK, TV>(stream, _marsh, _keepPortable);
+
+ return _filter.Evaluate(evt);
+ }
+
+ /** <inheritdoc /> */
+ public void Inject(Ignite grid)
+ {
+ throw new NotSupportedException("Should not be called.");
+ }
+
+ /** <inheritdoc /> */
+ public long Allocate()
+ {
+ throw new NotSupportedException("Should not be called.");
+ }
+
+ /** <inheritdoc /> */
+ public void Release()
+ {
+ _marsh.Ignite.HandleRegistry.Release(_hnd);
+ }
+
+ /** <inheritdoc /> */
+ public IQueryCursor<ICacheEntry<TK, TV>> GetInitialQueryCursor()
+ {
+ lock (this)
+ {
+ if (_disposed)
+ throw new ObjectDisposedException("Continuous query handle has been disposed.");
+
+ var cur = _initialQueryCursor;
+
+ if (cur == null)
+ throw new InvalidOperationException("GetInitialQueryCursor() can be called only once.");
+
+ _initialQueryCursor = null;
+
+ return cur;
+ }
+ }
+
+ /** <inheritdoc /> */
+ public void Dispose()
+ {
+ lock (this)
+ {
+ if (_disposed)
+ return;
+
+ Debug.Assert(_nativeQry != null);
+
+ try
+ {
+ UU.ContinuousQueryClose(_nativeQry);
+ }
+ finally
+ {
+ _nativeQry.Dispose();
+
+ _disposed = true;
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryUtils.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryUtils.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryUtils.cs
new file mode 100644
index 0000000..86c8300
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/Continuous/ContinuousQueryUtils.cs
@@ -0,0 +1,115 @@
+/*
+ * 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.Cache.Query.Continuous
+{
+ using System.Diagnostics;
+ using System.Diagnostics.CodeAnalysis;
+ using Apache.Ignite.Core.Cache.Event;
+ using Apache.Ignite.Core.Impl.Cache.Event;
+ using Apache.Ignite.Core.Impl.Portable;
+ using Apache.Ignite.Core.Impl.Portable.IO;
+
+ /// <summary>
+ /// Utility methods for continuous queries.
+ /// </summary>
+ static class ContinuousQueryUtils
+ {
+ /// <summary>
+ /// Read single event.
+ /// </summary>
+ /// <param name="stream">Stream to read data from.</param>
+ /// <param name="marsh">Marshaller.</param>
+ /// <param name="keepPortable">Keep portable flag.</param>
+ /// <returns>Event.</returns>
+ public static ICacheEntryEvent<TK, TV> ReadEvent<TK, TV>(IPortableStream stream,
+ PortableMarshaller marsh, bool keepPortable)
+ {
+ var reader = marsh.StartUnmarshal(stream, keepPortable);
+
+ return ReadEvent0<TK, TV>(reader);
+ }
+
+ /// <summary>
+ /// Read multiple events.
+ /// </summary>
+ /// <param name="stream">Stream.</param>
+ /// <param name="marsh">Marshaller.</param>
+ /// <param name="keepPortable">Keep portable flag.</param>
+ /// <returns>Events.</returns>
+ [SuppressMessage("ReSharper", "PossibleNullReferenceException")]
+ public static ICacheEntryEvent<TK, TV>[] ReadEvents<TK, TV>(IPortableStream stream,
+ PortableMarshaller marsh, bool keepPortable)
+ {
+ var reader = marsh.StartUnmarshal(stream, keepPortable);
+
+ int cnt = reader.ReadInt();
+
+ ICacheEntryEvent<TK, TV>[] evts = new ICacheEntryEvent<TK, TV>[cnt];
+
+ for (int i = 0; i < cnt; i++)
+ evts[i] = ReadEvent0<TK, TV>(reader);
+
+ return evts;
+ }
+
+ /// <summary>
+ /// Read event.
+ /// </summary>
+ /// <param name="reader">Reader.</param>
+ /// <returns>Event.</returns>
+ private static ICacheEntryEvent<TK, TV> ReadEvent0<TK, TV>(PortableReaderImpl reader)
+ {
+ reader.DetachNext();
+ TK key = reader.ReadObject<TK>();
+
+ reader.DetachNext();
+ TV oldVal = reader.ReadObject<TV>();
+
+ reader.DetachNext();
+ TV val = reader.ReadObject<TV>();
+
+ return CreateEvent(key, oldVal, val);
+ }
+
+ /// <summary>
+ /// Create event.
+ /// </summary>
+ /// <param name="key">Key.</param>
+ /// <param name="oldVal">Old value.</param>
+ /// <param name="val">Value.</param>
+ /// <returns>Event.</returns>
+ public static ICacheEntryEvent<TK, TV> CreateEvent<TK, TV>(TK key, TV oldVal, TV val)
+ {
+ if (oldVal == null)
+ {
+ Debug.Assert(val != null);
+
+ return new CacheEntryCreateEvent<TK, TV>(key, val);
+ }
+
+ if (val == null)
+ {
+ Debug.Assert(oldVal != null);
+
+ return new CacheEntryRemoveEvent<TK, TV>(key, oldVal);
+ }
+
+ return new CacheEntryUpdateEvent<TK, TV>(key, oldVal, val);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/FieldsQueryCursor.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/FieldsQueryCursor.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/FieldsQueryCursor.cs
new file mode 100644
index 0000000..f38346c
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/FieldsQueryCursor.cs
@@ -0,0 +1,54 @@
+/*
+ * 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.Cache.Query
+{
+ using System.Collections;
+ using Apache.Ignite.Core.Impl.Portable;
+ using Apache.Ignite.Core.Impl.Unmanaged;
+
+ /// <summary>
+ /// Cursor for entry-based queries.
+ /// </summary>
+ internal class FieldsQueryCursor : AbstractQueryCursor<IList>
+ {
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="target">Target.</param>
+ /// <param name="marsh">Marshaler.</param>
+ /// <param name="keepPortable">Keep poratble flag.</param>
+ public FieldsQueryCursor(IUnmanagedTarget target, PortableMarshaller marsh, bool keepPortable)
+ : base(target, marsh, keepPortable)
+ {
+ // No-op.
+ }
+
+ /** <inheritdoc /> */
+ protected override IList Read(PortableReaderImpl reader)
+ {
+ int cnt = reader.ReadInt();
+
+ var res = new ArrayList(cnt);
+
+ for (int i = 0; i < cnt; i++)
+ res.Add(reader.ReadObject<object>());
+
+ return res;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursor.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursor.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursor.cs
new file mode 100644
index 0000000..0b113f5
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Query/QueryCursor.cs
@@ -0,0 +1,50 @@
+/*
+ * 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.Cache.Query
+{
+ using Apache.Ignite.Core.Cache;
+ using Apache.Ignite.Core.Impl.Portable;
+ using Apache.Ignite.Core.Impl.Unmanaged;
+
+ /// <summary>
+ /// Cursor for entry-based queries.
+ /// </summary>
+ internal class QueryCursor<TK, TV> : AbstractQueryCursor<ICacheEntry<TK, TV>>
+ {
+ /// <summary>
+ /// Constructor.
+ /// </summary>
+ /// <param name="target">Target.</param>
+ /// <param name="marsh">Marshaler.</param>
+ /// <param name="keepPortable">Keep poratble flag.</param>
+ public QueryCursor(IUnmanagedTarget target, PortableMarshaller marsh,
+ bool keepPortable) : base(target, marsh, keepPortable)
+ {
+ // No-op.
+ }
+
+ /** <inheritdoc /> */
+ protected override ICacheEntry<TK, TV> Read(PortableReaderImpl reader)
+ {
+ TK key = reader.ReadObject<TK>();
+ TV val = reader.ReadObject<TV>();
+
+ return new CacheEntry<TK, TV>(key, val);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs
new file mode 100644
index 0000000..3fbc705
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs
@@ -0,0 +1,263 @@
+/*
+ * 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.Cache.Store
+{
+ using System.Collections;
+ using System.Diagnostics;
+ using Apache.Ignite.Core.Cache.Store;
+ using Apache.Ignite.Core.Common;
+ using Apache.Ignite.Core.Impl.Handle;
+ using Apache.Ignite.Core.Impl.Portable;
+ using Apache.Ignite.Core.Impl.Portable.IO;
+ using Apache.Ignite.Core.Impl.Resource;
+ using Apache.Ignite.Core.Impl.Unmanaged;
+ using Apache.Ignite.Core.Portable;
+
+ /// <summary>
+ /// Interop cache store.
+ /// </summary>
+ internal class CacheStore
+ {
+ /** */
+ private const byte OpLoadCache = 0;
+
+ /** */
+ private const byte OpLoad = 1;
+
+ /** */
+ private const byte OpLoadAll = 2;
+
+ /** */
+ private const byte OpPut = 3;
+
+ /** */
+ private const byte OpPutAll = 4;
+
+ /** */
+ private const byte OpRmv = 5;
+
+ /** */
+ private const byte OpRmvAll = 6;
+
+ /** */
+ private const byte OpSesEnd = 7;
+
+ /** */
+ private readonly bool _convertPortable;
+
+ /** Store. */
+ private readonly ICacheStore _store;
+
+ /** Session. */
+ private readonly CacheStoreSessionProxy _sesProxy;
+
+ /** */
+ private readonly long _handle;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="CacheStore" /> class.
+ /// </summary>
+ /// <param name="store">Store.</param>
+ /// <param name="convertPortable">Whether to convert portable objects.</param>
+ /// <param name="registry">The handle registry.</param>
+ private CacheStore(ICacheStore store, bool convertPortable, HandleRegistry registry)
+ {
+ Debug.Assert(store != null);
+
+ _store = store;
+ _convertPortable = convertPortable;
+
+ _sesProxy = new CacheStoreSessionProxy();
+
+ ResourceProcessor.InjectStoreSession(store, _sesProxy);
+
+ _handle = registry.AllocateCritical(this);
+ }
+
+ /// <summary>
+ /// Creates interop cache store from a stream.
+ /// </summary>
+ /// <param name="memPtr">Memory pointer.</param>
+ /// <param name="registry">The handle registry.</param>
+ /// <returns>
+ /// Interop cache store.
+ /// </returns>
+ internal static CacheStore CreateInstance(long memPtr, HandleRegistry registry)
+ {
+ using (var stream = IgniteManager.Memory.Get(memPtr).Stream())
+ {
+ var reader = PortableUtils.Marshaller.StartUnmarshal(stream, PortableMode.KeepPortable);
+
+ var assemblyName = reader.ReadString();
+ var className = reader.ReadString();
+ var convertPortable = reader.ReadBoolean();
+ var propertyMap = reader.ReadGenericDictionary<string, object>();
+
+ var store = (ICacheStore) IgniteUtils.CreateInstance(assemblyName, className);
+
+ IgniteUtils.SetProperties(store, propertyMap);
+
+ return new CacheStore(store, convertPortable, registry);
+ }
+ }
+
+ /// <summary>
+ /// Gets the handle.
+ /// </summary>
+ public long Handle
+ {
+ get { return _handle; }
+ }
+
+ /// <summary>
+ /// Initializes this instance with a grid.
+ /// </summary>
+ /// <param name="grid">Grid.</param>
+ public void Init(Ignite grid)
+ {
+ ResourceProcessor.Inject(_store, grid);
+ }
+
+ /// <summary>
+ /// Invokes a store operation.
+ /// </summary>
+ /// <param name="input">Input stream.</param>
+ /// <param name="cb">Callback.</param>
+ /// <param name="grid">Grid.</param>
+ /// <returns>Invocation result.</returns>
+ /// <exception cref="IgniteException">Invalid operation type: + opType</exception>
+ public int Invoke(IPortableStream input, IUnmanagedTarget cb, Ignite grid)
+ {
+ IPortableReader reader = grid.Marshaller.StartUnmarshal(input,
+ _convertPortable ? PortableMode.Deserialize : PortableMode.ForcePortable);
+
+ IPortableRawReader rawReader = reader.RawReader();
+
+ int opType = rawReader.ReadByte();
+
+ // Setup cache sessoin for this invocation.
+ long sesId = rawReader.ReadLong();
+
+ CacheStoreSession ses = grid.HandleRegistry.Get<CacheStoreSession>(sesId, true);
+
+ ses.CacheName = rawReader.ReadString();
+
+ _sesProxy.SetSession(ses);
+
+ try
+ {
+ // Perform operation.
+ switch (opType)
+ {
+ case OpLoadCache:
+ _store.LoadCache((k, v) => WriteObjects(cb, grid, k, v), rawReader.ReadObjectArray<object>());
+
+ break;
+
+ case OpLoad:
+ object val = _store.Load(rawReader.ReadObject<object>());
+
+ if (val != null)
+ WriteObjects(cb, grid, val);
+
+ break;
+
+ case OpLoadAll:
+ var keys = rawReader.ReadCollection();
+
+ var result = _store.LoadAll(keys);
+
+ foreach (DictionaryEntry entry in result)
+ WriteObjects(cb, grid, entry.Key, entry.Value);
+
+ break;
+
+ case OpPut:
+ _store.Write(rawReader.ReadObject<object>(), rawReader.ReadObject<object>());
+
+ break;
+
+ case OpPutAll:
+ _store.WriteAll(rawReader.ReadDictionary());
+
+ break;
+
+ case OpRmv:
+ _store.Delete(rawReader.ReadObject<object>());
+
+ break;
+
+ case OpRmvAll:
+ _store.DeleteAll(rawReader.ReadCollection());
+
+ break;
+
+ case OpSesEnd:
+ grid.HandleRegistry.Release(sesId);
+
+ _store.SessionEnd(rawReader.ReadBoolean());
+
+ break;
+
+ default:
+ throw new IgniteException("Invalid operation type: " + opType);
+ }
+
+ return 0;
+ }
+ finally
+ {
+ _sesProxy.ClearSession();
+ }
+ }
+
+ /// <summary>
+ /// Writes objects to the marshaller.
+ /// </summary>
+ /// <param name="cb">Optional callback.</param>
+ /// <param name="grid">Grid.</param>
+ /// <param name="objects">Objects.</param>
+ private static void WriteObjects(IUnmanagedTarget cb, Ignite grid, params object[] objects)
+ {
+ using (var stream = IgniteManager.Memory.Allocate().Stream())
+ {
+ PortableWriterImpl writer = grid.Marshaller.StartMarshal(stream);
+
+ try
+ {
+ foreach (var obj in objects)
+ {
+ writer.DetachNext();
+ writer.WriteObject(obj);
+ }
+ }
+ finally
+ {
+ grid.Marshaller.FinishMarshal(writer);
+ }
+
+ if (cb != null)
+ {
+ stream.SynchronizeOutput();
+
+ UnmanagedUtils.CacheStoreCallbackInvoke(cb, stream.MemoryPointer);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSession.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSession.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSession.cs
new file mode 100644
index 0000000..f771fe8
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSession.cs
@@ -0,0 +1,53 @@
+/*
+ * 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.Cache.Store
+{
+ using System.Collections.Generic;
+ using Apache.Ignite.Core.Cache.Store;
+
+ /// <summary>
+ /// Store session implementation.
+ /// </summary>
+ internal class CacheStoreSession : ICacheStoreSession
+ {
+ /** Properties. */
+ private IDictionary<object, object> _props;
+
+ /** <inheritdoc /> */
+
+ public string CacheName
+ {
+ get; internal set;
+ }
+
+ /** <inheritdoc /> */
+ public IDictionary<object, object> Properties
+ {
+ get { return _props ?? (_props = new Dictionary<object, object>(2)); }
+ }
+
+ /// <summary>
+ /// Clear session state.
+ /// </summary>
+ public void Clear()
+ {
+ if (_props != null)
+ _props.Clear();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f2eb16cd/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSessionProxy.cs
----------------------------------------------------------------------
diff --git a/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSessionProxy.cs b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSessionProxy.cs
new file mode 100644
index 0000000..3dd7354
--- /dev/null
+++ b/modules/platform/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStoreSessionProxy.cs
@@ -0,0 +1,63 @@
+/*
+ * 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.Cache.Store
+{
+ using System.Collections.Generic;
+ using System.Diagnostics.CodeAnalysis;
+ using System.Threading;
+ using Apache.Ignite.Core.Cache.Store;
+
+ /// <summary>
+ /// Store session proxy.
+ /// </summary>
+ [SuppressMessage("Microsoft.Design", "CA1001:TypesThatOwnDisposableFieldsShouldBeDisposable")]
+ internal class CacheStoreSessionProxy : ICacheStoreSession
+ {
+ /** Session. */
+ private readonly ThreadLocal<CacheStoreSession> _target = new ThreadLocal<CacheStoreSession>();
+
+ /** <inheritdoc /> */
+ public string CacheName
+ {
+ get { return _target.Value.CacheName; }
+ }
+
+ /** <inheritdoc /> */
+ public IDictionary<object, object> Properties
+ {
+ get { return _target.Value.Properties; }
+ }
+
+ /// <summary>
+ /// Set thread-bound session.
+ /// </summary>
+ /// <param name="ses">Session.</param>
+ internal void SetSession(CacheStoreSession ses)
+ {
+ _target.Value = ses;
+ }
+
+ /// <summary>
+ /// Clear thread-bound session.
+ /// </summary>
+ internal void ClearSession()
+ {
+ _target.Value = null;
+ }
+ }
+}