[HWKMETRICS-74] small refactoring for storing data

hawkular · May 19, 2015 · c598865 · c598865
1 parent b030c1d
commit c598865
Showing 1 changed file with 16 additions and 16 deletions.
diff --git a/...-impl/src/main/java/org/hawkular/metrics/core/impl/cassandra/MetricsServiceCassandra.java b/...-impl/src/main/java/org/hawkular/metrics/core/impl/cassandra/MetricsServiceCassandra.java
@@ -610,22 +610,22 @@ public Observable<Void> addGaugeData(Observable<Gauge> gaugeObservable) {
         // pass. This means that the behavior basically remains the same as before. The idea here is to implement a
         // pub/sub workflow.
 
-        return Observable.create(subscriber -> {
-            Map<Gauge, ResultSetFuture> insertsMap = new HashMap<>();
-            gaugeObservable.subscribe(
-                    gauge -> {
-                        int ttl = getTTL(gauge);
-                        insertsMap.put(gauge, dataAccess.insertData(gauge, ttl));
-                    },
-                    t -> logger.warn("There was an error receive gauge data to insert", t),
-                    () -> {
-                        List<ResultSetFuture> inserts = Lists.newArrayList(insertsMap.values());
-                        inserts.add(dataAccess.updateMetricsIndex(ImmutableList.copyOf(insertsMap.keySet())));
-                        ListenableFuture<List<ResultSet>> insertsFuture = Futures.allAsList(inserts);
-                        Observable<List<ResultSet>> dataInserted = RxUtil.from(insertsFuture, metricsTasks);
-                        dataInserted.subscribe(new VoidSubscriber<>(subscriber));
-                    });
-        });
+        return Observable.create(subscriber -> gaugeObservable
+                        .reduce(new HashMap<>(), (HashMap<Gauge, ResultSetFuture> map, Gauge gauge) -> {
+                            int ttl = getTTL(gauge);
+                            ResultSetFuture future = dataAccess.insertData(gauge, ttl);
+                            map.put(gauge, future);
+                            return map;
+                        })
+                        .map(insertsMap -> {
+                            List<ResultSetFuture> inserts = Lists.newArrayList(insertsMap.values());
+                            inserts.add(dataAccess.updateMetricsIndex(ImmutableList.copyOf(insertsMap.keySet())));
+                            return inserts;
+                        })
+                        .map(Futures::allAsList)
+                        .map(insertsFuture -> RxUtil.from(insertsFuture, metricsTasks))
+                        .subscribe(new VoidSubscriber<>(subscriber))
+        );
     }
 
     @Override