SDP（10）：文本式大数据运算环境-MongoDB-Engine功能设计详解编程语言

/** 
 * An interface for types that are able to render themselves into a [email protected] BsonDocument}. 
 * 
 * @since 3.0 
 */ 
public interface Bson { 
    /** 
     * Render the filter into a BsonDocument. 
     * 
     * @param documentClass the document class in scope for the collection.  This parameter may be ignored, but it may be used to alter 
     *                      the structure of the returned [email protected] BsonDocument} based on some knowledge of the document class. 
     * @param codecRegistry the codec registry.  This parameter may be ignored, but it may be used to look up [email protected] Codec} instances for 
     *                      the document class or any other related class. 
     * @param <TDocument> the type of the document class 
     * @return the BsonDocument 
     */ 
    <TDocument> BsonDocument toBsonDocument(Class<TDocument> documentClass, CodecRegistry codecRegistry); 
}

   private static final class OperatorFilter<TItem> implements Bson { 
        private final String operatorName; 
        private final String fieldName; 
        private final TItem value; 
 
        OperatorFilter(final String operatorName, final String fieldName, final TItem value) { 
            this.operatorName = notNull("operatorName", operatorName); 
            this.fieldName = notNull("fieldName", fieldName); 
            this.value = value; 
        } 
 
        @Override 
        public <TDocument> BsonDocument toBsonDocument(final Class<TDocument> documentClass, final CodecRegistry codecRegistry) { 
            BsonDocumentWriter writer = new BsonDocumentWriter(new BsonDocument()); 
 
            writer.writeStartDocument(); 
            writer.writeName(fieldName); 
            writer.writeStartDocument(); 
            writer.writeName(operatorName); 
            encodeValue(writer, value, codecRegistry); 
            writer.writeEndDocument(); 
            writer.writeEndDocument(); 
 
            return writer.getDocument(); 
        } 
 
        @Override 
        public String toString() { 
            return operatorFilterToString(fieldName, operatorName, value); 
        } 
    }

