core/metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccessImpl.java
private Observable.Transformer<BoundStatement, Integer> applyMicroBatching() {
        return tObservable -> tObservable
                .groupBy(b -> {
                    ByteBuffer routingKey = b.getRoutingKey(ProtocolVersion.NEWEST_SUPPORTED,
                            codecRegistry);
                    Token token = metadata.newToken(routingKey);
                    for (TokenRange tokenRange : session.getCluster().getMetadata().getTokenRanges()) {
                        if (tokenRange.contains(token)) {
                            return tokenRange;
                        }
                    }
                    log.warn("Unable to find any Cassandra node to insert token " + token.toString());
                    return session.getCluster().getMetadata().getTokenRanges().iterator().next();
                })
                .flatMap(g -> g.compose(new BoundBatchStatementTransformer()))
                .flatMap(batch -> rxSession
                        .execute(batch)
                        .compose(applyInsertRetryPolicy())
                        .map(resultSet -> batch.size())
                );
    }
private Observable.Transformer<BoundStatement, Integer> applyMicroBatching() {
        return tObservable -> tObservable
                .groupBy(b -> {
                    ByteBuffer routingKey = b.getRoutingKey(ProtocolVersion.NEWEST_SUPPORTED,
                            codecRegistry);
                    Token token = metadata.newToken(routingKey);
                    for (TokenRange tokenRange : session.getCluster().getMetadata().getTokenRanges()) {
                        if (tokenRange.contains(token)) {
                            return tokenRange;
                        }
                    }
                    log.warn("Unable to find any Cassandra node to insert token " + token.toString());
                    return session.getCluster().getMetadata().getTokenRanges().iterator().next();
                })
                .flatMap(g -> g.compose(new BoundBatchStatementTransformer()))
                .flatMap(batch -> rxSession
                        .execute(batch)
                        .compose(applyInsertRetryPolicy())
                        .map(resultSet -> batch.size())
                );
    }
private Observable . Transformer < BoundStatement , Integer > applyMicroBatching ( ) { return tObservable -> tObservable . groupBy ( b -> { ByteBuffer routingKey = b . getRoutingKey ( ProtocolVersion . NEWEST_SUPPORTED , codecRegistry ) ; Token token = metadata . newToken ( routingKey ) ; for ( TokenRange tokenRange : session . getCluster ( ) . getMetadata ( ) . getTokenRanges ( ) ) { if ( tokenRange . contains ( token ) ) { return tokenRange ; } } log . warn ( "Unable to find any Cassandra node to insert token " + token . toString ( ) ) ; return session . getCluster ( ) . getMetadata ( ) . getTokenRanges ( ) . iterator ( ) . next ( ) ; } ) . flatMap ( g -> g . compose ( new BoundBatchStatementTransformer ( ) ) ) . flatMap ( batch -> rxSession . execute ( batch ) . compose ( applyInsertRetryPolicy ( ) ) . map ( resultSet -> batch . size ( ) ) ) ; }
/*
Applies micro-batching capabilities by taking advantage of token ranges in the Cassandra
/ * Applies micro - batching capabilities by taking advantage of token ranges in the Cassandra