Java 类com.datastax.driver.core.ResultSet 实例源码

@Override
public Response getRecordById(String keyspaceName, String tableName, String identifier) {
  long startTime = System.currentTimeMillis();
  ProjectLogger.log("Cassandra Service getRecordById method started at ==" + startTime,
      LoggerEnum.PERF_LOG);
  Response response = new Response();
  try {
    Select selectQuery = QueryBuilder.select().all().from(keyspaceName, tableName);
    Where selectWhere = selectQuery.where();
    Clause clause = QueryBuilder.eq(Constants.IDENTIFIER, identifier);
    selectWhere.and(clause);
    ResultSet results = connectionManager.getSession(keyspaceName).execute(selectQuery);
    response = CassandraUtil.createResponse(results);
  } catch (Exception e) {
    ProjectLogger.log(Constants.EXCEPTION_MSG_FETCH + tableName + " : " + e.getMessage(), e);
    throw new ProjectCommonException(ResponseCode.SERVER_ERROR.getErrorCode(),
        ResponseCode.SERVER_ERROR.getErrorMessage(), ResponseCode.SERVER_ERROR.getResponseCode());
  }
  long stopTime = System.currentTimeMillis();
  long elapsedTime = stopTime - startTime;
  ProjectLogger.log("Cassandra Service getRecordById method end at ==" + stopTime
      + " ,Total time elapsed = " + elapsedTime, LoggerEnum.PERF_LOG);
  return response;
}

@Override
public Response getPropertiesValueById(String keyspaceName, String tableName, String id,
    String... properties) {
  long startTime = System.currentTimeMillis();
  ProjectLogger.log("Cassandra Service getPropertiesValueById method started at ==" + startTime,
      LoggerEnum.PERF_LOG);
  Response response = new Response();
  try {
    String selectQuery = CassandraUtil.getSelectStatement(keyspaceName, tableName, properties);
    PreparedStatement statement = connectionManager.getSession(keyspaceName).prepare(selectQuery);
    BoundStatement boundStatement = new BoundStatement(statement);
    ResultSet results =
        connectionManager.getSession(keyspaceName).execute(boundStatement.bind(id));
    response = CassandraUtil.createResponse(results);
  } catch (Exception e) {
    ProjectLogger.log(Constants.EXCEPTION_MSG_FETCH + tableName + " : " + e.getMessage(), e);
    throw new ProjectCommonException(ResponseCode.SERVER_ERROR.getErrorCode(),
        ResponseCode.SERVER_ERROR.getErrorMessage(), ResponseCode.SERVER_ERROR.getResponseCode());
  }
  long stopTime = System.currentTimeMillis();
  long elapsedTime = stopTime - startTime;
  ProjectLogger.log("Cassandra Service getPropertiesValueById method end at ==" + stopTime
      + " ,Total time elapsed = " + elapsedTime, LoggerEnum.PERF_LOG);
  return response;
}

@Override
    public Status selectByDeviceAndSensor(TsPoint point, Double max, Double min, Date startTime, Date endTime) {
        long costTime = 0L;
        Cluster cluster = null;
        try {
            cluster = Cluster.builder().addContactPoint(CASSADRA_URL).build();
            Session session = cluster.connect(KEY_SPACE_NAME);
            String createIndexCql = "CREATE INDEX IF NOT EXISTS value_index ON " + TABLE_NAME + "(value)";
//          System.out.println(createIndexCql);
            long startTime1 = System.nanoTime();
            session.execute(createIndexCql);
            String selectCql = "SELECT * FROM point WHERE device_code='" + point.getDeviceCode() + "' and sensor_code='"
                    + point.getSensorCode() + "' and value<" + max + " and value>" + min + " and timestamp>="
                    + startTime.getTime() + " and timestamp<=" + endTime.getTime() + " ALLOW FILTERING";
//          System.out.println(selectCql);
            ResultSet rs = session.execute(selectCql);
            long endTime1 = System.nanoTime();
            costTime = endTime1 - startTime1;
        } finally {
            if (cluster != null)
                cluster.close();
        }
//      System.out.println("此次查询消耗时间[" + costTime / 1000 + "]s");
        return Status.OK(costTime);
    }

