Method process

analytics/src/main/java/com/chain/analytics/Importer.java:199–249

Processes transactions, reading them from the transaction feed and inserting them into the configured Oracle database. This function blocks indefinitely. Errors are logged to log4j2. TODO(jackson): Support propagating persistent errors?

()

   * TODO(jackson): Support propagating persistent errors?
   */
  public void process() {
    ThreadContext.put("feed_id", mFeed.id);
    ThreadContext.put("feed_filter", mFeed.filter);

    while (true) {
      try {
        ThreadContext.put("feed_after", mFeed.after);

        // Retrieve another page of transactions matching the feed.
        final Transaction.Items page =
            new QueryBuilder()
                .setFilter(mFeed.filter)
                .setAfter(mFeed.after)
                .setTimeout(DEFAULT_TIMEOUT_MILLIS)
                .setAscendingWithLongPoll()
                .execute(mChain);

        // Commit the batch of transactions to the database.
        try {
          processBatch(page.list);
        } catch (SQLException ex) {
          logger.catching(ex);
          // Skip the ack so we re-fetch this page and try again.
          // TODO(jackson): Do we need any backoff? Don't want to
          // hammer Chain Core or the Oracle database if there's
          // an issue.
          // TODO(jackson): Add retry logic in processBatch so that
          // we don't re-query Chain Core if we don't need to?
          continue;
        }

        // Acknowledge that we've processed the entire page.
        Map<String, Object> requestBody = new TreeMap<>();
        requestBody.put("id", mFeed.id);
        requestBody.put("previous_after", mFeed.after);
        requestBody.put("after", page.next.after);
        mFeed = mChain.request("update-transaction-feed", requestBody, Transaction.Feed.class);
      } catch (APIException ex) {
        // If there was an issue retrieving the transactions,
        // log it. If the request just timed out, no matching
        // transactions were committed so just silently ignore it.
        if (!"CH001".equals(ex.code)) {
          logger.catching(ex);
        }
      } catch (ChainException ex) {
        logger.catching(ex);
      } finally {
        ThreadContext.remove("feed_after");
      }
    }
  }

  // processBatch inserts a batch of transactions into the Oracle
  // database.