blueflood-core/src/main/java/com/rackspacecloud/blueflood/inputs/processors/TokenDiscoveryWriter.java
public ListenableFuture<Boolean> processTokens(final List<List<IMetric>> input) {

        return getThreadPool().submit(() -> {
            boolean success = true;

            List<Locator> locators = getLocators(input);

            List<Token> tokens = new ArrayList<>();
            tokens.addAll(getUniqueTokens(locators));

            if (tokens.size() > 0) {
                for (TokenDiscoveryIO io : tokenDiscoveryIOs) {
                    try {
                        io.insertDiscovery(tokens);
                    } catch (Exception ex) {
                        getLogger().error(ex.getMessage(), ex);
                        writeErrorMeters.get(io.getClass()).mark();
                        success = false;
                    }
                }
            }

            if (success && tokens.size() > 0) {

                tokens.stream()
                      .filter(token -> !token.isLeaf()) //do not cache leaf nodes
                      .forEach(token -> {

                          //updating token cache
                          TokenCache.getInstance().setTokenCurrent(token);
                      });

                locators.stream()
                        .forEach(locator -> {

                            //updating locator cache
                            LocatorCache.getInstance().setLocatorCurrentInTokenDiscoveryLayer(locator);
                        });
            }

            return success;
        });
    }
public ListenableFuture<Boolean> processTokens(final List<List<IMetric>> input) {

        return getThreadPool().submit(() -> {
            boolean success = true;

            List<Locator> locators = getLocators(input);

            List<Token> tokens = new ArrayList<>();
            tokens.addAll(getUniqueTokens(locators));

            if (tokens.size() > 0) {
                for (TokenDiscoveryIO io : tokenDiscoveryIOs) {
                    try {
                        io.insertDiscovery(tokens);
                    } catch (Exception ex) {
                        getLogger().error(ex.getMessage(), ex);
                        writeErrorMeters.get(io.getClass()).mark();
                        success = false;
                    }
                }
            }

            if (success && tokens.size() > 0) {

                tokens.stream()
                      .filter(token -> !token.isLeaf()) //do not cache leaf nodes
                      .forEach(token -> {

                          //updating token cache
                          TokenCache.getInstance().setTokenCurrent(token);
                      });

                locators.stream()
                        .forEach(locator -> {

                            //updating locator cache
                            LocatorCache.getInstance().setLocatorCurrentInTokenDiscoveryLayer(locator);
                        });
            }

            return success;
        });
    }
public ListenableFuture < Boolean > processTokens ( final List < List < IMetric > > input ) { return getThreadPool ( ) . submit ( ( ) -> { boolean success = true ; List < Locator > locators = getLocators ( input ) ; List < Token > tokens = new ArrayList <> ( ) ; tokens . addAll ( getUniqueTokens ( locators ) ) ; if ( tokens . size ( ) > 0 ) { for ( TokenDiscoveryIO io : tokenDiscoveryIOs ) { try { io . insertDiscovery ( tokens ) ; } catch ( Exception ex ) { getLogger ( ) . error ( ex . getMessage ( ) , ex ) ; writeErrorMeters . get ( io . getClass ( ) ) . mark ( ) ; success = false ; } } } if ( success && tokens . size ( ) > 0 ) { tokens . stream ( ) . filter ( token -> ! token . isLeaf ( ) ) //do not cache leaf nodes . forEach ( token -> { //updating token cache TokenCache . getInstance ( ) . setTokenCurrent ( token ) ; } ) ; locators . stream ( ) . forEach ( locator -> { //updating locator cache LocatorCache . getInstance ( ) . setLocatorCurrentInTokenDiscoveryLayer ( locator ) ; } ) ; } return success ; } ) ; }
For a given batch of metrics, insert unique tokens using {@link TokenDiscoveryIO}.

This methods has the following steps.
1) For a batch of metrics, get all locators, which are not current by calling {@link #getLocators(List)}.
2) For all these locators, get unique tokens, which are not current, by calling {@link #getUniqueTokens(List)}
3) insert tokens
4) After successful insertion, update both {@link LocatorCache} and {@link TokenCache}

@param input
@return
For a given batch of metrics insert unique tokens using { @link TokenDiscoveryIO } .