如何在Haskell中编写事件总线？

Question

我正在努力解决Haskell中的一个设计问题,我似乎无法以优雅和令人满意的方式解决这个问题.我有一个系统,其核心是基于事件源的概念:系统的状态是由一系列事件应用到初始状态产生的.有不同类型的事件,每种类型通过类型系列与系统的特定组件相关:

class Model a where
   data Event a :: * 
   apply :: Event a -> a -> a

instance Model Foo where
   data Event Foo = Foo Int
   ...

instance Model Bar where
   data Event Bar = Bar String
   ...

目前系统是100%同步和耦合的,每个模型都可以访问所有其他模型的事件,这很快变得一团糟,所以我想通过引入事件总线 来解耦事情Bus Events,这样我应该能够编写类似于 dispatch :: Bus Events -> Consumer (Event Foo) -> Bus Events将某些消费者附加Event Foo到Bus Events假设中,在Event Foo和之间存在某种形式的子类型或包含Events.然后我可以通过确保消费者每个都在自己的线程中运行来添加异步性.

从系统的角度来看,这将允许我确保每个组件可独立包装,从而将依赖性限制为所有事件的子集.Eventstype将在整个应用程序级别定义.这个问题看起来与离散时间的FRP类似,但我似乎无法绕过它...

有没有人已经处理过类似的事情,如果有的话,怎么样？

编辑:

我提出了以下代码,这些代码没有用,Source但受到@ Cirdec提议的启发:

import           Control.Applicative
import           Control.Concurrent
import           Control.Concurrent.STM
import           Control.Monad.Reader
import qualified Data.Vector            as V

type Handlers e = V.Vector (Handler e)

data EventBus e = EventBus { handlers    :: Handlers e
                           , eventQueue  :: TChan e
                           , eventThread :: MVar ThreadId
                           }

newBus :: IO (EventBus e)
newBus = do
  chan <- newTChanIO
  var <- newEmptyMVar
  return $ EventBus V.empty chan var

addHandler :: Handler e -> EventBus e -> EventBus e
addHandler h b@EventBus{..} = b { handlers = V.snoc handlers h }

removeHandler :: Int -> EventBus e -> EventBus e
removeHandler idx b@EventBus{..} = b { handlers = let (h,t) = V.splitAt idx handlers
                                                  in h V.++ V.tail t }

startBus :: EventBus e -> IO (EventBus e)
startBus b@EventBus{..} = do
  tid <- forkIO (runBus b)
  putMVar eventThread tid
  return b

runBus :: EventBus e -> IO ()
runBus b@EventBus{..} = do
  _ <- takeMVar eventThread
  forever $ do
    e <- liftIO $ atomically $ readTChan eventQueue
    v <- newTVarIO b
    runReaderT (runEvents $ publish e) v

-- | A monad to handle pub/sub of events of type @e@
newtype Events e a = Events { runEvents :: ReaderT (TVar (EventBus e)) IO a }
                   deriving (Applicative, Functor, Monad, MonadIO, MonadReader (TVar (EventBus e)))

newtype Handler e = Handler { handle :: Events e ()                 -- Unsubscription function
                                     -> Events e (e -> Events e ())  -- what to do with events @e@
                            }


-- | Register a new @Handler e@ within given @Events e@ context
subscribe :: Handler e -> Events e ()
subscribe h = do
  bus <- ask
  liftIO $ atomically $ modifyTVar' bus (addHandler h)

unsubscribe :: Int -> Events e ()
unsubscribe idx = do
  bus <- ask
  liftIO $ atomically $ modifyTVar' bus (removeHandler idx)

publishBus :: EventBus e -> e -> IO ()
publishBus EventBus{..} = atomically . writeTChan eventQueue

publish :: e -> Events e ()
publish event = do
  EventBus{..} <- ask >>= liftIO . atomically . readTVar
  forM_ (zip (V.toList handlers) [0..]) (dispatch event)

dispatch :: e -> (Handler e, Int) -> Events e ()
dispatch event (Handler h, idx) = do
  hdl <- h (unsubscribe idx)
  hdl event

printer :: (Show s) => String -> Handler s
printer prefix = Handler ( \ _ -> return $ \ e -> liftIO (putStrLn $ prefix ++ show e))

Answer 1

可以订阅的携带 s 的事件源a有以下类型

type Source m a = (a -> m ()) -> m (m ())
                   |             |  ^--- how to unsubscribe             
                   |             ^--- how to subscribe
                   ^--- what to do when an `a` happens

事件的使用者或处理程序本质上是接受事件源并订阅它的东西

type Handler m a = (Source m a             ) -> m ()
                 = ((a -> m ()) -> m (m ())) -> m ()
                                                ^-- set up the consumer.

这有点复杂，我们可以反转事物并获得事件处理程序的更好表示：

type Handler m a = m () -> m (a -> m ())
                   |       |  ^-- what to do when an `a` happens
                   |       ^-- set up the consumer
                   ^-- how to unsubscribe

原始事件源使用起来有点棘手；订阅者可能希望取消订阅以响应事件发生，在这种情况下，他们需要递归地将产生的取消订阅操作转化为事件发生时要执行的操作。从更好的 a 定义开始，Handler我们就不存在这个问题了。事件源现在是接受事件处理程序并向其发布的东西。

type Source m a = (Handler m a          ) -> m ()
                = (m () -> m (a -> m ())) -> m ()
                                             ^-- how to subscribe