count 
disctinct 
aggregate 
mapReduce 
bulkWrite 
insert 
delete 
replace 
update

  count: 
   /** 
   * Counts the number of documents in the collection. 
   * @return a Observable with a single element indicating the number of documents 
   */ 
  def count(): SingleObservable[Long] = observeLong(wrapped.count(_: SingleResultCallback[java.lang.Long])) 
 
  /** 
   * Counts the number of documents in the collection according to the given options. 
   * @param filter  the query filter 
   * @param options the options describing the count 
   * @return a Observable with a single element indicating the number of documents 
   */ 
  def count(filter: Bson, options: CountOptions): SingleObservable[Long] = 
    observeLong(wrapped.count(filter, options, _: SingleResultCallback[java.lang.Long])) 
 
  distinct: 
    /** 
   * Gets the distinct values of the specified field name. 
   * @param fieldName the field name 
   * @tparam C       the target type of the observable. 
   * @return a Observable emitting the sequence of distinct values 
   */ 
  def distinct[C](fieldName: String)(implicit ct: ClassTag[C]): DistinctObservable[C] = 
    DistinctObservable(wrapped.distinct(fieldName, ct)) 
 
  /** 
   * Gets the distinct values of the specified field name. 
   * @param fieldName the field name 
   * @param filter  the query filter 
   * @tparam C       the target type of the observable. 
   * @return a Observable emitting the sequence of distinct values 
   */ 
  def distinct[C](fieldName: String, filter: Bson)(implicit ct: ClassTag[C]): DistinctObservable[C] = 
    DistinctObservable(wrapped.distinct(fieldName, filter, ct)) 
   
  find: 
    /** 
   * Finds all documents in the collection. 
   * @tparam C   the target document type of the observable. 
   * @return the find Observable 
   */ 
  def find[C]()(implicit e: C DefaultsTo TResult, ct: ClassTag[C]): FindObservable[C] = 
    FindObservable(wrapped.find[C](ct)) 
 
  /** 
   * Finds all documents in the collection. 
   * @param filter the query filter 
   * @tparam C    the target document type of the observable. 
   * @return the find Observable 
   */ 
  def find[C](filter: Bson)(implicit e: C DefaultsTo TResult, ct: ClassTag[C]): FindObservable[C] = 
    FindObservable(wrapped.find(filter, ct)) 
 
  aggregate: 
   /** 
   * Aggregates documents according to the specified aggregation pipeline. 
   * @param pipeline the aggregate pipeline 
   * @return a Observable containing the result of the aggregation operation 
   *         [[http://docs.mongodb.org/manual/aggregation/ Aggregation]] 
   */ 
  def aggregate[C](pipeline: Seq[Bson])(implicit e: C DefaultsTo TResult, ct: ClassTag[C]): AggregateObservable[C] = 
    AggregateObservable(wrapped.aggregate[C](pipeline.asJava, ct)) 
 
  mapReduce: 
    /** 
   * Aggregates documents according to the specified map-reduce function. 
   * @param mapFunction    A JavaScript function that associates or "maps" a value with a key and emits the key and value pair. 
   * @param reduceFunction A JavaScript function that "reduces" to a single object all the values associated with a particular key. 
   * @tparam C            the target document type of the observable. 
   * @return a Observable containing the result of the map-reduce operation 
   */ 
  def mapReduce[C](mapFunction: String, reduceFunction: String)(implicit e: C DefaultsTo TResult, ct: ClassTag[C]): MapReduceObservable[C] = 
    MapReduceObservable(wrapped.mapReduce(mapFunction, reduceFunction, ct)) 
 
  bulkWrite: 
    /** 
   * Executes a mix of inserts, updates, replaces, and deletes. 
   * @param requests the writes to execute 
   * @return a Observable with a single element the BulkWriteResult 
   */ 
  def bulkWrite(requests: Seq[_ <: WriteModel[_ <: TResult]]): SingleObservable[BulkWriteResult] = 
    observe(wrapped.bulkWrite( 
      requests.asJava.asInstanceOf[util.List[_ <: WriteModel[_ <: TResult]]], 
      _: SingleResultCallback[BulkWriteResult] 
    )) 
 
  /** 
   * Executes a mix of inserts, updates, replaces, and deletes. 
   * @param requests the writes to execute 
   * @param options  the options to apply to the bulk write operation 
   * @return a Observable with a single element the BulkWriteResult 
   */ 
  def bulkWrite(requests: Seq[_ <: WriteModel[_ <: TResult]], options: BulkWriteOptions): SingleObservable[BulkWriteResult] = 
    observe(wrapped.bulkWrite( 
      requests.asJava.asInstanceOf[util.List[_ <: WriteModel[_ <: TResult]]], 
      options, 
      _: SingleResultCallback[BulkWriteResult] 
    )) 
   
  insert: 
   /** 
   * Inserts the provided document. If the document is missing an identifier, the driver should generate one. 
   * @param document the document to insert 
   * @return a Observable with a single element indicating when the operation has completed or with either a 
   *         com.mongodb.DuplicateKeyException or com.mongodb.MongoException 
   */ 
  def insertOne(document: TResult): SingleObservable[Completed] = observeCompleted(wrapped.insertOne(document, _: SingleResultCallback[Void])) 
 
  /** 
   * Inserts the provided document. If the document is missing an identifier, the driver should generate one. 
   * @param document the document to insert 
   * @param options  the options to apply to the operation 
   * @return a Observable with a single element indicating when the operation has completed or with either a 
   *         com.mongodb.DuplicateKeyException or com.mongodb.MongoException 
   */ 
  def insertOne(document: TResult, options: InsertOneOptions): SingleObservable[Completed] = 
    observeCompleted(wrapped.insertOne(document, options, _: SingleResultCallback[Void])) 
 
  /** 
   * Inserts a batch of documents. The preferred way to perform bulk inserts is to use the BulkWrite API. However, when talking with  
   * server 2.6, using this method will be faster due to constraints in the bulk API related to error handling. 
   * @param documents the documents to insert 
   * @return a Observable with a single element indicating when the operation has completed or with either a 
   *         com.mongodb.DuplicateKeyException or com.mongodb.MongoException 
   */ 
  def insertMany(documents: Seq[_ <: TResult]): SingleObservable[Completed] = 
    observeCompleted(wrapped.insertMany(documents.asJava, _: SingleResultCallback[Void])) 
 
  /** 
   * Inserts a batch of documents. The preferred way to perform bulk inserts is to use the BulkWrite API. However, when talking with  
   * server 2.6, using this method will be faster due to constraints in the bulk API related to error handling. 
   * @param documents the documents to insert 
   * @param options   the options to apply to the operation 
   * @return a Observable with a single element indicating when the operation has completed or with either a 
   *         com.mongodb.DuplicateKeyException or com.mongodb.MongoException 
   */ 
  def insertMany(documents: Seq[_ <: TResult], options: InsertManyOptions): SingleObservable[Completed] = 
    observeCompleted(wrapped.insertMany(documents.asJava, options, _: SingleResultCallback[Void])) 
 
  delete: 
   /** 
   * Removes at most one document from the collection that matches the given filter.  If no documents match, the collection is not 
   * modified. 
   * @param filter the query filter to apply the the delete operation 
   * @return a Observable with a single element the DeleteResult or with an com.mongodb.MongoException 
   */ 
  def deleteOne(filter: Bson): SingleObservable[DeleteResult] = observe(wrapped.deleteOne(filter, _: SingleResultCallback[DeleteResult])) 
 
  /** 
   * Removes at most one document from the collection that matches the given filter.  If no documents match, the collection is not 
   * modified. 
   * @param filter the query filter to apply the the delete operation 
   * @param options the options to apply to the delete operation 
   * @return a Observable with a single element the DeleteResult or with an com.mongodb.MongoException 
   */ 
  def deleteOne(filter: Bson, options: DeleteOptions): SingleObservable[DeleteResult] = 
    observe(wrapped.deleteOne(filter, options, _: SingleResultCallback[DeleteResult])) 
 
  /** 
   * Removes all documents from the collection that match the given query filter.  If no documents match, the collection is not modified. 
   * @param filter the query filter to apply the the delete operation 
   * @return a Observable with a single element the DeleteResult or with an com.mongodb.MongoException 
   */ 
  def deleteMany(filter: Bson): SingleObservable[DeleteResult] = observe(wrapped.deleteMany(filter, _: SingleResultCallback[DeleteResult])) 
 
  /** 
   * Removes all documents from the collection that match the given query filter.  If no documents match, the collection is not modified. 
   * @param filter the query filter to apply the the delete operation 
   * @param options the options to apply to the delete operation 
   * @return a Observable with a single element the DeleteResult or with an com.mongodb.MongoException 
   */ 
  def deleteMany(filter: Bson, options: DeleteOptions): SingleObservable[DeleteResult] = 
    observe(wrapped.deleteMany(filter, options, _: SingleResultCallback[DeleteResult])) 
  
  replace: 
   /** 
   * Replace a document in the collection according to the specified arguments. 
   * [[http://docs.mongodb.org/manual/tutorial/modify-documents/#replace-the-document Replace]] 
   * @param filter      the query filter to apply the the replace operation 
   * @param replacement the replacement document 
   * @return a Observable with a single element the UpdateResult 
   */ 
  def replaceOne(filter: Bson, replacement: TResult): SingleObservable[UpdateResult] = 
    observe(wrapped.replaceOne(filter, replacement, _: SingleResultCallback[UpdateResult])) 
 
  /** 
   * Replace a document in the collection according to the specified arguments. 
   * @param filter      the query filter to apply the the replace operation 
   * @param replacement the replacement document 
   * @param options     the options to apply to the replace operation 
   * @return a Observable with a single element the UpdateResult 
   */ 
  def replaceOne(filter: Bson, replacement: TResult, options: UpdateOptions): SingleObservable[UpdateResult] = 
    observe(wrapped.replaceOne(filter, replacement, options, _: SingleResultCallback[UpdateResult])) 
 
  update: 
   /** 
   * Update a single document in the collection according to the specified arguments. 
   * @param filter  a document describing the query filter, which may not be null. This can be of any type for which a `Codec` is 
   *                registered 
   * @param update  a document describing the update, which may not be null. The update to apply must include only update operators. This 
   *                can be of any type for which a `Codec` is registered 
   * @return a Observable with a single element the UpdateResult 
   */ 
  def updateOne(filter: Bson, update: Bson): SingleObservable[UpdateResult] = 
    observe(wrapped.updateOne(filter, update, _: SingleResultCallback[UpdateResult])) 
 
  /** 
   * Update a single document in the collection according to the specified arguments. 
   * @param filter  a document describing the query filter, which may not be null. This can be of any type for which a `Codec` is 
   *                registered 
   * @param update  a document describing the update, which may not be null. The update to apply must include only update operators. This 
   *                can be of any type for which a `Codec` is registered 
   * @param options the options to apply to the update operation 
   * @return a Observable with a single element the UpdateResult 
   */ 
  def updateOne(filter: Bson, update: Bson, options: UpdateOptions): SingleObservable[UpdateResult] = 
    observe(wrapped.updateOne(filter, update, options, _: SingleResultCallback[UpdateResult])) 
  /** 
   * Update documents in the collection according to the specified arguments. 
   * @param filter  a document describing the query filter, which may not be null. This can be of any type for which a `Codec` is 
   *                registered 
   * @param update  a document describing the update, which may not be null. The update to apply must include only update operators. This 
   *                can be of any type for which a `Codec` is registered 
   * @return a Observable with a single element the UpdateResult 
   */ 
  def updateMany(filter: Bson, update: Bson): SingleObservable[UpdateResult] = 
    observe(wrapped.updateMany(filter, update, _: SingleResultCallback[UpdateResult]))

  createCollection 
  listCollection 
  dropCollection 
  createIndex 
  dropIndex 
  createView

