Add Runtime.Bus with channel-based effect interpreter

default.nix

@@ -1,5 +1,5 @@
-{ mkDerivation, aeson, base, bytestring, lens, lens-aeson, lib
-, network, text, time, websockets
+{ mkDerivation, aeson, base, bytestring, hspec, lens, lens-aeson
+, lib, network, stm, text, time, websockets
 }:
 mkDerivation {
   pname = "home-assistant-controller";
@@ -8,10 +8,11 @@ mkDerivation {
   isLibrary = true;
   isExecutable = true;
   libraryHaskellDepends = [
-    aeson base bytestring lens lens-aeson network text time websockets
+    aeson base bytestring lens lens-aeson network stm text time
+    websockets
   ];
   executableHaskellDepends = [ base ];
-  testHaskellDepends = [ base ];
+  testHaskellDepends = [ aeson base hspec stm text ];
   license = lib.meta.getLicenseFromSpdxId "BSD-3-Clause";
   mainProgram = "home-assistant-controller";
 }

home-assistant-controller.cabal

@@ -62,6 +62,7 @@ library
     exposed-modules:  AFRP
                     , HomeAssistant.Controller
                     , HomeAssistant.Runtime
+                    , HomeAssistant.Runtime.Bus
 
     -- Modules included in this library but not exported.
     -- other-modules:
@@ -79,6 +80,7 @@ library
                     , network
                     , bytestring
                     , time
+                    , stm
 
     -- Directories containing source files.
     hs-source-dirs:   src
@@ -118,7 +120,7 @@ test-suite home-assistant-controller-test
     default-language: GHC2024
 
     -- Modules included in this executable, other than Main.
-    -- other-modules:
+    other-modules:    BusSpec
 
     -- LANGUAGE extensions used by modules in this package.
     -- other-extensions:
@@ -135,4 +137,8 @@ test-suite home-assistant-controller-test
     -- Test dependencies.
     build-depends:
         base ^>=4.20.2.0,
-        home-assistant-controller
+        home-assistant-controller,
+        hspec,
+        stm,
+        aeson,
+        text

src/HomeAssistant/Controller.hs

@@ -40,7 +40,7 @@ data Service = Service
   , serviceData :: Maybe Value
   , serviceTarget :: T.Text
   }
-  deriving Show
+  deriving (Show, Eq)
 
 data HASSEff a where
   CallService :: Service -> HASSEff ()

src/HomeAssistant/Runtime.hs

@@ -16,6 +16,7 @@ module HomeAssistant.Runtime
 
 import AFRP (Mealy(..), Event(..))
 import HomeAssistant.Controller (HASSEff(..), lightController, Service(..))
+import HomeAssistant.Runtime.Bus (CallIdGen(..), mkCallIdGen)
 import Data.Aeson ((.=), Value (Null), encode, eitherDecode, object)
 import qualified Data.ByteString.Lazy as BL
 import qualified Data.Text as T
@@ -23,7 +24,6 @@ import qualified Network.WebSockets as WS
 import Network.Socket (withSocketsDo)
 import System.Environment (getEnv)
 import Data.Time (getCurrentTime)
-import Data.IORef (newIORef, atomicModifyIORef')
 
 step :: (forall x. eff x -> IO x) -> Mealy eff a b -> a -> IO (b, Mealy eff a b)
 step nt (Mealy f) a = do
@@ -102,13 +102,6 @@ wsCallService conn requestId domain service entityId =
         ]
     ]
 
-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))
-
 hassEval :: CallIdGen -> WS.Connection -> HASSEff a -> IO a
 hassEval gen conn = \case
   CallService x -> do

src/HomeAssistant/Runtime/Bus.hs

@@ -0,0 +1,52 @@
+{-# LANGUAGE LambdaCase #-}
+
+module HomeAssistant.Runtime.Bus
+  ( Bus(..)
+  , CallIdGen(..)
+  , mkCallIdGen
+  , newBus
+  , 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)
+
+-- | Shared runtime state: inbound is a broadcast channel (controllers
+-- read from 'dupTChanIO' 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 Service
+  , busConn     :: TVar (Maybe Connection)
+  , busGen      :: CallIdGen
+  }
+
+newBus :: Int -> IO Bus
+newBus start = Bus
+  <$> newBroadcastTChanIO
+  <*> newTChanIO
+  <*> newTVarIO Nothing
+  <*> mkCallIdGen start
+
+channelHassEval :: Bus -> HASSEff a -> IO a
+channelHassEval bus = \case
+  CallService svc -> atomically $ writeTChan (busOutbound bus) svc
+  Pure a          -> pure a
+
+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))

test/BusSpec.hs

@@ -0,0 +1,43 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module BusSpec (spec) where
+
+import Control.Concurrent.STM
+  ( atomically
+  , dupTChan
+  , readTChan
+  , writeTChan
+  )
+import Data.Aeson (Value (..))
+import HomeAssistant.Controller (HASSEff (..), Service (..))
+import HomeAssistant.Runtime.Bus
+import Test.Hspec
+
+spec :: Spec
+spec = describe "Bus" $ do
+  it "broadcasts inbound messages to every dup'd channel in order" $ do
+    bus <- newBus 0
+    p1 <- atomically $ dupTChan (busInbound bus)
+    p2 <- atomically $ dupTChan (busInbound bus)
+    atomically $ writeTChan (busInbound bus) (Number 1)
+    atomically $ writeTChan (busInbound bus) (Number 2)
+    r1 <- atomically $ (,) <$> readTChan p1 <*> readTChan p1
+    r2 <- atomically $ (,) <$> readTChan p2 <*> readTChan p2
+    r1 `shouldBe` (Number 1, Number 2)
+    r2 `shouldBe` (Number 1, Number 2)
+
+  it "channelHassEval writes CallService to the outbound channel" $ do
+    bus <- newBus 0
+    let svc = Service "light" "turn_on" Nothing "light.bedroom_masse"
+    channelHassEval bus (CallService svc)
+    atomically (readTChan (busOutbound bus)) `shouldReturn` svc
+
+  it "channelHassEval leaves Pure untouched" $ do
+    bus <- newBus 0
+    channelHassEval bus (Pure 42) `shouldReturn` (42 :: Int)
+
+  it "generates unique sequential call ids" $ do
+    gen <- mkCallIdGen 0
+    a <- generateCallId gen
+    b <- generateCallId gen
+    (a, b) `shouldBe` (1, 2)

test/Main.hs

@@ -1,4 +1,7 @@
 module Main (main) where
 
+import Test.Hspec (hspec)
+import qualified BusSpec
+
 main :: IO ()
-main = putStrLn "Test suite not yet implemented."
+main = hspec BusSpec.spec