72 lines
2.5 KiB
Haskell
72 lines
2.5 KiB
Haskell
{-# LANGUAGE LambdaCase #-}
|
|
{-# LANGUAGE OverloadedStrings #-}
|
|
|
|
module HomeAssistant.Runtime.Bus
|
|
( Bus(..)
|
|
, CallIdGen(..)
|
|
, mkCallIdGen
|
|
, withBus
|
|
, channelHassEval
|
|
) where
|
|
|
|
import Control.Concurrent.STM
|
|
( TChan
|
|
, TVar
|
|
, atomically
|
|
, newBroadcastTChanIO
|
|
, newTChanIO
|
|
, newTVarIO
|
|
, writeTChan
|
|
)
|
|
import Data.Aeson (Value)
|
|
import Data.IORef (atomicModifyIORef', newIORef)
|
|
import HomeAssistant.Controller (HASSEff (..), Service)
|
|
import Network.WebSockets (Connection)
|
|
import Katip (LogEnv, closeScribes, mkHandleScribe, ColorStrategy (..), permitItem, Severity (..), Verbosity (V2), registerScribe, defaultScribeSettings, initLogEnv, ls, sl, logFM, katipAddContext, KatipContext)
|
|
import Control.Exception (bracket)
|
|
import System.IO (stdout)
|
|
import Data.UUID (toText)
|
|
import AFRP (Request(..))
|
|
import Control.Monad.IO.Class (MonadIO, liftIO)
|
|
|
|
-- | Shared runtime state: inbound is a broadcast channel (controllers
|
|
-- read from 'dupTChan' copies), outbound queues service calls for the
|
|
-- writer, conn holds the current websocket (Nothing before first connect).
|
|
data Bus = Bus
|
|
{ busInbound :: TChan Value
|
|
, busOutbound :: TChan (Request, Service)
|
|
, busConn :: TVar (Maybe Connection)
|
|
, busGen :: CallIdGen
|
|
, busLogEnv :: LogEnv
|
|
}
|
|
|
|
withBus :: Severity -> (Bus -> IO a) -> IO a
|
|
withBus severity callback = do
|
|
handleScribe <- mkHandleScribe ColorIfTerminal stdout (permitItem severity) V2
|
|
let makeLogEnv = registerScribe "stdout" handleScribe defaultScribeSettings =<< initLogEnv "hass-controller" "production"
|
|
-- closeScribes will stop accepting new logs, flush existing ones and clean up resources
|
|
bracket makeLogEnv closeScribes $ \le -> do
|
|
bus <- Bus
|
|
<$> newBroadcastTChanIO
|
|
<*> newTChanIO
|
|
<*> newTVarIO Nothing
|
|
<*> mkCallIdGen 0
|
|
<*> pure le
|
|
callback bus
|
|
|
|
channelHassEval :: (MonadIO m, KatipContext m) => Bus -> HASSEff a -> m a
|
|
channelHassEval bus = \case
|
|
CallService req svc -> katipAddContext (sl "traceId" (toText (requestTraceId req))) $ do
|
|
logFM DebugS (ls $ show svc)
|
|
liftIO $ atomically $ writeTChan (busOutbound bus) (req, svc)
|
|
Debug x -> logFM DebugS (ls $ show x)
|
|
Trace req x -> katipAddContext (sl "traceId" (toText (requestTraceId req))) $
|
|
logFM InfoS (ls $ show x)
|
|
|
|
newtype CallIdGen = CallIdGen { generateCallId :: IO Int }
|
|
|
|
mkCallIdGen :: Int -> IO CallIdGen
|
|
mkCallIdGen start = do
|
|
gen <- newIORef start
|
|
pure $ CallIdGen $ atomicModifyIORef' gen (\old -> let new = old + 1 in new `seq` (new, new))
|