Method call

src/core/IncomingDataPoints.java:305–363

(final Boolean allowed)

     * should be written or not. */
    final class WriteCB implements Callback<Deferred<Object>, Boolean> {
      @Override
      public Deferred<Object> call(final Boolean allowed) throws Exception {
        if (!allowed) {
          return Deferred.fromResult(null);
        }
        

        last_ts = (ms_timestamp ? timestamp : timestamp * 1000);

        long base_time = baseTime();
        long incoming_base_time;
        if (ms_timestamp) {
          // drop the ms timestamp to seconds to calculate the base timestamp
          incoming_base_time = ((timestamp / 1000) - ((timestamp / 1000) % Const.MAX_TIMESPAN));
        } else {
          incoming_base_time = (timestamp - (timestamp % Const.MAX_TIMESPAN));
        }

        if (incoming_base_time - base_time >= Const.MAX_TIMESPAN) {
          // Need to start a new row as we've exceeded Const.MAX_TIMESPAN.
          base_time = updateBaseTime((ms_timestamp ? timestamp / 1000 : timestamp));
        }

        // Java is so stupid with its auto-promotion of int to float.
        final byte[] qualifier = Internal.buildQualifier(timestamp, flags);

        // TODO(tsuna): The following timing is rather useless. First of all,
        // the histogram never resets, so it tends to converge to a certain
        // distribution and never changes. What we really want is a moving
        // histogram so we can see how the latency distribution varies over time.
        // The other problem is that the Histogram class isn't thread-safe and
        // here we access it from a callback that runs in an unknown thread, so
        // we might miss some increments. So let's comment this out until we
        // have a proper thread-safe moving histogram.
        // final long start_put = System.nanoTime();
        // final Callback<Object, Object> cb = new Callback<Object, Object>() {
        // public Object call(final Object arg) {
        // putlatency.add((int) ((System.nanoTime() - start_put) / 1000000));
        // return arg;
        // }
        // public String toString() {
        // return "time put request";
        // }
        // };

        // TODO(tsuna): Add an errback to handle some error cases here.
        if (tsdb.getConfig().enable_appends()) {
          final AppendDataPoints kv = new AppendDataPoints(qualifier, value);
          final AppendRequest point = new AppendRequest(tsdb.table, row, TSDB.FAMILY, 
                  AppendDataPoints.APPEND_COLUMN_QUALIFIER, kv.getBytes());
          point.setDurable(!batch_import);
          return tsdb.client.append(point);/* .addBoth(cb) */
        } else {
          final PutRequest point = RequestBuilder.buildPutRequest(tsdb.getConfig(), tsdb.table, row, TSDB.FAMILY,
              qualifier, value, timestamp);
          point.setDurable(!batch_import);
          return tsdb.client.put(point)/* .addBoth(cb) */;
        }