Initial Commit ...

2024-01-25 04:41:17 +03:30
commit ff4d0d5dcb
51 changed files with 5949 additions and 0 deletions
@@ -0,0 +1,73 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using MongoDB.Driver;
+
+namespace xDataService.Mongo {
+    public class XAsyncEnumerableAdapter<T> : IAsyncEnumerable<T> {
+        //
+        private readonly IAsyncCursorSource<T> asyncCursorSource;
+
+        //
+        public XAsyncEnumerableAdapter (IAsyncCursorSource<T> asyncCursorSource) {
+            this.asyncCursorSource = asyncCursorSource;
+        }
+
+        //
+        public IAsyncEnumerator<T> GetAsyncEnumerator (CancellationToken cancellationToken) =>
+        new XAsyncEnumeratorAdapter<T> (asyncCursorSource, cancellationToken);
+    }
+
+    public class XAsyncEnumeratorAdapter<T> : IAsyncEnumerator<T> {
+        private readonly IAsyncCursorSource<T> asyncCursorSource;
+        private readonly CancellationToken cancellationToken;
+        private IAsyncCursor<T> asyncCursor;
+        private IEnumerator<T> batchEnumerator;
+
+        public T Current => batchEnumerator.Current;
+
+        public XAsyncEnumeratorAdapter (
+            IAsyncCursorSource<T> asyncCursorSource,
+            CancellationToken cancellationToken
+        ) {
+            //
+            this.asyncCursorSource = asyncCursorSource;
+            this.cancellationToken = cancellationToken;
+        }
+
+        public async ValueTask<bool> MoveNextAsync () {
+            //
+            if (asyncCursor == null) {
+                asyncCursor = await asyncCursorSource
+                    .ToCursorAsync (cancellationToken);
+            }
+
+            //
+            if (batchEnumerator != null &&
+                batchEnumerator.MoveNext ()) {
+                return true;
+            }
+
+            //
+            if (asyncCursor != null &&
+                await asyncCursor.MoveNextAsync (cancellationToken)) {
+                //
+                batchEnumerator?.Dispose ();
+                batchEnumerator = asyncCursor.Current.GetEnumerator ();
+
+                //
+                return batchEnumerator.MoveNext ();
+            }
+
+            //
+            return false;
+        }
+
+        public async ValueTask DisposeAsync () {
+            //
+            await Task.CompletedTask;
+            asyncCursor?.Dispose ();
+            asyncCursor = null;
+        }
+    }
+}
@@ -0,0 +1,31 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using MongoDB.Driver;
+
+namespace xDataService.Mongo {
+    public class XMongoAsyncCursor<T> : IAsyncCursor<T> {
+        public IEnumerable<T> Current { get; }
+        private IEnumerator<T> CurrentEnumerator { get; }
+
+        public XMongoAsyncCursor (IEnumerable<T> source) {
+            this.Current = source;
+            this.CurrentEnumerator = source.GetEnumerator ();
+        }
+
+        public void Dispose () {
+            CurrentEnumerator.Dispose ();
+        }
+
+        public bool MoveNext (CancellationToken cancellationToken = default) {
+            return CurrentEnumerator.MoveNext ();
+        }
+
+        public Task<bool> MoveNextAsync (CancellationToken cancellationToken = default) {
+            return Task.Run (
+                () => CurrentEnumerator.MoveNext (),
+                cancellationToken : cancellationToken
+            );
+        }
+    }
+}
@@ -0,0 +1,56 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+using System.Linq.Expressions;
+using System.Threading;
+using System.Threading.Tasks;
+using MongoDB.Driver;
+using MongoDB.Driver.Linq;
+using xModels.Base;
+
+namespace xDataService.Mongo {
+    public class XMongoQueryable<TEntity, TKey> : IMongoQueryable<TEntity>
+        where TEntity : XBaseEntity<TKey> {
+            private readonly IMongoCollection<TEntity> collection;
+
+            public Type ElementType { get; }
+            public IQueryable<TEntity> Items { get; }
+            public Expression Expression { get; }
+            public IQueryProvider Provider { get; }
+
+            public XMongoQueryable (
+                IQueryable<TEntity> items,
+                IMongoCollection<TEntity> collection
+            ) {
+                //
+                this.Items = items;
+                this.collection = collection;
+
+                //
+                Provider = items.Provider;
+                Expression = items.Expression;
+                ElementType = items.ElementType;
+            }
+
+            public IEnumerator<TEntity> GetEnumerator () {
+                return Items.GetEnumerator ();
+            }
+
+            public QueryableExecutionModel GetExecutionModel () {
+                return collection.AsQueryable ().GetExecutionModel ();
+            }
+
+            public IAsyncCursor<TEntity> ToCursor (CancellationToken cancellationToken = default) {
+                return new XMongoAsyncCursor<TEntity> (Items);
+            }
+
+            public Task<IAsyncCursor<TEntity>> ToCursorAsync (CancellationToken cancellationToken = default) {
+                return Task.Run (() => new XMongoAsyncCursor<TEntity> (Items) as IAsyncCursor<TEntity>);
+            }
+
+            IEnumerator IEnumerable.GetEnumerator () {
+                return Items.GetEnumerator ();
+            }
+        }
+}
@@ -0,0 +1,23 @@
+using MongoDB.Driver;
+
+namespace xDataService.Mongo {
+    public static class XMongoReadPreferenceResolver {
+        public static ReadPreference GetEffectiveReadPreference (
+            IClientSessionHandle session,
+            ReadPreference explicitReadPreference,
+            ReadPreference defaultReadPreference) {
+            if (explicitReadPreference != null) {
+                return explicitReadPreference;
+            }
+
+            if (session.IsInTransaction) {
+                var transactionReadPreference = session.WrappedCoreSession.CurrentTransaction.TransactionOptions.ReadPreference;
+                if (transactionReadPreference != null) {
+                    return transactionReadPreference;
+                }
+            }
+
+            return defaultReadPreference ?? ReadPreference.Primary;
+        }
+    }
+}