java类com.mongodb.client.MongoCursor的实例源码

@Override
public List<String> getBookingsByUser(String user) {
    List<String> bookings = new ArrayList<String>();
    if(logger.isLoggable(Level.FINE)){
        logger.fine("getBookingsByUser : " + user);
    }
    try (MongoCursor<Document> cursor = booking.find(eq("customerId", user)).iterator()){

        while (cursor.hasNext()){
            Document tempBookings = cursor.next();
            Date dateOfBooking = (Date)tempBookings.get("dateOfBooking");
            tempBookings.remove("dateOfBooking");
            tempBookings.append("dateOfBooking", dateOfBooking.toString());

            if(logger.isLoggable(Level.FINE)){
                logger.fine("getBookingsByUser cursor data : " + tempBookings.toJson());
            }
            bookings.add(tempBookings.toJson());
        }
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
    return bookings;
}

@Override
public List<KeyValueConfigEntity> findAll(@Nonnull KeyValueConfigName configName) throws Exception {
  Objects.requireNonNull(configName);

  String collectionName = configName.getQualifiedName();
  MongoCollection<RawBsonDocument> collection =
      connector.getDatabase().getCollection(collectionName, RawBsonDocument.class);

  MongoCursor<RawBsonDocument> it = collection.find().iterator();
  if (!it.hasNext()) {
    return Collections.emptyList();
  }

  RawBsonDocument document = it.next();
  ByteArrayInputStream bin = new ByteArrayInputStream(document.getByteBuffer().array());
  ObjectMapper objectMapper = MongoConfigObjectMapper.getInstance();
  ObjectReader objectReader = objectMapper.readerFor(MongoConfigEntity.class);
  List<KeyValueConfigEntity> result = ((MongoConfigEntity) objectReader.readValue(bin)).getConfig();

  // set groupName on returned config key-value pairs
  return result.stream().map(input -> input.setConfigName(configName)).collect(Collectors.toList());
}

/**
 * @decription 查询数据库表名
 * @author yi.zhang
 * @time 2017年6月30日 下午2:16:02
 * @param table 表名
 * @return
 */
public List<String> queryTables(){
    try {
        if(session==null){
            init(servers, database, schema, username, password);
        }
        MongoIterable<String> collection = session.listCollectionNames();
        if (collection == null) {
            return null;
        }
        List<String> tables = new ArrayList<String>();
        MongoCursor<String> cursor = collection.iterator();
        while(cursor.hasNext()){
            String table = cursor.next();
            tables.add(table);
        }
        return tables;
    } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return null;
}

/**
 * 查询并逐条处理
 *
 * @param collectionName 集合名
 * @param query          查询条件
 * @param fields         返回字段或者排除字段
 * @param sort           排序方式
 * @param consumer       记录处理
 * @return
 */
public void findAndConsumer(
        String collectionName,
        MongodbQuery query, MongodbFields fields,
        MongodbSort sort, Consumer<Map<String, Object>> consumer) {
    MongoCollection<Document> collection = sMongoDatabase.getCollection(collectionName);
    FindIterable<Document> findIterable = collection.find(query == null ? new Document() : query.getQuery());
    if (fields == null) {
        findIterable.projection(new MongodbFields().getDbObject());
    } else {
        findIterable.projection(fields.getDbObject());
    }
    if (sort != null) {
        findIterable.sort(sort.getDbObject());
    }
    MongoCursor<Document> cursor = findIterable.iterator();
    try {
        while (cursor.hasNext()) {
            Map<String, Object> document = cursor.next();
            consumer.accept(document);
        }
    } finally {
        cursor.close();
    }
}

/**
 * 查询
 *
 * @param clazz          类
 * @param collectionName 集合名
 * @param sort           排序
 * @param <T>
 * @return
 */
public <T> List<T> findAll(Class<T> clazz, String collectionName, MongodbSort sort) {
    List<T> resultMapList = new ArrayList<T>();
    MongoCollection<Document> collection = sMongoDatabase.getCollection(collectionName);
    FindIterable<Document> findIterable = collection.find();
    if(sort != null) {
        findIterable.sort(sort.getDbObject());
    }
    MongoCursor<Document> cursor = findIterable.iterator();
    try {
        while (cursor.hasNext()) {
            Document document = cursor.next();
            T parseObject = JSON.parseObject(JSON.toJSONString(document), clazz);
            resultMapList.add(parseObject);
        }
    } finally {
        cursor.close();
    }

    return resultMapList;
}

/**
 * 查询
 *
 * @param collectionName 集合名
 * @param sort           排序方式
 * @return
 */
public List<Map<String, Object>> findAll(String collectionName, MongodbSort sort) {
    List<Map<String, Object>> resultMapList = new ArrayList<>();
    MongoCollection<Document> collection = sMongoDatabase.getCollection(collectionName);
    FindIterable<Document> findIterable = collection.find();
    if (sort != null) {
        findIterable.sort(sort.getDbObject());
    }
    MongoCursor<Document> cursor = findIterable.iterator();
    try {
        while (cursor.hasNext()) {
            Document document = cursor.next();
            resultMapList.add(document);
        }
    } finally {
        cursor.close();
    }

    return resultMapList;
}

/**
 * 查询一个
 *
 * @param collectionName 集合名
 * @param query          查询条件
 * @param fields         返回字段或者排除字段
 * @param sort
 * @return
 */
public Map<String, Object> findOne(
        String collectionName,
        MongodbQuery query, MongodbFields fields, MongodbSort sort) {
    MongoCollection<Document> collection = sMongoDatabase.getCollection(collectionName);
    FindIterable<Document> findIterable = collection.find(query.getQuery());
    if (fields == null) {
        findIterable.projection(new MongodbFields().getDbObject());
    } else {
        findIterable.projection(fields.getDbObject());
    }
    if (sort != null) {
        findIterable.sort(sort.getDbObject());
    }
    findIterable.limit(1);
    MongoCursor<Document> cursor = findIterable.iterator();
    try {
        if (cursor.hasNext()) {
            return cursor.next();
        }
    } finally {
        cursor.close();
    }
    return null;
}

/**
 * 查询一个
 *
 * @param clazz          类
 * @param collectionName 集合名
 * @param query          查询条件
 * @param fields         返回字段或者排除字段
 * @param sort
 * @param <T>
 * @return
 */
public <T> T findOne(
        Class<T> clazz, String collectionName,
        MongodbQuery query, MongodbFields fields, MongodbSort sort) {
    MongoCollection<Document> collection = sMongoDatabase.getCollection(collectionName);
    FindIterable<Document> findIterable = collection.find(query.getQuery());
    if (fields == null) {
        findIterable.projection(new MongodbFields().getDbObject());
    } else {
        findIterable.projection(fields.getDbObject());
    }
    if (sort != null) {
        findIterable.sort(sort.getDbObject());
    }
    findIterable.limit(1);
    MongoCursor<Document> cursor = findIterable.iterator();
    try {
        if (cursor.hasNext()) {
            Document document = cursor.next();
            return JSON.parseObject(document.toJson(), clazz);
        }
    } finally {
        cursor.close();
    }
    return null;
}

/**
 * 根据统计字段计算统计结果（gte最小值）并排序
 *
 * @param collectionName 集合名
 * @param match          match条件
 * @param field          统计字段
 * @param minCount       最小值
 * @return
 */
public LinkedHashMap<String, Integer> sortMap(String collectionName, Document match, String field, int minCount) {
    AggregateIterable<Document> aggregate = getDB().getCollection(collectionName).aggregate(
            Arrays.asList(
                    match(match)
                    , group("$" + field, Accumulators.sum("_count", 1))
                    , match(new Document("_count", new Document("$gte", minCount)))
                    , sort(new Document("_count", -1))
            )
    );

    LinkedHashMap<String, Integer> map = new LinkedHashMap<String, Integer>();
    MongoCursor<Document> iterator = aggregate.iterator();
    while (iterator.hasNext()) {
        Document next = iterator.next();
        map.put(next.getString("_id"), next.getInteger("_count"));
    }
    return map;
}

@Override
long executeQuery(int threadId, long threadRunCount, long globalRunCount, long selectorId, long randomId){
    final MongoCursor<Document> cursor = mongoCollection.find(eq(queriedField, selectorId)).iterator();
    //final MongoCursor<Document> cursor = mongoCollection.find(in(queriedField, selectorId, selectorId+1, selectorId+2, selectorId+3, selectorId+4)).iterator();
    long result = 0;
    try {
        while (cursor.hasNext()) {
            final Document doc = cursor.next();
            LOG.debug("Document {}", doc.toJson());
            result++;
        }
    } finally {
        cursor.close();
    }

    return result;
}

public static List<OFFProduct> mkOFFProductsFromMongoCursor(MongoCursor<Document> cursor) throws IOException, JSONException{
    List<OFFProduct> list = new ArrayList<>();
    Document product;
    int count = 0;
    String out = "+";
    while(cursor.hasNext()){
        product = cursor.next();
        list.add(mkOFFProductFromBSON(product));
        count++;
        if(count%1000 == 0){
            out += "+";
            System.out.println(count + " products done");
            System.out.println(out);
        }
    }
    System.out.println(count + " products done");
    return list;
}

private Object find(String filter) throws ServiceException {
    try {
        List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();

        FindIterable<Document> cursor = null;
        if (filter != null) {
            BasicDBObject filterObj = BasicDBObject.parse(filter);
            cursor = this.collection.find(filterObj);
        } else {
            cursor = this.collection.find();
        }

        if (cursor != null) {
            MongoCursor<Document> iterator = cursor.iterator();
            while (iterator.hasNext()) {
                result.add(iterator.next());
            }
        }
        return result;
    } catch (Exception e) {
        throw new ServiceException("Search Exception: " + e.getMessage());
    }
}

private void _backup(PrintStream out, String tablename) {
    log.debug("backuping " + tablename);
    MongoCollection<Document> d1 = getCollection(Helper.DEFAULT, tablename);
    MongoCursor<Document> c1 = d1.find().iterator();
    int rows = 0;
    while (c1.hasNext()) {
        rows++;

        Document d2 = c1.next();
        JSON jo = new JSON();
        jo.put("_table", tablename);
        for (String name : d2.keySet()) {
            jo.put(name, d2.get(name));
        }
        out.println(jo.toString());
        if (rows % 1000 == 0)
            log.debug("backup " + tablename + ", rows=" + rows);
    }
}

@Override
public Document findByObjectId(String docName, String objectHexString) {
    if (objectHexString == null) {
        return null;
    }
    ObjectId objectId = new ObjectId(objectHexString);
    Document doc = new Document();
    doc.put(ObjectIdKey, objectId);

    MongoCursor<Document> cursor = baseFind(docName, doc);
    doc = null;
    if (cursor.hasNext()) {
        doc = cursor.next();
    }
    return doc;
}

@Test
public void queryDataTest() {
    BasicDBObject fields = new BasicDBObject();
    fields.put("likes", new BasicDBObject("$gt", 16));
    FindIterable<Document> iterable = mongoPool.getDB()
            .getCollection("test").find(fields);
    MongoCursor<Document> cursor = iterable.iterator();
    System.out.println(cursor);
    while (cursor.hasNext()) {
        Document document = cursor.next();

        System.out.println(document);

        if (document.get("append") != null) {
            System.out.println(document.get("append") instanceof Document);
        }
    }
}

@Override
public List<HeldPermission<UUID>> getUsersWithPermission(String permission) {
    ImmutableList.Builder<HeldPermission<UUID>> held = ImmutableList.builder();
    MongoCollection<Document> c = this.database.getCollection(this.prefix + "users");
    try (MongoCursor<Document> cursor = c.find().iterator()) {
        while (cursor.hasNext()) {
            Document d = cursor.next();
            UUID holder = d.get("_id", UUID.class);

            Set<NodeModel> nodes = new HashSet<>(nodesFromDoc(d));
            for (NodeModel e : nodes) {
                if (!e.getPermission().equalsIgnoreCase(permission)) {
                    continue;
                }
                held.add(NodeHeldPermission.of(holder, e));
            }
        }
    }
    return held.build();
}

@Override
public Group createAndLoadGroup(String name) {
    Group group = this.plugin.getGroupManager().getOrMake(name);
    group.getIoLock().lock();
    try {
        MongoCollection<Document> c = this.database.getCollection(this.prefix + "groups");
        try (MongoCursor<Document> cursor = c.find(new Document("_id", group.getName())).iterator()) {
            if (cursor.hasNext()) {
                Document d = cursor.next();
                Set<Node> nodes = nodesFromDoc(d).stream().map(NodeModel::toNode).collect(Collectors.toSet());
                group.setEnduringNodes(nodes);
            } else {
                c.insertOne(groupToDoc(group));
            }
        }
    } finally {
        group.getIoLock().unlock();
    }
    group.getRefreshBuffer().requestDirectly();
    return group;
}

@Override
public List<HeldPermission<String>> getGroupsWithPermission(String permission) {
    ImmutableList.Builder<HeldPermission<String>> held = ImmutableList.builder();
    MongoCollection<Document> c = this.database.getCollection(this.prefix + "groups");
    try (MongoCursor<Document> cursor = c.find().iterator()) {
        while (cursor.hasNext()) {
            Document d = cursor.next();

            String holder = d.getString("_id");
            Set<NodeModel> nodes = new HashSet<>(nodesFromDoc(d));
            for (NodeModel e : nodes) {
                if (!e.getPermission().equalsIgnoreCase(permission)) {
                    continue;
                }
                held.add(NodeHeldPermission.of(holder, e));
            }
        }
    }
    return held.build();
}

@Override
public Track createAndLoadTrack(String name) {
    Track track = this.plugin.getTrackManager().getOrMake(name);
    track.getIoLock().lock();
    try {
        MongoCollection<Document> c = this.database.getCollection(this.prefix + "tracks");
        try (MongoCursor<Document> cursor = c.find(new Document("_id", track.getName())).iterator()) {
            if (!cursor.hasNext()) {
                c.insertOne(trackToDoc(track));
            } else {
                Document d = cursor.next();
                //noinspection unchecked
                track.setGroups((List<String>) d.get("groups"));
            }
        }
    } finally {
        track.getIoLock().unlock();
    }
    return track;
}

@SuppressWarnings("unchecked")
@Test
public void testVerifyDataAfterFeatureExecution() throws DbFeatureException {
    // GIVEN
    final MongoIterable<String> collectionIterable = mock(MongoIterable.class);
    final MongoCursor<String> iterator = mock(MongoCursor.class);
    when(expectedDataSets.strict()).thenReturn(Boolean.FALSE);
    when(expectedDataSets.value()).thenReturn(new String[] {});
    when(expectedDataSets.orderBy()).thenReturn(new String[] {});
    when(expectedDataSets.excludeColumns()).thenReturn(new String[] {});
    when(connection.listCollectionNames()).thenReturn(collectionIterable);
    when(collectionIterable.iterator()).thenReturn(iterator);
    when(iterator.hasNext()).thenReturn(Boolean.FALSE);

    // WHEN
    final DbFeature<MongoDatabase> feature = featureExecutor.createVerifyDataAfterFeature(expectedDataSets);
    assertThat(feature, notNullValue());
    feature.execute(connection);

    // THEN
    verify(connection).listCollectionNames();
    verifyNoMoreInteractions(connection);
}

/**
 * @return
 */
public List<Park> getAll() {
    System.out.println("[DEBUG] MongoDBConnection.getAll()");
    ArrayList<Park> allParksList = new ArrayList<Park>();

    if (mongoDB != null) {
        try {
            MongoCollection parks = mongoDB.getCollection(COLLECTION);
            MongoCursor<Document> cursor = parks.find().iterator();
            try {
                while (cursor.hasNext()) {
                    allParksList.add(ParkReadConverter.convert(cursor.next()));
                }
            } finally {
                cursor.close();
            }
        } catch (Exception e) {
            System.out.println("[ERROR] Error connecting to MongoDB. " + e.getMessage());
        }
    } else {
        System.out.println("[ERROR] mongoDB could not be initiallized. No operation with DB will be performed");
    }
    return allParksList;
}

/**
 * @param query
 * @return
 */
public List<Park> getByQuery(BasicDBObject query) {
    System.out.println("[DEBUG] MongoDBConnection.getByQuery()");
    List<Park> parks = new ArrayList<Park>();
    if (mongoDB != null) {
        try {
            MongoCursor<Document> cursor = mongoDB.getCollection(COLLECTION).find(query).iterator();
            int i = 0;
            try {
                while (cursor.hasNext()) {
                    parks.add(ParkReadConverter.convert(cursor.next()));
                }
            } finally {
                cursor.close();
            }
        } catch (Exception e) {
            System.out.println("[ERROR] Error connecting to MongoDB. " + e.getMessage());
        }

    } else {
        System.out.println("[ERROR] mongoDB could not be initiallized. No operation with DB will be performed");
    }
    return parks;
}

@Override
public <P extends ParaObject> Map<String, P> readAll(String appid, List<String> keys, boolean getAllColumns) {
    if (keys == null || keys.isEmpty() || StringUtils.isBlank(appid)) {
        return new LinkedHashMap<String, P>();
    }
    Map<String, P> results = new LinkedHashMap<String, P>(keys.size(), 0.75f, true);
    BasicDBObject inQuery = new BasicDBObject();
    inQuery.put(ID, new BasicDBObject("$in", keys));

    MongoCursor<Document> cursor = getTable(appid).find(inQuery).iterator();
    while (cursor.hasNext()) {
        Document d = cursor.next();
        P obj = fromRow(d);
        if (d != null) {
            results.put(d.getString(ID), obj);
        }
    }

    logger.debug("DAO.readAll() {}", results.size());
    return results;
}

public List<String> searchIncome() {
    List<String> ret = new ArrayList<>();
    MongoCollection<Document> collection = mongo.getCollection("income");
    custRepo.findByName("test");

    Bson filter = and(eq("i", 74), gt("a", 6));
    Bson sort = null;

    MongoCursor<Document> docs = collection//
            .find()//
            .limit(1000)//
            .filter(filter)//
            .sort(sort)//
            .iterator();

    Document doc;
    while ((doc = docs.tryNext()) != null) {
        ret.add(doc.toJson());
    }
    return ret;
}

public boolean collectionExists(String collectionName) {
    if (this.database == null) {
        return false;
    }

    final MongoIterable<String> iterable = database.listCollectionNames();
    try (final MongoCursor<String> it = iterable.iterator()) {
        while (it.hasNext()) {
            if (it.next().equalsIgnoreCase(collectionName)) {
                return true;
            }
        }
    }

    return false;
}

public List<Document> getDocuments(BasicDBObject query, RESOURCE_TYPE resType, String sortKey, boolean asc, int limit) {

    ArrayList<Document> docList = new ArrayList<Document>();

    BasicDBObject sort = new BasicDBObject(sortKey, asc ? 1 : -1);

    MongoCollection<Document> collection = context.getDatabaseManager()
            .getCollection(collectionName);
    MongoCursor<Document> cursor = collection.find(query).sort(sort)
            .limit(limit).iterator();
    while (cursor.hasNext()) {
        docList.add(cursor.next());
    }

    return docList;

}

public List<Document> getDocuments(String keyName, String keyValue,
        RESOURCE_TYPE resType, String sortKey, boolean asc, int limit) {

    ArrayList<Document> docList = new ArrayList<Document>();

    BasicDBObject query = new BasicDBObject(keyName, keyValue).append(
            RESTYPE_KEY, resType.Value());
    BasicDBObject sort = new BasicDBObject(sortKey, asc ? 1 : -1);

    MongoCollection<Document> collection = context.getDatabaseManager()
            .getCollection(collectionName);
    MongoCursor<Document> cursor = collection.find(query).sort(sort)
            .limit(limit).iterator();
    while (cursor.hasNext()) {
        docList.add(cursor.next());
    }

    return docList;

}

/**
 * 分页查询
 *
 * @param collection
 * @param pageNumber
 * @param pageSize
 * @param filter
 * @param like
 * @param sort
 * @return
 */
public static Page<Record> paginate(String collection, int pageNumber, int pageSize, BasicDBObject filter, BasicDBObject like, BasicDBObject sort) {
    BasicDBObject conditons = new BasicDBObject();
    buildFilter(filter, conditons);
    buildLike(like, conditons);
    MongoCursor<Document> cursor = getCollection(collection).find(conditons)
            .skip((pageNumber - 1) * pageSize).limit(pageSize).sort(sort(sort)).iterator();

    List<Record> records = new ArrayList<>();
    Long totalRow = getCollection(collection).count(conditons);
    while (cursor.hasNext()) {
        records.add(toRecord(cursor.next()));
    }
    if (totalRow <= 0) {
        return new Page<>(new ArrayList<Record>(0), pageNumber, pageSize, 0, 0);
    }
    Long totalPage = totalRow / pageSize;
    if (totalRow % pageSize != 0) {
        totalPage++;
    }
    Page<Record> page = new Page<>(records, pageNumber, pageSize, totalPage.intValue(), totalRow.intValue());
    return page;
}

/**
 * Method that obtain the id of any collection based on a key and its value
 * @param collection
 * @param key
 * @param value
 * @return
 */
public String getIdByKey(String collection, String key, String value) {
    String res = "";
    MongoCollection<Document> docs =
            DBManager
                    .getInstance()
                    .getCollection(collection)
            ;
    MongoCursor<Document> cursor =
            docs
                    .find(eq(key, value))
                    .iterator();
    while (cursor.hasNext()) {
        Document doc = cursor.next();
        res += doc.get(Constant.ID);
    }
    return res;
}

/**
 * Get the MongoDB cursor.
 */
private MongoCursor<Document> getCursor(int skip) {
    if (cursor == null && cursorId == 0) {
        Document query = Document.parse(config.getMongo().getQuery());
        List<Bson> pipes = new ArrayList<>(3);
        pipes.add(match(query));
        pipes.add(skip(skip));

        Optional.ofNullable(config.getMongo().getProject()).ifPresent(p -> pipes.add(project(Document.parse(p))));

        AggregateIterable<Document> aggregate = collection.aggregate(pipes)
                .allowDiskUse(true)
                .useCursor(true);

        cursor = aggregate.iterator();

        // TODO: Persist cursor ID somewhere to allow restarts.
        Optional.ofNullable(cursor.getServerCursor()).ifPresent(serverCursor -> cursorId = serverCursor.getId());
    } else if (cursor == null && cursorId != 0) {
        // TODO: Lookup cursor ID for resume.
        // Open existing cursor in case of restart??
    }

    return cursor;
}