@Override
    public Status selectMaxByDeviceAndSensor(String deviceCode, String sensorCode, Date startTime, Date endTime) {
        long costTime = 0L;
        Cluster cluster = null;
        try {
//          cluster = Cluster.builder().addContactPoint(CASSADRA_URL).build();
//          Session session = cluster.connect(KEY_SPACE_NAME);
            Session session = SessionManager.getSession();
            String selectCql = "SELECT MAX(value) FROM point WHERE device_code='" + deviceCode + "' and sensor_code='"
                    + sensorCode + "' and timestamp>=" + startTime.getTime() + " and timestamp<=" + endTime.getTime()
                    + " ALLOW FILTERING";
            long startTime1 = System.nanoTime();
//          System.out.println("aaa");
            ResultSet rs = session.execute(selectCql);
//          System.out.println("bbb");
            long endTime1 = System.nanoTime();
            costTime = endTime1 - startTime1;
        } finally {
            if (cluster != null)
                cluster.close();
        }
//      System.out.println("此次查询消耗时间[" + costTime / 1000 + "]s");
        return Status.OK(costTime);
    }

@Override
    public Status selectAvgByDeviceAndSensor(String deviceCode, String sensorCode, Date startTime, Date endTime) {
        long costTime = 0L;
        Cluster cluster = null;
        try {
            cluster = Cluster.builder().addContactPoint(CASSADRA_URL).build();
            Session session = cluster.connect(KEY_SPACE_NAME);
            String selectCql = "SELECT AVG(value) FROM point WHERE device_code='" + deviceCode + "' and sensor_code='"
                    + sensorCode + "' and timestamp>=" + startTime.getTime() + " and timestamp<=" + endTime.getTime()
                    + " ALLOW FILTERING";
//          System.out.println(selectCql);
            long startTime1 = System.nanoTime();
            ResultSet rs = session.execute(selectCql);
            long endTime1 = System.nanoTime();
            costTime = endTime1 - startTime1;
        } finally {
            if (cluster != null)
                cluster.close();
        }
//      System.out.println("此次查询消耗时间[" + costTime / 1000 + "]s");
        return Status.OK(costTime);
    }

@Override
    public Status selectCountByDeviceAndSensor(String deviceCode, String sensorCode, Date startTime, Date endTime) {
        long costTime = 0L;
        Cluster cluster = null;
        try {
            cluster = Cluster.builder().addContactPoint(CASSADRA_URL).build();
            Session session = cluster.connect(KEY_SPACE_NAME);
            String selectCql = "SELECT COUNT(*) FROM point WHERE device_code='" + deviceCode + "' and sensor_code='"
                    + sensorCode + "' and timestamp>=" + startTime.getTime() + " and timestamp<=" + endTime.getTime()
                    + " ALLOW FILTERING";
//          System.out.println(selectCql);
            long startTime1 = System.nanoTime();
            ResultSet rs = session.execute(selectCql);
            long endTime1 = System.nanoTime();
            costTime = endTime1 - startTime1;
        } finally {
            if (cluster != null)
                cluster.close();
        }
//      System.out.println("此次查询消耗时间[" + costTime / 1000 + "]s");
        return Status.OK(costTime);
    }

protected ListenableFuture<List<D>> findListByStatementAsync(Statement statement) {
    if (statement != null) {
        statement.setConsistencyLevel(cluster.getDefaultReadConsistencyLevel());
        ResultSetFuture resultSetFuture = getSession().executeAsync(statement);
        return Futures.transform(resultSetFuture, new Function<ResultSet, List<D>>() {
            @Nullable
            @Override
            public List<D> apply(@Nullable ResultSet resultSet) {
                Result<E> result = getMapper().map(resultSet);
                if (result != null) {
                    List<E> entities = result.all();
                    return DaoUtil.convertDataList(entities);
                } else {
                    return Collections.emptyList();
                }
            }
        });
    }
    return Futures.immediateFuture(Collections.emptyList());
}

