Java 类com.mongodb.client.model.Filters 实例源码

@Test
public void additionalPropertiesInPojoTest() {
    MongoCollection<Document> collection = mongoClient.getDatabase("test").getCollection("documents");

    MorePropertiesPojo pojo = new MorePropertiesPojo();
    pojo.someNonPersistableString = "do not persist me!";
    pojo.aString = "persistable String";
    pojo.anInt = 22;

    // first insert
    collection.withDocumentClass(MorePropertiesPojo.class).insertOne(pojo);

    // second insert
    pojo.id = null;
    collection.withDocumentClass(CompletePojo.class).insertOne(pojo);

    CompletePojo readPojo = collection.withDocumentClass(CompletePojo.class).find(Filters.eq("_id", pojo.id)).first();

    Assert.assertEquals(readPojo.aString, pojo.aString);

}

public O2MSyncDataLoader getPendingDataLoader() {
    O2MSyncDataLoader loader = null;
    Document document = syncEventDoc.findOneAndUpdate(
            Filters.and(Filters.eq(SyncAttrs.STATUS, SyncStatus.PENDING),
                    Filters.eq(SyncAttrs.EVENT_TYPE, String.valueOf(EventType.System))),
            Updates.set(SyncAttrs.STATUS, SyncStatus.IN_PROGRESS),
            new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER)
                    .projection(Projections.include(SyncAttrs.SOURCE_DB_NAME, SyncAttrs.SOURCE_USER_NAME)));
    if (document != null && !document.isEmpty()) {
        Object interval = document.get(SyncAttrs.INTERVAL);
        String appName = document.getString(SyncAttrs.APPLICATION_NAME);
        if(interval!=null && interval instanceof Long){
            loader = new O2MSyncDataLoader((Long)interval, appName);
        }else{
            loader = new O2MSyncDataLoader(120000, appName);
        }
        loader.setEventId(document.getObjectId(SyncAttrs.ID));
        loader.setDbName(document.getString(SyncAttrs.SOURCE_DB_NAME));
        loader.setDbUserName(document.getString(SyncAttrs.SOURCE_USER_NAME));
        loader.setStatus(document.getString(SyncAttrs.STATUS));
    }
    return loader;
}

@Override
public Dataset write(Dataset dataset) {
  // we populate this on first write and retain it thereafter
  if (isBlank(dataset.getId())) {
    dataset.setId(ObjectId.get().toString());
  }

  Observable<Document> observable =
      getCollection()
          .findOneAndReplace(
              Filters.eq("id", dataset.getId()),
              documentTransformer.transform(dataset),
              new FindOneAndReplaceOptions().upsert(true).returnDocument(ReturnDocument.AFTER));

  return documentTransformer.transform(Dataset.class, observable.toBlocking().single());
}

/**
 * 删除记录
 * 
 * @param collectionName
 *            表名
 * @param mongoObj
 *            记录
 * @return
 */
public static boolean deleteById(String collectionName, MongoObj mongoObj) {
    MongoCollection<Document> collection = getCollection(collectionName);
    try {
        Bson filter = Filters.eq(MongoConfig.MONGO_ID, mongoObj.getDocument().getObjectId(MongoConfig.MONGO_ID));
        DeleteResult result = collection.deleteOne(filter);
        if (result.getDeletedCount() == 1) {
            return true;
        } else {
            return false;
        }
    } catch (Exception e) {
        if (log != null) {
            log.error("删除记录失败", e);
        }
        return false;
    }

}

/**
 * Get the current ClientState of a client,
 * connected or not.
 * @param uuid The UUID of the client. This must be a valid
 *             UUID which belongs to a client. If the UUID is
 *             not found in the database or connected clients,
 *             then a RuntimeException is thrown.
 * @return The ClientState of the specified client with the UUID.
 */
