Supervise workers and make auth failures fatal
This commit is contained in:
@@ -14,15 +14,16 @@ module HomeAssistant.Runtime
|
|||||||
) where
|
) where
|
||||||
|
|
||||||
import AFRP (Event (..), Mealy (..))
|
import AFRP (Event (..), Mealy (..))
|
||||||
import Control.Concurrent.Async (mapConcurrently_)
|
import Control.Concurrent.Async (async, waitAny)
|
||||||
import Control.Concurrent.STM (atomically, dupTChan, readTChan)
|
import Control.Concurrent.STM (atomically, dupTChan, readTChan)
|
||||||
import Data.Aeson (Value)
|
import Data.Aeson (Value)
|
||||||
import qualified Data.Text as T
|
import qualified Data.Text as T
|
||||||
import Data.Time (getCurrentTime)
|
import Data.Time (getCurrentTime)
|
||||||
import Data.Void (Void)
|
import Data.Void (Void, absurd)
|
||||||
import HomeAssistant.Controller (HASS, HASSEff (..), lightController)
|
import HomeAssistant.Controller (HASS, HASSEff (..), lightController)
|
||||||
import HomeAssistant.Runtime.Bus
|
import HomeAssistant.Runtime.Bus
|
||||||
import HomeAssistant.Runtime.Connection (readerAction, writerAction)
|
import HomeAssistant.Runtime.Connection (readerAction, writerAction)
|
||||||
|
import HomeAssistant.Runtime.Supervisor (defaultBackoff, supervised)
|
||||||
import Network.Socket (withSocketsDo)
|
import Network.Socket (withSocketsDo)
|
||||||
import System.Environment (getEnv)
|
import System.Environment (getEnv)
|
||||||
|
|
||||||
@@ -53,10 +54,13 @@ defaultMain :: IO ()
|
|||||||
defaultMain = withSocketsDo $ do
|
defaultMain = withSocketsDo $ do
|
||||||
token <- getEnv "HA_TOKEN"
|
token <- getEnv "HA_TOKEN"
|
||||||
bus <- newBus 0
|
bus <- newBus 0
|
||||||
mapConcurrently_ id $
|
let workers =
|
||||||
[ readerAction "last-resort-redux" 8123 token bus
|
[ ("reader", readerAction "last-resort-redux" 8123 token bus)
|
||||||
, writerAction bus
|
, ("writer", writerAction bus)
|
||||||
] ++ map (runController bus) controllers
|
] ++ [ (name, runController bus c) | c@(Controller name _) <- controllers ]
|
||||||
|
as <- mapM (\(name, act) -> async (supervised name defaultBackoff act)) workers
|
||||||
|
(_, v) <- waitAny as
|
||||||
|
absurd v
|
||||||
|
|
||||||
dryRunHassEval :: CallIdGen -> HASSEff a -> IO a
|
dryRunHassEval :: CallIdGen -> HASSEff a -> IO a
|
||||||
dryRunHassEval gen = \case
|
dryRunHassEval gen = \case
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ import Control.Concurrent.STM
|
|||||||
, writeTChan
|
, writeTChan
|
||||||
, writeTVar
|
, writeTVar
|
||||||
)
|
)
|
||||||
|
import Control.Exception.Annotated (throw)
|
||||||
import Control.Lens ((^?))
|
import Control.Lens ((^?))
|
||||||
import Control.Monad (forever)
|
import Control.Monad (forever)
|
||||||
import Data.Aeson (Value, eitherDecode, encode, object, (.=))
|
import Data.Aeson (Value, eitherDecode, encode, object, (.=))
|
||||||
@@ -24,6 +25,7 @@ import qualified Data.Text as T
|
|||||||
import Data.Void (Void)
|
import Data.Void (Void)
|
||||||
import HomeAssistant.Controller (Service (..))
|
import HomeAssistant.Controller (Service (..))
|
||||||
import HomeAssistant.Runtime.Bus
|
import HomeAssistant.Runtime.Bus
|
||||||
|
import HomeAssistant.Runtime.Supervisor (Fatal (..))
|
||||||
import qualified Network.WebSockets as WS
|
import qualified Network.WebSockets as WS
|
||||||
|
|
||||||
-- | Connect, authenticate, subscribe, then receive and broadcast forever.
|
-- | Connect, authenticate, subscribe, then receive and broadcast forever.
|
||||||
@@ -53,7 +55,7 @@ expectType :: T.Text -> Value -> IO ()
|
|||||||
expectType expected msg =
|
expectType expected msg =
|
||||||
case msg ^? key "type" . _String of
|
case msg ^? key "type" . _String of
|
||||||
Just t | t == expected -> pure ()
|
Just t | t == expected -> pure ()
|
||||||
_ -> fail $ "expected " <> T.unpack expected <> ", got: " <> show msg
|
_ -> throw (Fatal $ "expected " <> expected <> ", got: " <> T.pack (show msg))
|
||||||
|
|
||||||
subscribe :: Bus -> WS.Connection -> IO ()
|
subscribe :: Bus -> WS.Connection -> IO ()
|
||||||
subscribe bus conn = do
|
subscribe bus conn = do
|
||||||
@@ -77,7 +79,7 @@ receiveJSON :: WS.Connection -> IO Value
|
|||||||
receiveJSON conn = do
|
receiveJSON conn = do
|
||||||
msg <- WS.receiveData conn
|
msg <- WS.receiveData conn
|
||||||
case eitherDecode msg of
|
case eitherDecode msg of
|
||||||
Left err -> fail $ "Invalid JSON from Home Assistant: " ++ err
|
Left err -> throw (Fatal $ "Invalid JSON from Home Assistant: " <> T.pack err)
|
||||||
Right x -> pure x
|
Right x -> pure x
|
||||||
|
|
||||||
writerAction :: Bus -> IO Void
|
writerAction :: Bus -> IO Void
|
||||||
|
|||||||
Reference in New Issue
Block a user