private Optional<ComponentDescriptor> saveIfNotExist(ComponentDescriptorEntity entity) {
    if (entity.getId() == null) {
        entity.setId(UUIDs.timeBased());
    }

    ResultSet rs = executeRead(QueryBuilder.insertInto(getColumnFamilyName())
            .value(ModelConstants.ID_PROPERTY, entity.getId())
            .value(ModelConstants.COMPONENT_DESCRIPTOR_NAME_PROPERTY, entity.getName())
            .value(ModelConstants.COMPONENT_DESCRIPTOR_CLASS_PROPERTY, entity.getClazz())
            .value(ModelConstants.COMPONENT_DESCRIPTOR_TYPE_PROPERTY, entity.getType())
            .value(ModelConstants.COMPONENT_DESCRIPTOR_SCOPE_PROPERTY, entity.getScope())
            .value(ModelConstants.COMPONENT_DESCRIPTOR_CONFIGURATION_DESCRIPTOR_PROPERTY, entity.getConfigurationDescriptor())
            .value(ModelConstants.COMPONENT_DESCRIPTOR_ACTIONS_PROPERTY, entity.getActions())
            .value(ModelConstants.SEARCH_TEXT_PROPERTY, entity.getSearchText())
            .ifNotExists()
    );
    if (rs.wasApplied()) {
        return Optional.of(DaoUtil.getData(entity));
    } else {
        return Optional.empty();
    }
}

/**
 * userLikedPost
 * @param postId
 * @param userId
 * @return true if param userId liked param postId and false otherwise
 * @throws Exception
 */
public static boolean userLikedPost (
  UUID postId,
  UUID userId) throws Exception {

  ResultSet resultSet =
    PostLikesTime.i().executeSyncSelect(
      postId,
      userId);

  if (resultSet.isExhausted() == true) {

    return false;
  }

  return true;
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param userid
 * @param devicetoken
 * @param authcode
 * @param accesstoken
 * @param refreshtoken
 * @param ttl
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object userid,
  Object devicetoken,
  Object authcode,
  Object accesstoken,
  Object refreshtoken,
  Object ttl) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      userid,
      devicetoken,
      authcode,
      accesstoken,
      refreshtoken,
      ttl);
}

/**
 * 描述: 查询数据表字段名(key:字段名,value:字段类型名)
 * 时间: 2017年11月15日 上午11:29:32
 * @author yi.zhang
 * @param table 表名
 * @return
 */
public Map<String,String> queryColumns(String table){
    try {
        String sql = "select * from "+table;
        ResultSet rs = session.execute(sql);
        ColumnDefinitions rscd = rs.getColumnDefinitions();
        int count = rscd.size();
        Map<String,String> reflect = new HashMap<String,String>();
        for (int i = 0; i < count; i++) {
            String column = rscd.getName(i);
            String type = rscd.getType(i).getName().name().toLowerCase();
            reflect.put(column, type);
        }
        return reflect;
    } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return null;
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param yearweek
 * @param rank
 * @param userid
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object yearweek,
  Object rank,
  Object userid) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      yearweek,
      rank,
      userid);
}

