Cursor_1.cs

using System;
using System.Collections.Generic;
using System.IO;
using MongoDB.Connections;
using MongoDB.Protocol;
using MongoDB.Serialization;
using System.Linq;

namespace MongoDB
{
    /// <summary>
    /// 
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class Cursor<T> : ICursor<T> where T : class
    {
        private readonly Connection _connection;
        private readonly Document _specOpts = new Document();
        private bool _isModifiable = true;
        private object _spec;
        private object _fields;
        private int _limit;
        private QueryOptions _options;
        private ReplyMessage<T> _reply;
        private int _skip;
        private readonly ISerializationFactory _serializationFactory;

        /// <summary>
        /// Initializes a new instance of the <see cref="Cursor&lt;T&gt;"/> class.
        /// </summary>
        /// <param name="serializationFactory">The serialization factory.</param>
        /// <param name="connection">The conn.</param>
        /// <param name="fullCollectionName">Full name of the collection.</param>
        public Cursor(ISerializationFactory serializationFactory, Connection connection, string fullCollectionName)
        {
            //Todo: should be internal
            Id = -1;
            _connection = connection;
            FullCollectionName = fullCollectionName;
            _serializationFactory = serializationFactory;
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="Cursor&lt;T&gt;"/> class.
        /// </summary>
        /// <param name="serializationFactory">The serialization factory.</param>
        /// <param name="connection">The conn.</param>
        /// <param name="fullCollectionName">Full name of the collection.</param>
        /// <param name="spec">The spec.</param>
        /// <param name="limit">The limit.</param>
        /// <param name="skip">The skip.</param>
        /// <param name="fields">The fields.</param>
        public Cursor(ISerializationFactory serializationFactory, Connection connection, string fullCollectionName, object spec, int limit, int skip, object fields)
            : this(serializationFactory, connection, fullCollectionName)
        {
            //Todo: should be internal
            if (spec == null)
                spec = new Document();
            _spec = spec;
            _limit = limit;
            _skip = skip;
            _fields = fields;
        }

        /// <summary>
        /// Gets or sets the full name of the collection.
        /// </summary>
        /// <value>The full name of the collection.</value>
        public string FullCollectionName { get; private set; }

        /// <summary>
        /// Gets or sets the id.
        /// </summary>
        /// <value>The id.</value>
        public long Id { get; private set; }

        /// <summary>
        /// Specs the specified spec.
        /// </summary>
        /// <param name="spec">The spec.</param>
        /// <returns></returns>
        public ICursor<T> Spec(object spec){
            TryModify();
            _spec = spec;
            return this;
        }

        /// <summary>
        /// Limits the specified limit.
        /// </summary>
        /// <param name="limit">The limit.</param>
        /// <returns></returns>
        public ICursor<T> Limit(int limit){
            TryModify();
            _limit = limit;
            return this;
        }

        /// <summary>
        /// Skips the specified skip.
        /// </summary>
        /// <param name="skip">The skip.</param>
        /// <returns></returns>
        public ICursor<T> Skip(int skip){
            TryModify();
            _skip = skip;
            return this;
        }

        /// <summary>
        /// Fieldses the specified fields.
        /// </summary>
        /// <param name="fields">The fields.</param>
        /// <returns></returns>
        public ICursor<T> Fields(object fields){
            TryModify();
            _fields = fields;
            return this;
        }

        /// <summary>
        ///   Sorts the specified field.
        /// </summary>
        /// <param name = "field">The field.</param>
        /// <returns></returns>
        public ICursor<T> Sort(string field){
            return Sort(field, IndexOrder.Ascending);
        }

        /// <summary>
        ///   Sorts the specified field.
        /// </summary>
        /// <param name = "field">The field.</param>
        /// <param name = "order">The order.</param>
        /// <returns></returns>
        public ICursor<T> Sort(string field, IndexOrder order){
            return Sort(new Document().Add(field, order));
        }

        /// <summary>
        /// Sorts the specified fields.
        /// </summary>
        /// <param name="fields">The fields.</param>
        /// <returns></returns>
        public ICursor<T> Sort(object fields){
            TryModify();
            AddOrRemoveSpecOpt("$orderby", fields);
            return this;
        }

        /// <summary>
        /// Hints the specified index.
        /// </summary>
        /// <param name="index">The index.</param>
        /// <returns></returns>
        public ICursor<T> Hint(object index){
            TryModify();
            AddOrRemoveSpecOpt("$hint", index);
            return this;
        }

        /// <summary>
        /// Snapshots the specified index.
        /// </summary>
        public ICursor<T> Snapshot(){
            TryModify();
            AddOrRemoveSpecOpt("$snapshot", true);
            return this;
        }

        /// <summary>
        ///   Explains this instance.
        /// </summary>
        /// <returns></returns>
        public Document Explain(){
            //Fixme Return a single Document and not T
            TryModify();
            _specOpts["$explain"] = true;

            var explainResult = GetQueryRepley<Document>();
            try
            {
                var explain = explainResult.Documents.FirstOrDefault();

                if(explain==null)
                    throw new InvalidOperationException("Explain failed. No documents where returned.");

                return explain;
            }
            finally 
            {
                if(explainResult.CursorId > 0)
                    KillCursor(explainResult.CursorId);
            }
        }

        /// <summary>
        ///   Gets a value indicating whether this <see cref = "Cursor&lt;T&gt;" /> is modifiable.
        /// </summary>
        /// <value><c>true</c> if modifiable; otherwise, <c>false</c>.</value>
        public bool IsModifiable {
            get { return _isModifiable; }
        }

        /// <summary>
        ///   Gets the documents.
        /// </summary>
        /// <value>The documents.</value>
        public IEnumerable<T> Documents {
            get {
                if (_reply == null)
                    RetrieveData();
                if (_reply == null)
                    throw new InvalidOperationException("Expecting reply but get null");
                
                var documents = _reply.Documents;
                var documentCount = 0;
                var shouldBreak = false;
                
                while (!shouldBreak) {
                    foreach (var document in documents)
                        if ((_limit == 0) || (_limit != 0 && documentCount < _limit)) {
                            documentCount++;
                            yield return document;
                        } else
                            yield break;
                    
                    if (Id != 0) {
                        RetrieveMoreData();
                        documents = _reply.Documents;
                        if (documents == null)
                            shouldBreak = true;
                    } else
                        shouldBreak = true;
                }
            }
        }

        /// <summary>
        ///   Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        public void Dispose()
        {
            if (Id == 0)
                //All server side resources disposed of.
                return;

            KillCursor(Id);
        }

        /// <summary>
        /// Kills the cursor.
        /// </summary>
        private void KillCursor(long cursorId)
        {
            var killCursorsMessage = new KillCursorsMessage(cursorId);
            
            try {
                _connection.SendMessage(killCursorsMessage);
                Id = 0;
            } catch (IOException exception) {
                throw new MongoConnectionException("Could not read data, communication failure", _connection, exception);
            }
        }

        /// <summary>
        ///   Optionses the specified options.
        /// </summary>
        /// <param name = "options">The options.</param>
        /// <returns></returns>
        public ICursor<T> Options(QueryOptions options){
            TryModify();
            _options = options;
            return this;
        }

        /// <summary>
        /// Gets the query repley.
        /// </summary>
        /// <typeparam name="TReply">The type of the reply.</typeparam>
        /// <returns></returns>
        private ReplyMessage<TReply> GetQueryRepley<TReply>() where TReply : class
        {
            var writerSettings = _serializationFactory.GetBsonWriterSettings(typeof(T));

            var query = new QueryMessage(writerSettings)
            {
                FullCollectionName = FullCollectionName,
                Query = BuildSpec(),
                NumberToReturn = _limit,
                NumberToSkip = _skip,
                Options = _options
            };

            if(_fields != null)
                query.ReturnFieldSelector = _fields;

            var readerSettings = _serializationFactory.GetBsonReaderSettings(typeof(T));

            try
            {
                return _connection.SendTwoWayMessage<TReply>(query, readerSettings);
            }
            catch(IOException exception)
            {
                throw new MongoConnectionException("Could not read data, communication failure", _connection, exception);
            }
        }

        /// <summary>
        ///   Retrieves the data.
        /// </summary>
        private void RetrieveData(){
            _reply = GetQueryRepley<T>();
            
            Id = _reply.CursorId;
            
            if(_limit < 0)
                _limit = _limit * -1;
            
            _isModifiable = false;
        }

        /// <summary>
        ///   Retrieves the more data.
        /// </summary>
        private void RetrieveMoreData(){
            var getMoreMessage = new GetMoreMessage(FullCollectionName, Id, _limit);
            
            var readerSettings = _serializationFactory.GetBsonReaderSettings(typeof(T));
            
            try {
                _reply = _connection.SendTwoWayMessage<T>(getMoreMessage, readerSettings);
                Id = _reply.CursorId;
            } catch (IOException exception) {
                Id = 0;
                throw new MongoConnectionException("Could not read data, communication failure", _connection, exception);
            }
        }

        /// <summary>
        ///   Tries the modify.
        /// </summary>
        private void TryModify(){
            if (_isModifiable)
                return;
            throw new InvalidOperationException("Cannot modify a cursor that has already returned documents.");
        }

        /// <summary>
        ///   Adds the or remove spec opt.
        /// </summary>
        /// <param name = "key">The key.</param>
        /// <param name = "doc">The doc.</param>
        private void AddOrRemoveSpecOpt(string key, object doc){
            if (doc == null)
                _specOpts.Remove(key);
            else
                _specOpts[key] = doc;
        }

        /// <summary>
        ///   Builds the spec.
        /// </summary>
        /// <returns></returns>
        private object BuildSpec(){
            if (_specOpts.Count == 0)
                return _spec;
            
            var document = new Document();
            _specOpts.CopyTo(document);
            document["$query"] = _spec;
            return document;
        }
    }
}