1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TemplateHaskell #-}
import Control.Monad (forM_)
import Data.ByteString (ByteString)
import Data.Function ((&))
import Data.Text (Text, pack)
import Effectful (Eff, (:>), runEff)
import Effectful.Concurrent (Concurrent, runConcurrent, threadDelay)
import Effectful.Log (Log, localDomain, logInfo_, runLog)
import Effectful.Reader.Static (Reader, asks, runReader)
import Effectful.Resource (runResource)
import Hsm.Core.App (launch)
import Hsm.Core.Env (deriveFromYaml)
import Hsm.Core.Message (body)
import Hsm.Core.Zmq.Client (poll, runClient)
import Streamly.Data.Fold qualified as S (drain)
import Streamly.Data.Stream qualified as S (Stream, fold, mapM)
import System.IO.Echo (withoutInputEcho)
data Env = Env
{ name :: Text
, subEps :: [Text]
, topics :: [Text]
, period :: Word
}
$(deriveFromYaml ''Env)
handle ::
(Concurrent :> es, Log :> es, Reader Env :> es)
=> S.Stream (Eff es) [ByteString]
-> Eff es ()
handle = S.fold S.drain . S.mapM handler
where
receiverDomain = "receiver"
delay = asks period >>= threadDelay . fromIntegral
handler [] = do
localDomain receiverDomain $ logInfo_ "No pulse received yet"
delay
handler ps = do
localDomain receiverDomain
$ forM_ ps
$ logInfo_ . mappend "Received pulse #" . pack . show . body @Word
delay
-- Dummy Poller Service: A proof of concept that polls for "pulses" via ZMQ at
-- a set interval and logs a message each time one is received.
main :: IO ()
main =
launch @Env "dummy-poller" withoutInputEcho $ \env logger level ->
(poll & handle)
& runClient @Env
& runConcurrent
& runLog env.name logger level
& runReader env
& runResource
& runEff
|