public ClientState queryClientState(String uuid) {
    for(ClientSession session : this.sessions.values()) {
        if(session.getToken().getUuid().equals(uuid)) {
            return session.getState();
        }
    }

    // The session is not currently connected, so we need to check the database
    MongoCollection<Document> clients = NectarServerApplication.getDb().getCollection("clients");
    Document doc = clients.find(Filters.eq("uuid", uuid)).first();
    if(doc != null) {
        return ClientState.fromInt(doc.getInteger("state", ClientState.UNKNOWN.toInt()));
    }

    // We couldn't find the client in the database, so throw an exception
    throw new RuntimeException("Failed to find UUID " + uuid + "in connected sessions or in database. Is it invalid?");
}

@SuppressWarnings("unchecked")
private IndexJSON[] constructIndexJSON(MongoCollection<Document> index, boolean isPublic, String loggedInUser) {
    List<IndexJSON> list = new ArrayList();

    index.find(Filters.eq("isPublic", isPublic)).forEach((Consumer<? super Document>) (Document doc) -> {
        if(loggedInUser != null && doc.getString("storePath").startsWith(loggedInUser) && !isPublic) {
            // It's the user store
            list.add(
                    new IndexJSON(doc.getString("storePath"), doc.getString("checksum"), doc.getString("lastUpdatedBy")));
        } else if(isPublic)
            list.add(
                new IndexJSON(doc.getString("storePath"), doc.getString("checksum"), doc.getString("lastUpdatedBy")));
    });

    return list.toArray(new IndexJSON[list.size()]);
}

@Override
protected void collectVectors(Collection<String> terms, int limit) {
    Set<String> toFetch = terms.stream()
            .filter(t -> !this.vectorsCache.containsKey(t))
            .collect(Collectors.toSet());

    logger.debug("Cache has {} vectors, need to fetch more {}",
            terms.size() - toFetch.size(), toFetch.size());

    if (!toFetch.isEmpty()) {
        logger.info("Collecting {} term vectors from {}", toFetch.size(), dbName);
        FindIterable<Document> docs = getTermsColl().find(Filters.in(TERM_FIELD_NAME, toFetch));
        if (docs != null) {
            docs.batchSize(toFetch.size());
            for (Document doc : docs) {
                this.vectorsCache.put(doc.getString(TERM_FIELD_NAME), unmarshall(doc, limit));
            }
        }
    }
}

private Map<String, List<String>> doTranslate(Set<String> tokens) {
    MongoCollection<Document> lexColl = getLexCollection();
    FindIterable<Document> lexs = lexColl.find(Filters.in(TERM_FIELD, tokens));

    Map<String, List<String>> res = new HashMap<>();
    for (Document doc : lexs) {
        Document tr = (Document) doc.get(TRANSLATION_FIELD);

        if (tr != null) {
            tr.remove(NULL_VALUE);
            res.put(doc.getString(TERM_FIELD), getRelevantTranslations((Map) tr));
        }
    }

    return res;
}

/**
 * @decription 更新数据
 * @author yi.zhang
 * @time 2017年6月2日 下午6:19:08
 * @param table 文档名称(表名)
 * @param obj
 * @return
 */
public int update(String table, Object obj) {
    try {
        if(session==null){
            init(servers, database, schema, username, password);
        }
        MongoCollection<Document> collection = session.getCollection(table);
        if (collection == null) {
            return 0;
        }
        JSONObject json = JSON.parseObject(JSON.toJSONString(obj));
        Document value = Document.parse(JSON.toJSONString(obj));
        collection.replaceOne(Filters.eq("_id", json.containsKey("_id")?json.get("_id"):json.get("id")), value);
        return 1;
    } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return -1;
}

/**
 * @decription 删除数据
 * @author yi.zhang
 * @time 2017年6月2日 下午6:19:25
 * @param table 文档名称(表名)
 * @param obj
 * @return
 */
public int delete(String table, Object obj) {
    try {
        if(session==null){
            init(servers, database, schema, username, password);
        }
        MongoCollection<Document> collection = session.getCollection(table);
        if (collection == null) {
            return 0;
        }
        JSONObject json = JSON.parseObject(JSON.toJSONString(obj));
        collection.findOneAndDelete(Filters.eq("_id", json.containsKey("_id")?json.get("_id"):json.get("id")));
        return 1;
    } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return -1;
}