/**
 * executeSyncSelectTopSmallerThanLimit
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectTopSmallerThanLimit Query synchronously
 * @param yearweekgridid
 * @param rank
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectTopSmallerThanLimit (
  Object yearweekgridid,
  Object rank) throws Exception {

  return
    this.getQuery(kSelectTopSmallerThanLimitName).executeSync(
      yearweekgridid,
      rank);
}

@Override
public ListenableFuture<Optional<AttributeKvEntry>> find(EntityId entityId, String attributeType, String attributeKey) {
    Select.Where select = select().from(ATTRIBUTES_KV_CF)
            .where(eq(ENTITY_TYPE_COLUMN, entityId.getEntityType()))
            .and(eq(ENTITY_ID_COLUMN, entityId.getId()))
            .and(eq(ATTRIBUTE_TYPE_COLUMN, attributeType))
            .and(eq(ATTRIBUTE_KEY_COLUMN, attributeKey));
    log.trace("Generated query [{}] for entityId {} and key {}", select, entityId, attributeKey);
    return Futures.transform(executeAsyncRead(select), (Function<? super ResultSet, ? extends Optional<AttributeKvEntry>>) input ->
                    Optional.ofNullable(convertResultToAttributesKvEntry(attributeKey, input.one()))
            , readResultsProcessingExecutor);
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param yearweekcountrycode
 * @param rank
 * @param userid
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object yearweekcountrycode,
  Object rank,
  Object userid) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      yearweekcountrycode,
      rank,
      userid);
}

/**
 * executeSyncSelectAfter
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectAfter Query synchronously
 * @param yearmonthdayuserid
 * @param logtime
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectAfter (
  Object yearmonthdayuserid,
  Object logtime) throws Exception {

  return
    this.getQuery(kSelectAfterName).executeSync(
      yearmonthdayuserid,
      logtime);
}

private void assertResult(ResultSet set) {
  List<Row> results = set.all();
  assertThat(1).isEqualTo(results.size());
  Row row1 = results.get(0);
  boolean column1 = row1.getBool("applied");
  assertThat(column1).isTrue();
}

/**
 * executeSyncSelectTopEqual
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectTopEqual Query synchronously
 * @param yearweekgridid
 * @param rank
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectTopEqual (
  Object yearweekgridid,
  Object rank) throws Exception {

  return
    this.getQuery(kSelectTopEqualName).executeSync(
      yearweekgridid,
      rank);
}

/**
 * executeSyncSelectBefore
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectBefore Query synchronously
 * @param yearmonthdayuserid
 * @param logtime
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectBefore (
  Object yearmonthdayuserid,
  Object logtime) throws Exception {

  return
    this.getQuery(kSelectBeforeName).executeSync(
      yearmonthdayuserid,
      logtime);
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param yearweek
 * @param rank
 * @param userid
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object yearweek,
  Object rank,
  Object userid) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      yearweek,
      rank,
      userid);
}

/**
 * executeSyncSelectBefore
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectBefore Query synchronously
 * @param yearmonthdayuserid
 * @param logtime
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectBefore (
  Object yearmonthdayuserid,
  Object logtime) throws Exception {

  return
    this.getQuery(kSelectBeforeName).executeSync(
      yearmonthdayuserid,
      logtime);
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param messageid
 * @param message
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object messageid,
  Object message) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      messageid,
      message);
}

/**
 * executeSyncIncrementValue
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes IncrementValue Query synchronously
 * @param userscount
 * @param yearmonthdayairportcode
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncIncrementValue (
  Object userscount,
  Object yearmonthdayairportcode) throws Exception {

  return
    this.getQuery(kIncrementValueName).executeSync(
      userscount,
      yearmonthdayairportcode);
}

/**
 * executeSyncSelectTopSmallerThanOrEqualLimit
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectTopSmallerThanOrEqualLimit Query synchronously
 * @param yearweekgridid
 * @param rank
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectTopSmallerThanOrEqualLimit (
  Object yearweekgridid,
  Object rank) throws Exception {

  return
    this.getQuery(kSelectTopSmallerThanOrEqualLimitName).executeSync(
      yearweekgridid,
      rank);
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param photoid
 * @param photothumnbnail
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object photoid,
  Object photothumnbnail) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      photoid,
      photothumnbnail);
}

/**
 * executeSyncIncrementNumber
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes IncrementNumber Query synchronously
 * @param number
 * @param userid
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncIncrementNumber (
  Object number,
  Object userid) throws Exception {

  return
    this.getQuery(kIncrementNumberName).executeSync(
      number,
      userid);
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param postid
 * @param userid
 * @param commenttime
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object postid,
  Object userid,
  Object commenttime) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      postid,
      userid,
      commenttime);
}

/**
 * executeSyncSelectAtOrBeforeTimeLimit
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectAtOrBeforeTimeLimit Query synchronously
 * @param postid
 * @param commenttime
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectAtOrBeforeTimeLimit (
  Object postid,
  Object commenttime) throws Exception {

  return
    this.getQuery(kSelectAtOrBeforeTimeLimitName).executeSync(
      postid,
      commenttime);
}

/**
 * executeSyncSelectTopSmallerThanLimit
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectTopSmallerThanLimit Query synchronously
 * @param yearweekcountrycode
 * @param rank
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectTopSmallerThanLimit (
  Object yearweekcountrycode,
  Object rank) throws Exception {

  return
    this.getQuery(kSelectTopSmallerThanLimitName).executeSync(
      yearweekcountrycode,
      rank);
}

public static ResultSet batch(Session session, BatchStatement.Type type, Consumer<BatchStatement> transaction) {
    BatchStatement batchStatement = new BatchStatement(type);
    transaction.accept(batchStatement);
    if (batchStatement.size() == 0) return null;
    if (batchStatement.size() > 1) {
        return session.execute(batchStatement);
    } else {
        Statement statement = Iterables.getOnlyElement(batchStatement.getStatements());
        return session.execute(statement);
    }
}

private List<Data> execute(PreparedStatement statement, boolean reverse, Object... parameters) {
    BoundStatement boundStatement = bind(statement, parameters);
    ResultSet resultSet = session.execute(boundStatement);
    List<Data> data = mapper.map(resultSet).all();
    if (!data.isEmpty() && reverse) {
        data = Lists.reverse(data);
    }
    return data;
}

/**
 * executeSyncSelect
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Select Query synchronously
 * @param postid
 * @param userid
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelect (
  Object postid,
  Object userid) throws Exception {

  return
    this.getQuery(kSelectName).executeSync(
      postid,
      userid);
}

/**
 * executeSyncSelectAfterLimit
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectAfterLimit Query synchronously
 * @param yearmonthday
 * @param registrationtime
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectAfterLimit (
  Object yearmonthday,
  Object registrationtime) throws Exception {

  return
    this.getQuery(kSelectAfterLimitName).executeSync(
      yearmonthday,
      registrationtime);
}

/**
 * executeSyncDelete
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Delete Query synchronously
 * @param userid
 * @param followeruserid
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncDelete (
  Object userid,
  Object followeruserid) throws Exception {

  return
    this.getQuery(kDeleteName).executeSync(
      userid,
      followeruserid);
}

/**
 * Get entity key by entity id.
 *
 * @param entityId entity id
 * @param tenant   tenant name
 * @return entity key
 */