dropCollection 
 /** 
   * Drops this collection from the Database. 
   */ 
  def drop(): SingleObservable[Completed] = observeCompleted(wrapped.drop(_: SingleResultCallback[Void])) 
 
createIndex 
  /** 
   * @param key     an object describing the index key(s), which may not be null. This can be of any type for which a `Codec` is 
   *                registered 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def createIndex(key: Bson): SingleObservable[String] = 
    observe(wrapped.createIndex(key, _: SingleResultCallback[String])) 
 
  /** 
   * @param key     an object describing the index key(s), which may not be null. This can be of any type for which a `Codec` is 
   *                registered 
   * @param options the options for the index 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def createIndex(key: Bson, options: IndexOptions): SingleObservable[String] = 
    observe(wrapped.createIndex(key, options, _: SingleResultCallback[String])) 
   
  dropIndex 
    /** 
   * Drops the given index. 
   * @param indexName the name of the index to remove 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def dropIndex(indexName: String): SingleObservable[Completed] = observeCompleted(wrapped.dropIndex(indexName, _: SingleResultCallback[Void])) 
 
  /** 
   * Drops the given index. 
   * @param indexName the name of the index to remove 
   * @param dropIndexOptions options to use when dropping indexes 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def dropIndex(indexName: String, dropIndexOptions: DropIndexOptions): SingleObservable[Completed] = 
    observeCompleted(wrapped.dropIndex(indexName, dropIndexOptions, _: SingleResultCallback[Void])) 
 
  /** 
   * Drops the index given the keys used to create it. 
   * @param keys the keys of the index to remove 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def dropIndex(keys: Bson): SingleObservable[Completed] = observeCompleted(wrapped.dropIndex(keys, _: SingleResultCallback[Void])) 
 
  /** 
   * Drops the index given the keys used to create it. 
   * @param keys the keys of the index to remove 
   * @param dropIndexOptions options to use when dropping indexes 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def dropIndex(keys: Bson, dropIndexOptions: DropIndexOptions): SingleObservable[Completed] = 
    observeCompleted(wrapped.dropIndex(keys, dropIndexOptions, _: SingleResultCallback[Void])) 
 
  /** 
   * Drop all the indexes on this collection, except for the default on _id. 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def dropIndexes(): SingleObservable[Completed] = 
    observeCompleted(wrapped.dropIndexes(_: SingleResultCallback[Void])) 
 
  /** 
   * Drop all the indexes on this collection, except for the default on _id. 
   * @param dropIndexOptions options to use when dropping indexes 
   * @return a Observable with a single element indicating when the operation has completed 
   */ 
  def dropIndexes(dropIndexOptions: DropIndexOptions): SingleObservable[Completed] = 
    observeCompleted(wrapped.dropIndexes(dropIndexOptions, _: SingleResultCallback[Void])) 
 
  listCollection 
    /** 
   * Finds all the collections in this database. 
   * @tparam TResult the target document type of the iterable. 
   * @return the fluent list collections interface 
   */ 
  def listCollections[TResult]()(implicit e: TResult DefaultsTo Document, ct: ClassTag[TResult]): ListCollectionsObservable[TResult] = ListCollectionsObservable(wrapped.listCollections(ct)) 
 
 createCollection 
    /** 
   * Create a new collection with the given name. 
   * @param collectionName the name for the new collection to create 
   * @return a Observable identifying when the collection has been created 
   */ 
  def createCollection(collectionName: String): SingleObservable[Completed] = 
    observeCompleted(wrapped.createCollection(collectionName, _: SingleResultCallback[Void])) 
 
  /** 
   * Create a new collection with the selected options 
   * @param collectionName the name for the new collection to create 
   * @param options        various options for creating the collection 
   * @return a Observable identifying when the collection has been created 
   */ 
  def createCollection(collectionName: String, options: CreateCollectionOptions): SingleObservable[Completed] = 
    observeCompleted(wrapped.createCollection(collectionName, options, _: SingleResultCallback[Void])) 
 
  createView 
    /** 
   * Creates a view with the given name, backing collection/view name, and aggregation pipeline that defines the view. 
   * @param viewName the name of the view to create 
   * @param viewOn   the backing collection/view for the view 
   * @param pipeline the pipeline that defines the view 
   */ 
  def createView(viewName: String, viewOn: String, pipeline: Seq[Bson]): SingleObservable[Completed] = 
    observeCompleted(wrapped.createView(viewName, viewOn, pipeline.asJava, _: SingleResultCallback[Void])) 
 
  /** 
   * Creates a view with the given name, backing collection/view name, aggregation pipeline, and options that defines the view. 
   * @param viewName          the name of the view to create 
   * @param viewOn            the backing collection/view for the view 
   * @param pipeline          the pipeline that defines the view 
   * @param createViewOptions various options for creating the view 
   */ 
  def createView(viewName: String, viewOn: String, pipeline: Seq[Bson], createViewOptions: CreateViewOptions): SingleObservable[Completed] = 
    observeCompleted(wrapped.createView(viewName, viewOn, pipeline.asJava, createViewOptions, _: SingleResultCallback[Void]))