public SyncEvent retryEvent(ObjectId eventId, boolean retryFailed, boolean retryEntire, boolean dropCollection) {
    Document updateQuery = new Document();
    updateQuery.append("$set", new Document(SyncAttrs.STATUS, SyncStatus.PENDING).append(SyncAttrs.IS_RETRY, true))
            .append("$unset", new Document(SyncAttrs.ERRORS, true).append(SyncAttrs.MARKER, true));
    if (retryFailed) {
        syncEvents.updateMany(
                Filters.and(Filters.eq(SyncAttrs.PARENT_EVENT_ID, eventId),
                        Filters.eq(SyncAttrs.STATUS, SyncStatus.FAILED), Filters.ne(SyncAttrs.ID, eventId)),
                updateQuery);
        syncEvents.updateOne(Filters.eq(SyncAttrs.ID, eventId),
                Updates.set(SyncAttrs.STATUS, SyncStatus.IN_PROGRESS));
    } else {
        if (retryEntire) {
            syncEvents.updateMany(Filters.eq(SyncAttrs.PARENT_EVENT_ID, eventId),
                    Updates.set(SyncAttrs.STATUS, SyncStatus.CANCELLED));
            syncEvents.updateOne(Filters.eq(SyncAttrs.ID, eventId), updateQuery);
        }
    }
    return getEvent(eventId);
}

public static void main(String[] args) {
    MongoClientOptions mongoClientOptions = new MongoClientOptions.Builder().codecRegistry(getCodecRegistry()).build();
    MongoClient mongoClient = new MongoClient(new ServerAddress("localhost", 27017), mongoClientOptions);

    MongoDatabase database = mongoClient.getDatabase("tutorial");
    MongoCollection<PolymorphicPojo> collection = database.getCollection("entities").withDocumentClass(PolymorphicPojo.class);

    // create some pojo
    Pojo pojo = new Pojo();
    pojo.setName("A nice name");
    pojo.setPojos(Arrays.asList(new SubPojo(42), new SubPojo(48)));

    // insert into db
    collection.insertOne(pojo);

    // read from db
    PolymorphicPojo foundPojo = collection.find(Filters.eq("_id", pojo.getId())).first();

    // output
    LOGGER.debug("Found pojo {}", foundPojo);
}

public static ChatObj getChatById(String chatId, String chatGroupId) {
    if (StringUtil.stringIsNull(chatId) || StringUtil.stringIsNull(chatGroupId)) {
        return null;
    }
    Bson filter = Filters.eq(ChatObj.CHAT_ID, chatId);
    List<ChatObj> list = MongodbManager.find(chatGroupId, filter, ChatObj.class, 0, 0);
    if (list != null && list.size() != 0) {
        return list.get(0);
    } else {
        return null;
    }
}

public static ChatObj getChatById(String chatId, String key1, String key2) {
    if (StringUtil.stringIsNull(chatId) || StringUtil.stringIsNull(key1) || StringUtil.stringIsNull(key2)) {
        return null;
    }
    Bson filter = Filters.eq(ChatObj.CHAT_ID, chatId);
    List<ChatObj> list = MongodbManager.find(getUserToUserCollectionName(key1, key2), filter, ChatObj.class, 0, 0);
    if (list != null && list.size() != 0) {
        return list.get(0);
    } else {
        return null;
    }
}

public static long getUserChatNum(String toTypeId, String fromUserId, String chatCreateTime) {
    if (StringUtil.stringIsNull(toTypeId) || StringUtil.stringIsNull(fromUserId)) {
        return 0;
    }
    Date date = null;
    if (!StringUtil.stringIsNull(chatCreateTime)) {
        date = TimeUtils.stringToDateDay(chatCreateTime);
    }
    Bson filter = null;
    if (!StringUtil.stringIsNull(chatCreateTime)) {
        filter = Filters.lt(ChatObj.CHAT_CREATE_TIME, String.valueOf(date.getTime()));
    }
    long count = MongodbManager.count(getUserToUserCollectionName(toTypeId, fromUserId), filter);
    return count;
}