public String getKeyById(Long entityId, String tenant) {
    Select select = QueryBuilder.select(ENTITY_KEY_COL).from(tenant, VIEW_ID_KEY);
    select.where(eq(ENTITY_ID_COL, entityId));
    ResultSet resultSet = session.execute(select);
    Row row = resultSet.one();
    return row == null ? null : row.getString(ENTITY_KEY_COL);
}

@Override
protected void processRequest (final Request request) throws Exception {

  // use the following request Object to process the request and set
  //   the response to be returned
  RequestGetPostPhotoId requestGetPostPhotoId =
    (RequestGetPostPhotoId)request.getRequestJsonBody();

  // set/check post_id

  UUID postId = UUID.fromString(requestGetPostPhotoId.post_id);

  if (CheckersInl.postExists(postId) == false) {

    throw new BadRequestException(
      413,
      1,
      "Post with post_id ["
        + requestGetPostPhotoId.post_id
        + "] doesn't exist, request issued by user_id ["
        + requestGetPostPhotoId.user_id
        + "] from device_token ["
        + requestGetPostPhotoId.device_token
        +"]",
      ExceptionClass.INVALID);
  }

  // select post's photo_id from the database

  ResultSet resultSet =
    Posts.i().executeSyncSelectPhotoId(postId);

  String photoId =
    resultSet.one().getUUID(Posts.kPhotoIdColumnName).toString();

  // set response
  ((ResponseGetPostPhotoId)request.getResponseBody() ).set(
    requestGetPostPhotoId.request_tracking_id,
    photoId);
}

/**
 * executeSyncSelectDuring
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes SelectDuring Query synchronously
 * @param yearmonthdayuserid
 * @param logtimestart
 * @param logtimeend
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncSelectDuring (
  Object yearmonthdayuserid,
  Object logtimestart,
  Object logtimeend) throws Exception {

  return
    this.getQuery(kSelectDuringName).executeSync(
      yearmonthdayuserid,
      logtimestart,
      logtimeend);
}

/**
 * executeSyncInsert
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes Insert Query synchronously
 * @param jobid
 * @param jobtime
 * @param job
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncInsert (
  Object jobid,
  Object jobtime,
  Object job) throws Exception {

  return
    this.getQuery(kInsertName).executeSync(
      jobid,
      jobtime,
      job);
}

/**
 * executeSyncRemoveDeviceToken
 * BLOCKING-METHOD: blocks till the ResultSet is ready
 * executes RemoveDeviceToken Query synchronously
 * @param devicetokens
 * @param userid
 * @return ResultSet
 * @throws Exception
 */
public ResultSet executeSyncRemoveDeviceToken (
  Object devicetokens,
  Object userid) throws Exception {

  return
    this.getQuery(kRemoveDeviceTokenName).executeSync(
      devicetokens,
      userid);
}

protected EntityResultSet<E> saveWithResult(E entity) {
    log.debug("Save entity {}", entity);
    if (entity.getId() == null) {
        entity.setId(UUIDs.timeBased());
    } else if (isDeleteOnSave()) {
        removeById(entity.getId());
    }
    Statement saveStatement = getSaveQuery(entity);
    saveStatement.setConsistencyLevel(cluster.getDefaultWriteConsistencyLevel());
    ResultSet resultSet = executeWrite(saveStatement);
    return new EntityResultSet<>(resultSet, entity);
}