import org.bson.conversions.Bson 
import org.mongodb.scala._ 
 
trait MGOCommands 
 
object MGOCommands  { 
  case class Count(filter: Option[Bson], options: Option[Any]) extends MGOCommands 
  case class Distict(fieldName: String, filter: Option[Bson]) extends MGOCommands 
  case class Find(filter: Option[Bson]) extends MGOCommands 
  case class Aggregate(pipeLine: Seq[Bson]) extends MGOCommands 
  case class MapReduce(mapFunction: String, reduceFunction: String) extends MGOCommands 
  case class Insert(newdocs: Seq[Document], options: Option[Any]) extends MGOCommands 
  case class Delete(filter: Bson, options: Option[Any], onlyOne: Boolean = false) extends MGOCommands 
  case class Replace(filter: Bson, replacement: Document, options: Option[Any]) extends MGOCommands 
  case class Update(filter: Bson, update: Bson, options: Option[Any]) extends MGOCommands 
  case class BulkWrite(commands: Seq[MGOCommands], options: Option[Any]) extends MGOCommands 
} 
 
object MGOAdmins { 
  case class DropCollection(collName: String) extends MGOCommands 
  case class CreateCollection(collName: String, options: Option[Any]) extends MGOCommands 
  case class ListCollection(dbName: String) extends MGOCommands 
  case class CreateView(viewName: String, viewOn: String, pipeline: Seq[Bson], options: Option[Any]) extends MGOCommands 
  case class CreateIndex(filter: Option[Bson], options: Option[Any]) extends MGOCommands 
  case class DropIndexByName(indexName: String, options: Option[Any]) extends MGOCommands 
  case class DropIndexByKey(key: Bson, options: Option[Any]) extends MGOCommands 
  case class DropAllIndexes(options: Option[Any]) extends MGOCommands 
} 
 
case class MGOContext ( 
     dbName: String, 
     collName: String, 
     action: MGOCommands = null 
) { ctx => 
  def setDbName(name: String): MGOContext = ctx.copy(dbName = name) 
  def setCollName(name: String): MGOContext = ctx.copy(collName = name) 
  def setCommand(cmd: MGOCommands): MGOContext = ctx.copy(action = cmd) 
} 
object MGOContext { 
  def apply(db: String, coll: String) = new MGOContext(db,coll) 
  def apply(db: String, coll: String, command: MGOCommands) = 
    new MGOContext(db,coll,command) 
 
}