@Test
public void testStrangeId() {
    final MongoCollection<EntityWithStrangeId> collection = mongoClient.getDatabase(DB_NAME).getCollection("documents").withDocumentClass(EntityWithStrangeId.class);
    EntityWithStrangeId entityWithStrangeId = new EntityWithStrangeId();
    collection.insertOne(entityWithStrangeId);
    assertThat(entityWithStrangeId.id, is(not(nullValue())));
    EntityWithStrangeId foundEntityWithStrangeId = collection.find(Filters.eq("_id", entityWithStrangeId.id)).first();
    assertThat(foundEntityWithStrangeId, is(not(nullValue())));
    assertThat(foundEntityWithStrangeId.id, equalTo(entityWithStrangeId.id));
}

@Override
public boolean exists(String id) {
  Observable<Long> count =
      getCollection().count(Filters.eq("id", id), new CountOptions().limit(1));

  return count.toBlocking().single() > 0;
}

@Override
public long delete(String id) {
  Observable<Document> deleted = getCollection().findOneAndDelete(Filters.eq("id", id));

  Document document = deleted.toBlocking().singleOrDefault(null);

  return document != null ? 1 : 0;
}

private void assertDatasetIsWritten(CannedDataset cannedDataset) {
  FindObservable<Document> datasets =
      getCollection(configuration.getDatabaseName(), configuration.getDatasetStorageName())
          .find(Filters.eq("name", cannedDataset.getDataset().getName()));

  assertThat(
      documentTransformer.transform(Dataset.class, datasets.first().toBlocking().single()),
      is(cannedDataset.getDataset()));
}

@RequestMapping(NectarServerApplication.ROOT_PATH + "/auth/removeClient")
public ResponseEntity removeClient(@RequestParam(value = "token") String jwtRaw, @RequestParam(value = "uuid") String uuid,
                                   HttpServletRequest request) {
    ResponseEntity r = Util.verifyJWT(jwtRaw, request);
    if(r != null)
        return r;

    ManagementSessionToken token = ManagementSessionToken.fromJSON(Util.getJWTPayload(jwtRaw));
    if(token == null)
        return ResponseEntity.status(HttpStatus.BAD_REQUEST).body("Invalid TOKENTYPE.");

    if(SessionController.getInstance().checkManagementToken(token)) {
        MongoCollection<Document> clients = NectarServerApplication.getDb().getCollection("clients");

        Document client = clients.find(Filters.eq("uuid", uuid)).first();

        if(client == null) // Check if the client exists
            return ResponseEntity.status(HttpStatus.NOT_FOUND).body("Client not found in database.");

        if(!(client.getOrDefault("loggedInUser", "null").equals("null"))) { // Check if a user is currently signed into the client.
            NectarServerApplication.getLogger().warn("Attempted client deletion from " + request.getRemoteAddr() + ", a user is already signed into client " + uuid);
            return ResponseEntity.status(HttpStatus.CONFLICT).body("A user is currently signed into this client.");
        }

        if(SessionController.getInstance().sessions.containsKey(uuid)) { // Check if the client is currently online with a session open
            SessionController.getInstance().sessions.remove(uuid); // Remove the session and it's token
            NectarServerApplication.getLogger().info("Revoked token for " + uuid + ": client deleted");
        }

        clients.deleteOne(Filters.eq("uuid", uuid)); // Delete client from the MongoDB database

        NectarServerApplication.getEventLog().logEntry(EventLog.EntryLevel.NOTICE, "Deleted client " + uuid + ", traced from " + request.getRemoteAddr());
    } else {
        return ResponseEntity.status(HttpStatus.FORBIDDEN).body("Token expired/not valid.");
    }

    return ResponseEntity.status(HttpStatus.NO_CONTENT).body("Success.");
}

