Leonhard Riedißer L7R7

Building a generic element counter that transiently counts the elements that go through a fs2 Stream

During our migration from Akka to the Typelevel stack, we're switching from Akka Streams to fs2. For one of our services, we needed something that counts the elements that go through a stream and writes a metric for it. Previously that was done using an AtomicInteger, which worked alright but now we're in a pure functional land and writing metrics as well as using mutable variables is considered impure and has to be implemented as an IO action, which rules out AtomicInteger (one could use cats.effect.Ref as an almost-identical replacement, but we can do better than that).
Implementing this turned out to be a nice little exercise, so I thought it makes sense to write down how we did it.

So, suppose we have a stream of product data:

	[WARNING] A Cats Effect worker thread was detected to be in a blocked state (TIMED_WAITING)
	at app//scala.collection.immutable.List.collect(List.scala:270)
	at app//cats.data.NonEmptyList.collect(NonEmptyList.scala:261)
	at app//com.example.FooSelector.$anonfun$select$2(FooSelector.scala:28)
	at app//com.example.FooSelector$$Lambda$3392/0x00000008016c2f18.apply(Unknown Source)
	at app//scala.collection.StrictOptimizedIterableOps.flatMap(StrictOptimizedIterableOps.scala:118)
	at app//scala.collection.StrictOptimizedIterableOps.flatMap$(StrictOptimizedIterableOps.scala:105)
	at app//scala.collection.immutable.Vector.flatMap(Vector.scala:113)
	at app//com.example.FooSelector.select(FooSelector.scala:27)
	at app//com.example.FooDomainService.$anonfun$toBlabla$1(FooDomainService.scala:100)

	[WARNING] A Cats Effect worker thread was detected to be in a blocked state (TIMED_WAITING)
	at app//cats.syntax.FlatMapOps$.$anonfun$$greater$greater$1(flatMap.scala:54)
	at app//cats.syntax.FlatMapOps$$$Lambda$1016/0x000000080123b0f0.apply(Unknown Source)
	at app//cats.effect.IOFiber.next$2(IOFiber.scala:377)
	at app//cats.effect.IOFiber.runLoop(IOFiber.scala:388)
	at app//cats.effect.IOFiber.asyncContinueSuccessfulR(IOFiber.scala:1353)
	at app//cats.effect.IOFiber.run(IOFiber.scala:119)
	at app//cats.effect.unsafe.WorkerThread.run(WorkerThread.scala:567)

	[WARNING] A Cats Effect worker thread was detected to be in a blocked state (TIMED_WAITING)

	{-# LANGUAGE DataKinds #-}

	import Test.Syd

	data Outers = Outers
	{ foo :: Int,
	bar :: String
	}

	fullSpec :: TestDef (Outers : otherOuters) ()

	{-
	compiles with:

	extra-deps:
	- github: haskell-servant/servant
	commit: 1fba9dc6048cea6184964032b861b052cd54878c
	subdirs:
	- servant
	- servant-server
	- servant-auth/servant-auth

	data FooBarBaz = Foo Int String \| Bar Bool Text \| Baz Bool Bool Bool

	instance HasCodec FooBarBaz where
	codec =
	dimapCodec f g $
	disjointEitherCodec
	( object "Foo" $
	(,)
	<$> requiredField' "int" .= fst
	<*> requiredField' "string" .= snd

	instance FromJSON URI where
	parseJSON = withText "URI" $ \v -> maybe (fail "Bad URI") pure (parseURI (toString v))

	class Profunctor p where
	{-# MINIMAL dimap \| lmap, rmap #-}
	dimap :: (c -> a) -> (b -> d) -> p a b -> p c d
	dimap ca bd = rmap bd . lmap ca
	lmap :: (c -> a) -> p a b -> p c b
	lmap ca = dimap ca id
	rmap :: (b -> c) -> p a b -> p a c
	rmap bc = dimap id bc

	mapInput :: (input -> out) -> (newInput -> input) -> (newInput -> out)

	instance Append (List a) where
	append Empty as = as
	append as Empty = as
	append (Cons a as) bs = Cons a (append as bs)

	instance (Append a) => Append (Maybe a) where
	append (Just a1) (Just a2) = Just (append a1 a2)
	append ma Nothing = ma
	append Nothing ma = ma

	takeEven :: [a] -> [a]
	takeEven [] = []
	takeEven [a] = [a]
	takeEven (a : _ : as) = a : takeEven as