4 files changed, 174 insertions, 0 deletions
diff --git a/hsm-stream/Hsm/Stream.hs b/hsm-stream/Hsm/Stream.hs
new file mode 100644
index 0000000..e0b2b5b
--- /dev/null
+++ b/hsm-stream/Hsm/Stream.hs
@@ -0,0 +1,69 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE TypeFamilies #-}
+
+module Hsm.Stream
+  ( Stream
+  , startStream
+  , stopStream
+  , runStream
+  )
+where
+
+import Control.Monad (void, when)
+import Effectful (Dispatch (Static), DispatchOf, Eff, IOE, liftIO, (:>))
+import Effectful.Dispatch.Static (SideEffects (WithSideEffects), StaticRep, evalStaticRep, getStaticRep, unsafeEff_)
+import Effectful.Exception (finally)
+import Foreign.C.String (withCString)
+import Foreign.Ptr (Ptr, nullPtr)
+import Hsm.Log (Log, Severity (Info), logMsg)
+import Hsm.Stream.FFI
+  ( GstElement
+  , gstDeinit
+  , gstElementSetState
+  , gstInit
+  , gstObjectUnref
+  , gstParseLaunch
+  , gstStateNull
+  , gstStatePlaying
+  )
+import System.Environment (setEnv)
+
+data Stream (a :: * -> *) (b :: *)
+
+type instance DispatchOf Stream = Static WithSideEffects
+
+newtype instance StaticRep Stream
+  = Stream (Ptr GstElement)
+
+startStream :: (Log "stream" :> es, Stream :> es) => Eff es ()
+startStream = do
+  Stream pipeline <- getStaticRep
+  logMsg Info "Starting stream"
+  unsafeEff_ . void $ gstElementSetState pipeline gstStatePlaying
+
+stopStream :: (Log "stream" :> es, Stream :> es) => Eff es ()
+stopStream = do
+  Stream pipeline <- getStaticRep
+  logMsg Info "Stopping stream"
+  unsafeEff_ . void $ gstElementSetState pipeline gstStateNull
+
+runStream :: (IOE :> es, Log "stream" :> es) => Bool -> Eff (Stream : es) a -> Eff es a
+runStream suppressXLogs action = do
+  when suppressXLogs $ do
+    logMsg Info "Suppressing external loggers"
+    liftIO $ setEnv "GST_DEBUG" "none"
+    liftIO $ setEnv "LIBCAMERA_LOG_LEVELS" "FATAL"
+    liftIO $ setEnv "WEBRTCSINK_SIGNALLING_SERVER_LOG" "none"
+  logMsg Info "Initializing gstreamer library"
+  liftIO $ gstInit nullPtr nullPtr
+  logMsg Info $ "Parsing gstreamer pipeline: " <> pipelineStr
+  pipeline <- liftIO . withCString pipelineStr $ \cStr -> gstParseLaunch cStr nullPtr
+  evalStaticRep (Stream pipeline) . finally action $ stopStream >> endStream
+  where
+    pipelineStr = "libcamerasrc ! videoconvert ! vp8enc deadline=1 ! webrtcsink run-signalling-server=true"
+    endStream = do
+      Stream pipeline <- getStaticRep
+      logMsg Info "Unrefing gstreamer pipeline"
+      liftIO $ gstObjectUnref pipeline
+      logMsg Info "De-initializing gstreamer library"
+      liftIO gstDeinit
diff --git a/hsm-stream/Hsm/Stream/FFI.hsc b/hsm-stream/Hsm/Stream/FFI.hsc
new file mode 100644
index 0000000..3ef4f98
--- /dev/null
+++ b/hsm-stream/Hsm/Stream/FFI.hsc
@@ -0,0 +1,48 @@
+{-# LANGUAGE CApiFFI #-}
+
+module Hsm.Stream.FFI
+  ( GstElement
+  , gstInit
+  , gstDeinit
+  , gstParseLaunch
+  , gstStatePlaying
+  , gstStateNull
+  , gstElementSetState
+  , gstObjectUnref
+  )
+where
+
+import Foreign.C.String (CString)
+import Foreign.C.Types (CChar, CInt)
+import Foreign.Ptr (Ptr)
+
+data GstElement
+
+data GError
+
+newtype GStateChangeReturn
+  = GStateChangeReturn Int
+
+newtype GState
+  = GState Int
+
+foreign import capi safe "gst/gst.h gst_init"
+  gstInit :: Ptr CInt -> Ptr (Ptr (Ptr CChar)) -> IO ()
+
+foreign import capi safe "gst/gst.h gst_deinit"
+  gstDeinit :: IO ()
+
+foreign import capi safe "gst/gst.h gst_parse_launch"
+  gstParseLaunch :: CString -> Ptr GError -> IO (Ptr GstElement)
+
+foreign import capi safe "gst/gst.h value GST_STATE_PLAYING"
+  gstStatePlaying :: GState
+
+foreign import capi safe "gst/gst.h value GST_STATE_NULL"
+  gstStateNull :: GState
+
+foreign import capi safe "gst/gst.h gst_element_set_state"
+  gstElementSetState :: Ptr GstElement -> GState -> IO GStateChangeReturn
+
+foreign import capi safe "gst/gst.h gst_object_unref"
+  gstObjectUnref :: Ptr GstElement -> IO ()
diff --git a/hsm-stream/Test/Stream.hs b/hsm-stream/Test/Stream.hs
new file mode 100644
index 0000000..010ebcc
--- /dev/null
+++ b/hsm-stream/Test/Stream.hs
@@ -0,0 +1,8 @@
+import Control.Concurrent (threadDelay)
+import Data.Function ((&))
+import Effectful (liftIO, runEff)
+import Hsm.Log (Severity (Info), runLog)
+import Hsm.Stream (runStream, startStream)
+
+main :: IO ()
+main = (startStream >> liftIO (threadDelay $ maxBound @Int)) & runStream True & runLog @"stream" Info & runEff
diff --git a/hsm-stream/hsm-stream.cabal b/hsm-stream/hsm-stream.cabal
new file mode 100644
index 0000000..96bca1d
--- /dev/null
+++ b/hsm-stream/hsm-stream.cabal
@@ -0,0 +1,49 @@
+cabal-version: 3.8
+author:        Paul Oliver <contact@pauloliver.dev>
+name:          hsm-stream
+version:       0.1.0.0
+
+library
+  build-depends:
+    , base
+    , effectful-core
+    , effectful-plugin
+    , hsm-log
+
+  default-language: GHC2024
+  exposed-modules:  Hsm.Stream
+  extra-libraries:  gstreamer-1.0
+  ghc-options:
+    -O2 -Wall -Werror -Wno-star-is-type -Wunused-packages
+    -fplugin=Effectful.Plugin
+
+  include-dirs:
+    /usr/include/gstreamer-1.0 /usr/include/glib-2.0
+    /usr/lib/glib-2.0/include
+
+  other-modules:    Hsm.Stream.FFI
+
+executable test-stream
+  build-depends:
+    , base
+    , effectful-core
+    , effectful-plugin
+    , hsm-log
+
+  default-language: GHC2024
+  extra-libraries:  gstreamer-1.0
+  ghc-options:
+    -O2 -threaded -Wall -Werror -Wno-star-is-type -Wunused-packages
+    -fplugin=Effectful.Plugin
+
+  if !arch(x86_64)
+    ghc-options: -optl=-mno-fix-cortex-a53-835769
+
+  include-dirs:
+    /usr/include/gstreamer-1.0 /usr/include/glib-2.0
+    /usr/lib/glib-2.0/include
+
+  main-is:          Test/Stream.hs
+  other-modules:
+    Hsm.Stream
+    Hsm.Stream.FFI