protected static ResponseEntity checkUserAdmin(SessionToken token, MongoCollection<Document> users, Document doc) {
    // getString will throw an exception if the key is not present in the document
    String loggedInUser = doc.getString("loggedInUser");
    if(loggedInUser.equals("none")) {
        // No user is logged in
        throw new RuntimeException(); // Move to catch block
    }

    Document userDoc = users.find(Filters.eq("username", loggedInUser)).first();

    if(userDoc == null) { // We can't find the logged in user in the users database, strange
        NectarServerApplication.getLogger().warn("Failed to find logged in user \"" + loggedInUser + "\" for session "
                + token.getUuid() + " while processing user registration"
        );
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Failed to find current logged in user in DB.");
    } else { // User found, check admin now
        if(!userDoc.getBoolean("admin", false)) {
            NectarServerApplication.getLogger().warn("ATTEMPTED CLIENT REGISTRATION BY NON_ADMIN USER \"" + loggedInUser + "\""
                    + " from session " + token.getUuid()
            );

            NectarServerApplication.getEventLog().addEntry(EventLog.EntryLevel.WARNING, "A non-admin user attempted to register a client from " + token.getUuid());
            throw new RuntimeException(); // Move to catch block
        }
        // User is confirmed logged in and admin, all checks passed.
    }
    return null;
}

@RequestMapping(NectarServerApplication.ROOT_PATH + "/query/queryState")
public ResponseEntity<Integer> queryState(@RequestParam(value = "token") String jwtRaw,
                                          @RequestParam(value = "uuid") String uuid) {

    ManagementSessionToken token = ManagementSessionToken.fromJSON(Util.getJWTPayload(jwtRaw));
    if(token == null)
        return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(-1);

    if(!SessionController.getInstance().checkManagementToken(token)) {
        return ResponseEntity.status(HttpStatus.FORBIDDEN).body(-1);
    }

    if(SessionController.getInstance().sessions.containsKey(uuid)) {
        return ResponseEntity.ok(SessionController.getInstance().sessions.get(uuid).getState().toInt());
    }

    // The session requested is not connected. Check the database then.
    MongoCollection<Document> clients = NectarServerApplication.getDb().getCollection("clients");
    Document doc = clients.find(Filters.eq("uuid", uuid)).first();
    if(doc == null) {
        // We couldn't find a client with the UUID, that means it's an invalid client UUID
        return ResponseEntity.status(HttpStatus.NOT_FOUND).body(-1);
    } else {
        // We found the client
        // Default value is UNKNOWN, in case the state was never set in the database yet.
        return ResponseEntity.ok(doc.getInteger("state", ClientState.UNKNOWN.toInt()));
    }
}

@SuppressWarnings("unchecked")
@RequestMapping(NectarServerApplication.ROOT_PATH + "/query/queryUsers")
public ResponseEntity queryUsers(@RequestParam(value = "token") String jwtRaw, HttpServletRequest request) {
    ManagementSessionToken token = ManagementSessionToken.fromJSON(Util.getJWTPayload(jwtRaw));
    if(token == null)
        return ResponseEntity.status(HttpStatus.BAD_REQUEST).body("Invalid TOKENTYPE.");

    if(!SessionController.getInstance().checkManagementToken(token)) {
        return ResponseEntity.status(HttpStatus.FORBIDDEN).body("Token expired/not valid.");
    }

    JSONObject returnJSON = new JSONObject();

    MongoCollection<Document> clients = NectarServerApplication.getDb().getCollection("clients");
    MongoCollection<Document> users = NectarServerApplication.getDb().getCollection("users");
    users.find().forEach((Block<Document>) document -> {
        String username = document.getString("username");
        boolean admin = document.getBoolean("admin", false);

        JSONObject userJSON = new JSONObject();
        userJSON.put("admin", admin);

        Document clientDoc = clients.find(Filters.eq("loggedInUser", username)).first();
        if(clientDoc == null) {
            userJSON.put("signedIn", false);
        } else {
            userJSON.put("signedIn", true);
        }

        returnJSON.put(username, userJSON);
    });

    return ResponseEntity.ok(returnJSON.toJSONString());
}

