public IObservable<ParkingArea> Extract(IMongoQuery query)
        {
            return Observable.Create<ParkingArea>(async observer =>
            {
                var totalEntitiesToProcess  = serverProvider.Database.OsmCompoundQueryCount(query);
                var operationCounter        = new OperationCounter(log, string.Format("OSM Parking Area Extraction from {0}",query), totalEntitiesToProcess);
                var cursor                  = serverProvider.Database.OsmCompoundQuery(query);

                foreach (var entity in cursor)
                {
                    var entityContext   = new OsmEntityContext(serverProvider.Database, entity);
                    var parkingArea     = new ParkingArea() { Id = Guid.NewGuid(), Source = ParkingAreaSource.Maps, Confidence = configuration.Model.Confidence };
                    var isValid         = true;

                    foreach (var extractionRule in extractionRules)
                    {
                        var continueExtraction = await extractionRule.Extract(entityContext, parkingArea);

                        if (!continueExtraction)
                        {
                            log.Info("Entity {0} was dropped from extraction by rule {1}", entity, extractionRule.GetType().Name.Replace("Extraction", string.Empty).Replace("Rule", string.Empty));
                            isValid = false;
                            break;
                        }
                    }

                    if (isValid)
                    {
                        observer.OnNext(parkingArea);
                    }

                    operationCounter.Increment();
                }
            });
        }

        public static List<供应商> 抽选供应商(int count, IMongoQuery conditions, IEnumerable<long> selected, IEnumerable<long> avoid)
        {
            var ret = new List<供应商>(count);
            var rnd = new Random();

            conditions = conditions.And(Query<供应商>.EQ(o => o.审核数据.审核状态, 审核状态.审核通过));
            if (null != selected && 0 != selected.Count())
            {
                var q = Query.NotIn("_id", new BsonArray(selected));
                conditions = Query.And(conditions, q);
            }
            if (null != avoid && 0 != avoid.Count())
            {
                var q = Query.NotIn("_id", new BsonArray(avoid));
                conditions = Query.And(conditions, q);
            }
            int total = (int)Mongo.计数<供应商>(0, 0, conditions);
            if (total < count) return new List<供应商>();
            var r = Mongo.查询<供应商>(0, 0, conditions);
            var ns = new HashSet<int>();
            for (int i = 0; i < count; i++)
            {
                int n;
                do
                {
                    n = rnd.Next(total);
                } while (ns.Contains(n));
                ret.Add(r.ElementAt(n));
                ns.Add(n);
            }
            return ret;
        }

 // constructors
 internal MongoDeleteMessage(BsonBinaryWriterSettings writerSettings, string collectionFullName, RemoveFlags flags, IMongoQuery query)
     : base(MessageOpcode.Delete, null, writerSettings)
 {
     _collectionFullName = collectionFullName;
     _flags = flags;
     _query = query;
 }

        public IEnumerable<KeyValuePair<string, IEnumerable<string>>> GetAllTags(IMongoQuery query)
        {
            var dataSet = new Dictionary<string, HashSet<string>>(StringComparer.OrdinalIgnoreCase);

            foreach (var entity in serverProvider.Database.OsmCompoundQuery(query))
            {
                foreach (var tag in entity.Tags)
                {
                    if (!dataSet.ContainsKey(tag.Key))
                    {
                        dataSet.Add(tag.Key, new HashSet<string>(StringComparer.OrdinalIgnoreCase));
                    }

                    var valueSet = dataSet[tag.Key];
                    if (!valueSet.Contains(tag.Value))
                    {
                        valueSet.Add(tag.Value);
                    }
                }
            }

            var result = dataSet.Select(kv => new KeyValuePair<string, IEnumerable<string>>(kv.Key, kv.Value));

            return result;
        }

        public bool Remove(IMongoQuery query)
        {
            if (query == null)
                throw new ArgumentNullException("Collection must not be null");

            return _connection.GetCurrentCollection().Remove(query).Ok;
        }

        public static long OsmCompoundQueryCount(this MongoDatabase database,IMongoQuery query)
        {
            var r1 = database.GetOsmCollection(OsmEntityType.Node).FindAs<OsmNode>(query);
            var r2 = database.GetOsmCollection(OsmEntityType.Way).FindAs<OsmWay>(query);
            var r3 = database.GetOsmCollection(OsmEntityType.Relation).FindAs<OsmRelation>(query);

            return r1.Count() + r2.Count() + r3.Count();
        }

 public static IEnumerable<新闻> 查询图片新闻(int skip, int limit, IMongoQuery conditions = null)
 {
     var q = Query.SizeGreaterThan("内容主体.图片", 0);
     if (conditions != null)
     {
         q = q.And(conditions);
     }
     return Mongo.查询<新闻>(skip, limit,q , true, null, false, false);
 }

 // constructors
 internal MongoUpdateMessage(BsonBinaryWriterSettings writerSettings, string collectionFullName, bool checkUpdateDocument, UpdateFlags flags, IMongoQuery query, IMongoUpdate update)
     : base(MessageOpcode.Update, null, writerSettings)
 {
     this.collectionFullName = collectionFullName;
     this.checkUpdateDocument = checkUpdateDocument;
     this.flags = flags;
     this.query = query;
     this.update = update;
 }

 internal MongoQueryMessage(BsonBuffer buffer, BsonBinaryWriterSettings writerSettings, string collectionFullName, QueryFlags flags, int numberToSkip, int numberToReturn, IMongoQuery query, IMongoFields fields)
     : base(MessageOpcode.Query, buffer, writerSettings)
 {
     this.collectionFullName = collectionFullName;
     this.flags = flags;
     this.numberToSkip = numberToSkip;
     this.numberToReturn = numberToReturn;
     this.query = query;
     this.fields = fields;
 }

        public static IMongoQuery GeoQuery(MongoLocation loc, double prox, IMongoQuery query)
        {
            var locQuery = QueryConstants.GeoQuery(loc, prox);

            var q = new QueryDocument();
            q.AddRange(BsonDocument.Parse(locQuery.ToString()));
            if (query != null) q.AddRange(BsonDocument.Parse(query.ToString()));

            return q;
        }

        private ICollection<KeyValuePair<string, int>> GetPageViewsByUserAgent(IMongoQuery query)
        {
            var reduce = new BsonJavaScript("function(o, agg) { agg.count++; }");
            var results = DB.PageViews.Group(query, "UserAgent", new { count = 0 }.ToBsonDocument(), reduce, null).ToList();

            return results
                .ToKeyValuePairs(x => x["UserAgent"].AsString, x => x["count"].ToInt32())
                .OrderByDescending(kvp => kvp.Value)
                .ToList();
        }

 // constructors
 internal MongoQueryMessage(
     BsonBinaryWriterSettings writerSettings,
     string collectionFullName,
     QueryFlags flags,
     int numberToSkip,
     int numberToReturn,
     IMongoQuery query,
     IMongoFields fields)
     : this(null, writerSettings, collectionFullName, flags, numberToSkip, numberToReturn, query, fields)
 {
 }

 internal MongoQueryMessage(
     MongoServer server,
     string collectionFullName,
     QueryFlags flags,
     int numberToSkip,
     int numberToReturn,
     IMongoQuery query,
     IMongoFields fields
 ) :
     this(server, collectionFullName, flags, numberToSkip, numberToReturn, query, fields, null) {
 }

 public void SetQuery(IMongoQuery mongoQuery)
 {
     if (this.MongoQuery != null)
     {
         this.MongoQuery = QueryBuilder.Query.And(this.MongoQuery, mongoQuery);
     }
     else
     {
         this.MongoQuery = mongoQuery;
     }
 }

 internal MongoDeleteMessage(
     MongoServer server,
     string collectionFullName,
     RemoveFlags flags,
     IMongoQuery query
 ) :
     base(server, MessageOpcode.Delete) {
     this.collectionFullName = collectionFullName;
     this.flags = flags;
     this.query = query;
 }

        public static IEnumerable<OsmEntity> OsmCompoundQuery(this MongoDatabase database,IMongoQuery query)
        {
            var r1 = database.GetOsmCollection(OsmEntityType.Node).FindAs<OsmNode>(query);
            var r2 = database.GetOsmCollection(OsmEntityType.Way).FindAs<OsmWay>(query);
            var r3 = database.GetOsmCollection(OsmEntityType.Relation).FindAs<OsmRelation>(query);

            PrepareCursorForBatchProcessing(r1);
            PrepareCursorForBatchProcessing(r2);
            PrepareCursorForBatchProcessing(r3);

            return r1.Cast<OsmEntity>().Concat(r2).Concat(r3);
        }

 public RemoveOperation(
     string databaseName,
     string collectionName,
     BsonBinaryReaderSettings readerSettings,
     BsonBinaryWriterSettings writerSettings,
     WriteConcern writeConcern,
     IMongoQuery query,
     RemoveFlags flags)
     : base(databaseName, collectionName, readerSettings, writerSettings, writeConcern)
 {
     _query = query;
     _flags = flags;
 }

 internal MongoUpdateMessage(
     MongoConnection connection,
     string collectionFullName,
     UpdateFlags flags,
     IMongoQuery query,
     IMongoUpdate update
 ) :
     base(connection, MessageOpcode.Update) {
     this.collectionFullName = collectionFullName;
     this.flags = flags;
     this.query = query;
     this.update = update;
 }

        protected IMongoQuery WrapQuery(IMongoQuery query, BsonDocument options, ReadPreference readPreference, bool forShardRouter)
        {
            BsonDocument formattedReadPreference = null;
            if (forShardRouter && readPreference != null && readPreference.ReadPreferenceMode != ReadPreferenceMode.Primary)
            {
                BsonArray tagSetsArray = null;
                if (readPreference.TagSets != null)
                {
                    tagSetsArray = new BsonArray();
                    foreach (var tagSet in readPreference.TagSets)
                    {
                        var tagSetDocument = new BsonDocument();
                        foreach (var tag in tagSet)
                        {
                            tagSetDocument.Add(tag.Name, tag.Value);
                        }
                        tagSetsArray.Add(tagSetDocument);
                    }
                }

                if (tagSetsArray != null || readPreference.ReadPreferenceMode != ReadPreferenceMode.SecondaryPreferred)
                {
                    formattedReadPreference = new BsonDocument
                    {
                        { "mode", MongoUtils.ToCamelCase(readPreference.ReadPreferenceMode.ToString()) },
                        { "tags", tagSetsArray, tagSetsArray != null } // optional
                    };
                }
            }

            if (options == null && formattedReadPreference == null)
            {
                return query;
            }
            else
            {
                var queryDocument = (query == null) ? (BsonValue)new BsonDocument() : BsonDocumentWrapper.Create(query);
                var wrappedQuery = new QueryDocument
                {
                    { "$query", queryDocument },
                    { "$readPreference", formattedReadPreference, formattedReadPreference != null }, // only if sending query to a mongos
                };
                if (options != null)
                {
                    wrappedQuery.Merge(options);
                }
                return wrappedQuery;
            }
        }

        public static List<专家> 抽选专家(int count, IMongoQuery conditions, IEnumerable<long> selected, IEnumerable<long> avoid, string 所属单位)
        {
            var ret = new List<专家>(count);
            var rnd = new Random();

            var now = DateTime.Now;
            //筛选条件，专家3个月内不能被同一单位抽取，一年内不能被同一单位抽取3次
            var querydate = Query.Where(
                        " function(){" +
                        " var count = obj.历史抽取信息." + 所属单位 + ".length;" +
                        " if (count == 0) return true;" +
                        " if (count >0 && count <3 && new Date(obj.历史抽取信息." + 所属单位 + "[count-1]) < new Date(\"" + now.AddMonths(-3).ToString() + "\")) return true;" +
                        " if (count >=3 &&  new Date(obj.历史抽取信息." + 所属单位 + "[count-1]) < new Date(\"" + now.AddMonths(-3).ToString() + "\") && new Date(obj.历史抽取信息." + 所属单位 + "[count-3]) < new Date(\"" + now.AddYears(-1).ToString() + "\")) return true;" +
                        " return false;" +
                        " }");
            var q = Query<专家>.EQ(o => o.审核数据.审核状态, 审核状态.审核通过).And(Query.NotExists("历史抽取信息." + 所属单位).Or(querydate));
            if (null != conditions)
                conditions = Query.And(conditions, q);
            else
                conditions = q;

            if (null != selected && 0 != selected.Count())
            {
                q = Query.NotIn("_id", new BsonArray(selected));
                conditions = Query.And(conditions, q);
            }
            if (null != avoid && 0 != avoid.Count())
            {
                q = Query.NotIn("_id", new BsonArray(avoid));
                conditions = Query.And(conditions, q);
            }
            int total = (int)Mongo.计数<专家>(0, 0, conditions);
            if (total < count) return new List<专家>();
            var r = Mongo.查询<专家>(0, 0, conditions);
            var ns = new HashSet<int>();
            for (int i = 0; i < count; i++)
            {
                int n;
                do
                {
                    n = rnd.Next(total);
                } while (ns.Contains(n));
                ret.Add(r.ElementAt(n));
                ns.Add(n);
            }
            return ret;
        }