Refactor aggregateListBy and add some tracing

Author: harendra-kumar

lib/Streamly/Metrics/Channel/Common.hs

@@ -23,26 +23,31 @@ import Prelude hiding (showList)
 -- Event processing
 -------------------------------------------------------------------------------
 
+-- XXX We can use incremental window folds instead. We do not need to collect
+-- all stats and then calculate.
 aggregateListBy :: (MonadAsync m, Ord k, Fractional a) =>
     Double -> Int -> Stream m (AbsTime, (k, [a])) -> Stream m (k, [a])
 aggregateListBy timeout batchsize stream =
     fmap (second fromJust)
         $ Stream.filter (isJust . snd)
-        $ Stream.classifySessionsBy
-            0.1 False (return . (> 1000)) timeout f stream
+        --  $ Stream.trace (\x -> liftIO $ putStrLn $ "after classify: " ++ show x)
+        $ Stream.classifySessionsBy 0.1 False (return . (> 1000)) timeout f
+        --  $ Stream.trace (liftIO . print)
+        $ stream
 
     where
 
     scale Nothing _ = Nothing
-    scale (Just xs) count = Just $ map (/ count) xs
+    scale (Just xs) cnt = Just $ map (/ cnt) xs
 
-    f =
-        Fold.teeWithFst
-            scale
-            (Fold.take batchsize (Fold.foldl1' (zipWith (+))))
-            (Fold.lmap (const 1) Fold.sum)
+    addMetrics = Fold.foldl1' (zipWith (+))
+    collectBatch = Fold.take batchsize addMetrics
+    count = fmap fromIntegral Fold.length
+
+    f = Fold.teeWithFst scale collectBatch count
 
 printKV :: (MonadIO m, Show k, Show a, Indexable a) => Stream m (k, [a]) -> m b
 printKV stream =
     let f (k, xs) = liftIO $ putStrLn $ show k ++ ":\n" ++ showList xs
-     in Stream.fold (Fold.drainMapM f) stream >> error "printChannel: Metrics channel closed"
+     in Stream.fold (Fold.drainMapM f) stream
+        >> error "printChannel: Metrics channel closed"