private static void buildChecksumFile(File file, MongoCollection<Document> index, List<Document> toInsert, boolean isPublic) throws IOException {
    // Is a file, build the checksum then
    String checksum = Util.computeFileSHA256Checksum(file);
    Document fileDoc = index.find(Filters.eq("path", file.getAbsolutePath())).first();
    if(fileDoc == null) {
        toInsert.add(new Document()
                .append("path", file.getAbsolutePath())
                .append("storePath", Util.absoluteFTSToRelativeStore(file.getAbsolutePath()))
                .append("isPublic", isPublic)
                .append("checksum", checksum)
                .append("lastUpdatedBy", "server"));
    } else {
        String dbChecksum = fileDoc.getString("checksum");
        if(!checksum.equals(dbChecksum)) {
            // Checksum has changed, we assume the file has been changed by the server
            // This is because if a client changes it, the database will be updated
            index.updateOne(Filters.eq("path", file.getAbsolutePath()),
                    new Document("$set", new Document("checksum", checksum))
            ); // Update the checksum into the database

            index.updateOne(Filters.eq("path", file.getAbsolutePath()),
                    new Document("$set", new Document("lastUpdatedBy", "server"))
            ); // Change lastUpdatedBy to "server"
        }
        // else: Checksum has not changed, all is well
    }
}

public void updateState(ClientState state) {
    NectarServerApplication.getLogger().info("Client " + token.getUuid() + " state updated to: " + state.toString());
    this.state = state;

    MongoCollection<Document> clients = NectarServerApplication.getDb().getCollection("clients");
    clients.updateOne(Filters.eq("uuid", token.getUuid()),
            new Document("$set", new Document("state", state.toInt())));
}

public static int deleteById(MongoCollection<Document> col, Object id) {
    int count = 0;
    Bson filter = Filters.eq("_id", id);
    DeleteResult deleteResult = col.deleteOne(filter);
    count = (int) deleteResult.getDeletedCount();
    return count;
}

public static Document updateById(MongoCollection<Document> col, Object id, Document newDoc,
        boolean repalce) {
    Bson filter = Filters.eq("_id", id);
    if (repalce)
        col.replaceOne(filter, newDoc);
    else
        col.updateOne(filter, new Document("$set", newDoc));
    return newDoc;
}

public void saveUserData(UserData userData) {
    MongoCollection<Document> collection = getDatabase().getCollection("users");

    Document userDocument = new Document("_id", userData.getUniqueId())
            .append("seen", userData.getLastSeen())
            .append("joined", userData.getUserJoined());

    collection.replaceOne(
            Filters.eq("_id", userData.getUniqueId().toString()),
            userDocument,
            (new UpdateOptions()).upsert(true)
    );
}

/**
 * Sets a value to the given config type
 *
 * @param type The config type
 * @param val  The value to be set
 */
public void set(NetworkConfigType type, String val) {
    Object castedVal = ReflectionUtil.safeCast(val);
    connection.upsert(DatabaseType.CLOUD_CONFIG, Filters.eq(DbModifier.CONFIG_CATEGORY.getFieldName(), type.getCategory().getName()),
            new DbQuery().equate(type.getKey(), castedVal).toDocument(), aLong -> {
            });

    // set to moo cache
    MooCache.getInstance().getConfigMap().putAsync(type.getKey(), castedVal);
}

/**
 * Loads the config from the database
 */
public void load() {
    for(NetworkConfigCategory category : NetworkConfigCategory.values()) {
        if(category == NetworkConfigCategory.NONE) {
            load(category, null);
            continue;
        }
        String databaseKey = category.getName();

        // fine the document with given key
        connection.findOne(DatabaseType.CLOUD_CONFIG, Filters.eq(DbModifier.CONFIG_CATEGORY.getFieldName(), databaseKey), document -> {
            if(document == null) {
                // create the document with default values
                Document defaultDocument = new Document();

                defaultDocument.put(DbModifier.CONFIG_CATEGORY.getFieldName(), databaseKey);
                for(NetworkConfigType configType : NetworkConfigType.getConfigTypes(category)) {
                    defaultDocument.put(configType.getKey(), configType.getDefaultValue());
                }

                connection.insert(DatabaseType.CLOUD_CONFIG, defaultDocument);
                load(category, defaultDocument);
                return;
            }

            // otherwise get all the values
            load(category, document);
        });
    }

    // init PunishmentManager
    PunishmentManager.getInstance().init(
            get(NetworkConfigType.PUNISHMENT_CATEGORIES.getKey()),
            get(NetworkConfigType.PUNISHMENT_SUBTYPES.getKey())
    );
}

/**
 * Converts the standard uuid to a binary uuid
 *
 * @param uuid The uuid
 * @return The uuid
 */
private String convertStandardUUID(UUID uuid) {
    String json = Filters.eq("", uuid).toBsonDocument(BsonDocument.class, MongoClient.getDefaultCodecRegistry()).toJson();
    json = json.split(":", 2)[1].replace(" ", "").replace("{", "").replace("}", "");

    return json;
}

@Override
public String getData(long code) {
  if (!containsData(code)) {
    throw new NoSuchElementException("There is no data for that code");
  }

  Document entry = collection.find(Filters.eq("code", code))
      .iterator().next();

  return (String)entry.get("data");
}

/**
 * 与操作
 *
 * @param queries
 * @return
 */
public static MongodbQuery and(MongodbQuery... queries) {
    List<Bson> queryList = new ArrayList<Bson>();
    for (MongodbQuery query : queries) {
        if (query != null && query.getQuery() != null) {
            queryList.add(query.getQuery());
        }
    }
    return new MongodbQuery(Filters.and(queryList));
}

@Override
public Bson filter() {
    List<Bson> filters = Stream.of(
            between(),
            filter("caller", caller),
            filter("callee", callee)
    ).filter(Objects::nonNull).collect(Collectors.toList());

    return Filters.and(filters);
}

@Override
public Bson filter() {
    return Filters.and(
            between(),
            Filters.in("call_id", callIds)
    );
}

public static void saveFailureStatusProcesssed(MongoDatabase mongoDatabase, long id, boolean processed) {
    MongoCollection<Document> failureCollection = mongoDatabase.getCollection("failures");
    failureCollection.replaceOne(
            Filters.eq("tweet_id", id),
            new Document().append("tweet_id", id).append("processed", processed),
            new UpdateOptions().upsert(true));
}

public SyncMap saveMapping(SyncMap map) {
    // TODO : check why this is needed
    if (map.getMapId() == null) {
        map.setMapId(new ObjectId());
    }
    return syncMappings.findOneAndReplace(Filters.eq(SyncAttrs.ID, map.getMapId()), map,
            new FindOneAndReplaceOptions().returnDocument(ReturnDocument.AFTER).upsert(true));
}

@Test
public void testExternalId() {
    Pojo pojo = Pojo.builder().id(null).someOtherProperty("some nice string").build();
    MongoCollection<Pojo> collection = mongoClient.getDatabase("test").getCollection("documents").withDocumentClass(Pojo.class);
    collection.insertOne(pojo);

    Pojo readPojo = collection.find(Filters.eq("_id", pojo.getId())).first();

    Assert.assertNotNull(readPojo);
}

public SyncEvent getPendingEvent(List<String> eventTypes) {
    return syncEvents.findOneAndUpdate(
            Filters.and(Filters.eq(SyncAttrs.STATUS, SyncStatus.PENDING),
                    Filters.in(SyncAttrs.EVENT_TYPE, eventTypes)),
            Updates.set(SyncAttrs.STATUS, SyncStatus.IN_PROGRESS),
            new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER));
}

public void markEventAsCompleted(ObjectId eventId) {
    updateEventStatus(eventId, SyncStatus.COMPLETE);
    SyncEvent event = getEvent(eventId);
    long incompleteCount = syncEvents.count(Filters.and(Filters.ne(SyncAttrs.STATUS, SyncStatus.COMPLETE),
            Filters.ne(SyncAttrs.ID, event.getParentEventId()),
            Filters.eq(SyncAttrs.PARENT_EVENT_ID, event.getParentEventId())));
    if (incompleteCount == 0) {
        updateEventStatus(event.getParentEventId(), SyncStatus.COMPLETE);
        Mailer.sendmail(event.getParentEventId(), null, null, Mailer.COMPLETED);
    }
}