module BrokerTest (test) where

import Control.Exception (throw)
import Data.List (sort)
import RIO
import RIO.ProcessPool.Broker
import Test.Tasty
import Test.Tasty.HUnit
import UnliftIO.MessageBox.Class
import UnliftIO.MessageBox.Unlimited
import Utils
  ( MockBox (MkMockBox),
    MockBoxInit (MkMockBoxInit),
    NoOpArg (..),
    NoOpBox,
    NoOpInput (..),
    runTestApp,
  )

noBrokerConfig :: BrokerConfig k w' w a m
noBrokerConfig =
  MkBrokerConfig
    { demultiplexer = const $ error "unexpected invokation: demultiplexer",
      messageDispatcher = const $ error "unexpected invokation: messageDispatcher",
      resourceCreator = const $ error "unexpected invokation: resourceCreator",
      resourceCleaner = const $ error "unexpected invokation: resourceCleaner"
    }

expectedException :: StringException
expectedException = stringException "Test"

test :: HasCallStack => TestTree
test =
  testGroup
    "BrokerTests"
    [ testCase
        "the show instance of BrokerResult makes sense"
        (assertEqual "" "MkBrokerResult" (show MkBrokerResult)),
      testCase
        "when a broker message box creation throws an exception, \
        \ the exception is returned in a Left..."
        $ do
          Just (Left a) <-
            runTestApp $
              timeout 1000000 $
                spawnBroker @_ @Int @() @() @NoOpArg
                  ( MkMockBoxInit @NoOpBox
                      (throwIO expectedException)
                      Nothing
                  )
                  noBrokerConfig
          assertEqual
            "exception expected"
            (show (SomeException expectedException))
            (show a),
      testCase
        "when a broker message box input creation throws an exception,\
        \ the exception is returned in a Left..."
        $ do
          Just (Left a) <-
            runTestApp $
              timeout 1000000 $
                spawnBroker @_ @Int @() @() @NoOpArg
                  ( MkMockBoxInit
                      ( return
                          ( MkMockBox @NoOpInput
                              (throwIO expectedException)
                              (error "unexpected invokation: receive")
                              (error "unexpected invokation: tryReceive")
                          )
                      )
                      Nothing
                  )
                  noBrokerConfig
          assertEqual
            "exception expected"
            (show (SomeException expectedException))
            (show a),
      testCase
        "when the receive function throws a synchronous exception,\
        \ then waiting on the broker will return the exception"
        $ do
          Just (Right (_, a)) <-
            runTestApp $
              timeout 1000000 $
                spawnBroker @_ @Int @() @() @NoOpArg
                  ( MkMockBoxInit
                      ( return
                          ( MkMockBox
                              ( return
                                  (OnDeliver (error "unexpected invokation: OnDeliver"))
                              )
                              (throwIO expectedException)
                              (error "unexpected invokation: tryReceive")
                          )
                      )
                      Nothing
                  )
                  noBrokerConfig
          r <- runTestApp $ waitCatch a
          assertEqual
            "exception expected"
            ( show
                ( Left (SomeException expectedException) ::
                    Either SomeException ()
                )
            )
            (show r),
      testCase
        "when evaluation of an incoming message causes an exception,\
        \ then the broker ignores the error and continues"
        $ runTestApp $ do
          spRes <-
            spawnBroker
              ( MkMockBoxInit
                  ( return
                      ( MkMockBox
                          ( return
                              (OnDeliver (const (pure True)))
                          )
                          (return (Just (throw expectedException)))
                          (error "unexpected invokation: tryReceive")
                      )
                  )
                  Nothing
              )
              ( MkBrokerConfig
                  { demultiplexer = Dispatch (777 :: Int),
                    messageDispatcher =
                      const (error "unexpected invokation: messageDispatcher"),
                    resourceCreator =
                      const (error "unexpected invokation: resourceCreator"),
                    resourceCleaner =
                      const (error "unexpected invokation: resourceCleaner")
                  }
              )
          case spRes of
            Left err -> error (show err)
            Right (brokerIn, brokerA) -> do
              deliver_ brokerIn ()
              cancel brokerA
              r <- waitCatch brokerA
              liftIO $
                assertEqual
                  "exception expected"
                  "Left AsyncCancelled"
                  (show r),
      testCase
        "when evaluation of the first incoming message causes an async\
        \ exception, then the broker exits with that exception"
        $ runTestApp $ do
          spawnBroker
            ( MkMockBoxInit
                ( return
                    ( MkMockBox
                        ( return
                            (OnDeliver (const (pure True)))
                        )
                        (return (Just (throw (AsyncExceptionWrapper expectedException))))
                        (error "unexpected invokation: tryReceive")
                    )
                )
                Nothing
            )
            ( MkBrokerConfig
                { demultiplexer = Dispatch (777 :: Int),
                  messageDispatcher =
                    const (error "unexpected invokation: messageDispatcher"),
                  resourceCreator =
                    const (error "unexpected invokation: resourceCreator"),
                  resourceCleaner =
                    const (error "unexpected invokation: resourceCleaner")
                }
            )
            >>= \case
              Left err -> error (show err)
              Right (brokerIn, brokerA) -> do
                deliver_ brokerIn ()
                r <- waitCatch brokerA
                liftIO $
                  assertEqual
                    "exception expected"
                    ( show
                        ( Left (SomeException (AsyncExceptionWrapper expectedException)) ::
                            Either SomeException ()
                        )
                    )
                    (show r),
      testCase
        "when a broker is cancelled while waiting for the first message,\
        \ then the broker exits with AsyncCancelled"
        $ runTestApp $ do
          goOn <- newEmptyMVar
          spawnBroker @_ @Int @() @()
            ( MkMockBoxInit
                ( return
                    ( MkMockBox
                        ( return
                            (OnDeliver (const (pure True)))
                        )
                        ( do
                            putMVar goOn ()
                            threadDelay 1_000_000
                            return (Just (error "unexpected evaluation"))
                        )
                        (error "unexpected invokation: tryReceive")
                    )
                )
                Nothing
            )
            noBrokerConfig
            >>= \case
              Left err -> error (show err)
              Right (_brokerIn, brokerA) -> do
                takeMVar goOn
                cancel brokerA
                r <- waitCatch brokerA
                liftIO $
                  assertEqual
                    "exception expected"
                    "Left AsyncCancelled"
                    (show r),
      testCase
        "when a broker receives a message for a missing resource,\
        \ it silently drops the message"
        $ runTestApp $ do
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = Dispatch (777 :: Int),
                    messageDispatcher =
                      const (error "unexpected invokation: messageDispatcher"),
                    resourceCreator =
                      const (error "unexpected invokation: resourceCreator"),
                    resourceCleaner =
                      const (error "unexpected invokation: resourceCleaner")
                  }
          spawnBroker BlockingUnlimited brokerCfg
            >>= \case
              Left err -> error (show err)
              Right (brokerIn, brokerA) -> do
                deliver_ brokerIn ()
                cancel brokerA
                r <- waitCatch brokerA
                liftIO $
                  assertEqual
                    "success expected"
                    "Left AsyncCancelled"
                    (show r),
      testCase
        "when an empty broker receives a start message without payload\
        \ and the creator callback throws an exception,\
        \ a normal message for that key will be ignored,\
        \ no cleanup is performed, and the broker lives on"
        $ runTestApp $ do
          workerInitialized <- newEmptyMVar
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer =
                      \m ->
                        if m
                          then Initialize (777 :: Int) Nothing
                          else Dispatch (777 :: Int) (),
                    messageDispatcher =
                      const (error "unexpected invokation: messageDispatcher"),
                    resourceCreator = \_k _mw -> do
                      putMVar workerInitialized ()
                      throwIO expectedException,
                    resourceCleaner =
                      const (error "unexpected invokation: resourceCleaner")
                  }
          spawnBroker BlockingUnlimited brokerCfg
            >>= \case
              Left err -> error (show err)
              Right (brokerIn, brokerA) -> do
                deliver_ brokerIn True
                timeout 1000000 (takeMVar workerInitialized)
                  >>= liftIO
                    . assertEqual
                      "resourceCreator wasn't executed"
                      (Just ())
                deliver_ brokerIn False
                cancel brokerA
                r <- waitCatch brokerA
                liftIO $
                  assertEqual
                    "exception expected"
                    "Left AsyncCancelled"
                    (show r),
      testCase
        "when an empty broker receives a start message with a payload\
        \ and when the MessageHandler callback throws an exception when\
        \ applied to that payload, cleanup is performed once, and\
        \ incoming messages for that key will be ignored,\
        \ and the broker lives on"
        $ runTestApp $ do
          cleanupCalls <- newEmptyMVar
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer =
                      \isInitPayload ->
                        if isInitPayload
                          then Initialize (777 :: Int) (Just True)
                          else Dispatch (777 :: Int) False,
                    messageDispatcher =
                      \_k isInitPayload _ ->
                        if isInitPayload
                          then throwIO expectedException
                          else error "unexpected invokation: messageDispatcher",
                    resourceCreator = \_k _mw -> do
                      putMVar cleanupCalls (0 :: Int)
                      return (),
                    resourceCleaner =
                      \_k () ->
                        modifyMVar
                          cleanupCalls
                          (\cnt -> return (cnt + 1, ()))
                  }
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn True
          deliver_ brokerIn False
          threadDelay 10_000
          cancel brokerA
          r <- waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "Left AsyncCancelled"
              (show r)
          takeMVar cleanupCalls
            >>= liftIO . assertEqual "resourceCleaner wasn't executed" 1
          -- prevent GC of msg box input:
          deliver_ brokerIn False,
      testCase
        "when 3 resources are initialized and then the broker is cancelled,\
        \ cleanup is performed foreach resource."
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          cleanupCalled <- newTVarIO []
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \k -> Initialize k (Just k),
                    messageDispatcher = \k _w _a -> do
                      putMVar resourceCreated k
                      threadDelay 10_000 -- delay here so we make sure to
                      -- be cancelled before we leave this
                      -- function.
                      -- Now if the implementation isn't
                      -- handling async exceptions well,
                      -- the resource isn't in the resource
                      -- map when cleanup is called,
                      -- and hence won't be properly
                      -- cleaned up!
                      return KeepResource,
                    resourceCreator =
                      \k _mw -> return k,
                    resourceCleaner =
                      \k a ->
                        atomically (modifyTVar cleanupCalled ((k, a) :))
                  }
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          deliver_ brokerIn 2
          deliver_ brokerIn 3
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 1
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 2
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 3
          cancel brokerA
          r <- waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "Left AsyncCancelled"
              (show r)
          readTVarIO cleanupCalled
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                [(1, 1), (2, 2), (3, 3)]
              . sort
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "when 3 resources are initialized and then the broker is cancelled,\
        \ cleanup is performed foreach resource, even if exceptions are thrown\
        \ from the cleanup callbacks"
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          cleanupCalled <- newTVarIO []
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \k -> Initialize k (Just k),
                    messageDispatcher = \k _w _a -> do
                      putMVar resourceCreated k
                      threadDelay 10_000 -- delay here so we make sure to
                      -- be cancelled before we leave this
                      -- function.
                      -- Now if the implementation isn't
                      -- handling async exceptions well,
                      -- the resource isn't in the resource
                      -- map when cleanup is called,
                      -- and hence won't be properly
                      -- cleaned up!
                      return KeepResource,
                    resourceCreator =
                      \k _mw -> return k,
                    resourceCleaner =
                      \k a -> do
                        atomically (modifyTVar cleanupCalled ((k, a) :))
                        throwIO expectedException
                  }
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          deliver_ brokerIn 2
          deliver_ brokerIn 3
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 1
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 2
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 3
          cancel brokerA
          r <- waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "Left AsyncCancelled"
              (show r)
          readTVarIO cleanupCalled
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                [(1, 1), (2, 2), (3, 3)]
              . sort
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "when 2 resources are added and\
        \ while adding a 3rd an async exception is thrown\
        \ when handling the initial message,\
        \ then the broker cleans up the 3 resources and exists"
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          cleanupCalled <- newTVarIO []
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \k -> Initialize k (Just k),
                    messageDispatcher = \k _w _a -> do
                      putMVar resourceCreated k
                      return KeepResource,
                    resourceCreator =
                      \k _mw -> return k,
                    resourceCleaner =
                      \k a -> do
                        atomically (modifyTVar cleanupCalled ((k, a) :))
                  }
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          deliver_ brokerIn 2
          deliver_ brokerIn 3
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 1
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 2
          takeMVar resourceCreated
            >>= liftIO . assertEqual "invalid resource created" 3
          throwTo (asyncThreadId brokerA) expectedException
          r <- either id (error . show) <$> waitCatch brokerA
          readTVarIO cleanupCalled
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                [(1, 1), (2, 2), (3, 3)]
              . sort
          liftIO $
            assertEqual
              "exception expected"
              (show expectedException)
              (show r)
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "when adding a new resource with an extra initial message,\
        \ when the messageHandler returns KeepResource,\
        \ then the resource returned from the create callback is passed to\
        \ the cleanup function"
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          cleanupCalled <- newEmptyTMVarIO
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \_k -> Initialize (777 :: Int) (Just ()),
                    messageDispatcher = \_k _w a -> do
                      putMVar resourceCreated a
                      return KeepResource,
                    resourceCreator =
                      \_k _mw -> return initialResource,
                    resourceCleaner =
                      \_k a -> do
                        atomically (putTMVar cleanupCalled a)
                  }
              initialResource :: Int
              initialResource = 123
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          takeMVar resourceCreated
            >>= liftIO
              . assertEqual "invalid resource created" initialResource
          cancel brokerA
          r <- either id (error . show) <$> waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "AsyncCancelled"
              (show r)
          atomically (takeTMVar cleanupCalled)
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                initialResource
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "when adding a new resource with an extra initial message,\
        \ when the messageHandler returns (UpdateResource x),\
        \ then x is passed to the cleanup function"
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          cleanupCalled <- newEmptyTMVarIO
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),
                    messageDispatcher = \_k _w a -> do
                      void $ async (threadDelay 100000 >> putMVar resourceCreated a)
                      return (UpdateResource x),
                    resourceCreator =
                      \_k _mw -> return initialResource,
                    resourceCleaner =
                      \_k -> atomically . putTMVar cleanupCalled
                  }
              initialResource :: Int
              initialResource = 123
              x :: Int
              x = 234
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          takeMVar resourceCreated
            >>= liftIO
              . assertEqual
                "invalid resource created"
                initialResource
          cancel brokerA
          r <- either id (error . show) <$> waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "AsyncCancelled"
              (show r)
          atomically (takeTMVar cleanupCalled)
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                x
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "when adding a new resource with an extra initial message,\
        \ when the messageHandler returns (RemoveResource Nothing),\
        \ then the initial resource is passed to the cleanup function"
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          readyForCleanup <- newEmptyMVar
          cleanupCalled <- newEmptyTMVarIO
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),
                    messageDispatcher = \_k _w a -> do
                      void . async $ do
                        () <- takeMVar readyForCleanup
                        putMVar resourceCreated a
                      return (RemoveResource Nothing),
                    resourceCreator =
                      \_k _mw -> return initialResource,
                    resourceCleaner =
                      \_k a -> do
                        putMVar readyForCleanup ()
                        atomically (putTMVar cleanupCalled a)
                  }
              initialResource :: Int
              initialResource = 123
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          takeMVar resourceCreated
            >>= liftIO
              . assertEqual
                "invalid resource created"
                initialResource
          cancel brokerA
          r <- either id (error . show) <$> waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "AsyncCancelled"
              (show r)
          atomically (takeTMVar cleanupCalled)
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                initialResource
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "when adding a new resource with an extra initial message,\
        \ when the messageHandler returns (RemoveResource (Just x)),\
        \ then x is passed to the cleanup function"
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          readyForCleanup <- newEmptyMVar
          cleanupCalled <- newEmptyTMVarIO
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),
                    messageDispatcher = \_k _w a -> do
                      void . async $ do
                        () <- takeMVar readyForCleanup
                        putMVar resourceCreated a
                      return (RemoveResource (Just x)),
                    resourceCreator =
                      \_k _mw -> return initialResource,
                    resourceCleaner =
                      \_k a -> do
                        putMVar readyForCleanup ()
                        atomically (putTMVar cleanupCalled a)
                  }
              initialResource :: Int
              initialResource = 123
              x :: Int
              x = 234
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          takeMVar resourceCreated
            >>= liftIO
              . assertEqual
                "invalid resource created"
                initialResource
          cancel brokerA
          r <- either id (error . show) <$> waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "AsyncCancelled"
              (show r)
          atomically (takeTMVar cleanupCalled)
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                x
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "when adding a new resource without an extra initial message,\
        \ then the initial resource is passed to the cleanup function"
        $ runTestApp $ do
          resourceCreated <- newEmptyMVar
          cleanupCalled <- newEmptyTMVarIO
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),
                    messageDispatcher = \_k _w a -> do
                      void $ async (threadDelay 100000 >> putMVar resourceCreated a)
                      return (UpdateResource x),
                    resourceCreator =
                      \_k _mw -> return initialResource,
                    resourceCleaner =
                      \_k -> atomically . putTMVar cleanupCalled
                  }
              initialResource :: Int
              initialResource = 123
              x :: Int
              x = 234
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int)
          takeMVar resourceCreated
            >>= liftIO
              . assertEqual
                "invalid resource created"
                initialResource
          cancel brokerA
          r <- either id (error . show) <$> waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "AsyncCancelled"
              (show r)
          atomically (takeTMVar cleanupCalled)
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                x
          -- prevent GC of msg box input:
          deliver_ brokerIn 666,
      testCase
        "on a broker with two resources a and b with keys 1 and 2,\
        \ for an incoming message with key=1 the MessageDispatcher is\
        \ called with resource a and for key=2 with resource b"
        $ runTestApp $ do
          messageDispatched <- newEmptyMVar
          cleanupCalled <- newTVarIO []
          let brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \(k, isInit) ->
                      if isInit
                        then Initialize k Nothing
                        else Dispatch k k,
                    resourceCreator =
                      \k _mw -> return k,
                    messageDispatcher = \k _w a -> do
                      putMVar messageDispatched (k, a)
                      return KeepResource,
                    resourceCleaner =
                      \k a -> do
                        atomically (modifyTVar cleanupCalled ((k, a) :))
                  }
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn (1 :: Int, True)
          deliver_ brokerIn (2, True)
          deliver_ brokerIn (1, False)
          deliver_ brokerIn (2, False)
          takeMVar messageDispatched
            >>= liftIO
              . assertEqual "invalid resource" (1, 1)
          takeMVar messageDispatched
            >>= liftIO
              . assertEqual "invalid resource" (2, 2)
          throwTo (asyncThreadId brokerA) expectedException
          r <- either id (error . show) <$> waitCatch brokerA
          readTVarIO cleanupCalled
            >>= liftIO
              . assertEqual
                "resourceCleaner wasn't executed"
                [(1, 1), (2, 2)]
              . sort
          liftIO $
            assertEqual
              "exception expected"
              (show expectedException)
              (show r)
          -- prevent GC of msg box input:
          deliver_ brokerIn (666, False),
      testCase
        "When the MessageDispatcher is called with resource x and \
        \ returns (UpdateResource y), then next time it is called with\
        \ resource y"
        $ runTestApp $ do
          messageDispatched <- newEmptyMVar
          let x :: Int
              x = 42
              y = x + 1295
              brokerCfg =
                MkBrokerConfig
                  { demultiplexer = \isInit ->
                      if isInit
                        then Initialize (787 :: Int) Nothing
                        else Dispatch (787 :: Int) (),
                    resourceCreator = \_k _mw ->
                      return x,
                    messageDispatcher = \_k _w a -> do
                      putMVar messageDispatched a
                      return (UpdateResource y),
                    resourceCleaner = \_k _a ->
                      return ()
                  }
          (brokerIn, brokerA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited brokerCfg
          deliver_ brokerIn True
          deliver_ brokerIn False
          deliver_ brokerIn False
          takeMVar messageDispatched
            >>= liftIO . assertEqual "invalid resource" x
          takeMVar messageDispatched
            >>= liftIO . assertEqual "invalid resource" y
          cancel brokerA
          r <- either id (error . show) <$> waitCatch brokerA
          liftIO $
            assertEqual
              "exception expected"
              "AsyncCancelled"
              (show r)
          -- prevent GC of msg box input:
          deliver_ brokerIn False,
      testCase
        "when the receive function throws a synchronous exception x,\
        \ then all resources will be cleaned up and\
        \ the broker will re-throw x"
        $ runTestApp $ do
          cr <- newTVarIO []
          cl <- newTVarIO []
          let bCfg =
                MkBrokerConfig
                  { demultiplexer = (`Initialize` Nothing),
                    resourceCreator = \k _mw -> do
                      atomically $
                        modifyTVar cr (k :)
                      return (),
                    messageDispatcher = \_k _w _a ->
                      return KeepResource,
                    resourceCleaner = \k _a ->
                      atomically $ modifyTVar cl (k :)
                  }
              inter =
                MkInterceptingBoxArg
                  ( MkInterceptor $ do
                      rs <- readTVarIO cr
                      if rs == [3, 2, 1]
                        then throwIO expectedException
                        else return Nothing
                  )
                  (MkInterceptor (return Nothing))
                  BlockingUnlimited
          (b, bA) <-
            either (error . show) id
              <$> spawnBroker inter bCfg
          deliver_ b (1 :: Int)
          deliver_ b (2 :: Int)
          deliver_ b (3 :: Int)
          deliver_ b (4 :: Int)
          ex <- either id (error . show) <$> waitCatch bA
          liftIO $
            assertEqual
              "exception expected"
              (show expectedException)
              (show ex)
          readTVarIO cl
            >>= liftIO
              . assertEqual
                "all resources must be cleaned"
                [3, 2, 1]
          -- prevent GC of msg box input:
          deliver_ b 666,
      testCase
        "when the receive function returns Nothing,\
        \ then all resources will be cleaned up and\
        \ the broker will exit."
        $ runTestApp $ do
          cr <- newTVarIO []
          cl <- newTVarIO []
          let bCfg =
                MkBrokerConfig
                  { demultiplexer = (`Initialize` Nothing),
                    resourceCreator = \k _mw -> do
                      atomically $
                        modifyTVar cr (k :)
                      return (),
                    messageDispatcher = \_k _w _a ->
                      return KeepResource,
                    resourceCleaner = \k _a ->
                      atomically $ modifyTVar cl (k :)
                  }
              inter =
                MkInterceptingBoxArg
                  ( MkInterceptor $ do
                      rs <- readTVarIO cr
                      if rs == [3, 2, 1]
                        then return (Just False)
                        else return Nothing
                  )
                  (MkInterceptor (return Nothing))
                  BlockingUnlimited
          (b, bA) <-
            either (error . show) id
              <$> spawnBroker inter bCfg
          deliver_ b (1 :: Int)
          deliver_ b (2 :: Int)
          deliver_ b (3 :: Int)
          deliver_ b (4 :: Int)
          wait bA
            >>= liftIO
              . assertEqual "exit expected" MkBrokerResult
          readTVarIO cl
            >>= liftIO
              . assertEqual "all resources must be cleaned" [3, 2, 1]
          -- prevent GC of msg box input:
          deliver_ b 666,
      testCase
        "when an initialisation message without payload for an existant resource\
        \ is received, the message is ignored"
        $ runTestApp $ do
          cr <- newTVarIO []
          cl <- newTVarIO []
          let bCfg =
                MkBrokerConfig
                  { demultiplexer = (`Initialize` Nothing),
                    resourceCreator = \k _mw ->
                      atomically $ modifyTVar cr (k :),
                    messageDispatcher = \_k _w _a ->
                      error "unexpected call to messageDispatcher",
                    resourceCleaner = \k _a ->
                      atomically $ modifyTVar cl (k :)
                  }
          (b, bA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited bCfg
          deliver_ b (1 :: Int)
          deliver_ b (1 :: Int)
          deliver_ b (2 :: Int)
          atomically $ do
            crs <- readTVar cr
            checkSTM (crs == [2, 1])
          cancel bA
          waitCatch bA
            >>= liftIO
              . assertEqual
                "exit expected"
                "Left AsyncCancelled"
              . show
          readTVarIO cl
            >>= liftIO
              . assertEqual
                "all resources must be cleaned"
                [2, 1]
          -- prevent GC of msg box input:
          deliver_ b 666,
      testCase
        "when an initialisation message with payload, \
        \ then after the resource creation is done, \
        \ the payload is dispatched to the messageDispatcher"
        $ runTestApp $ do
          done <- newEmptyMVar
          let bCfg =
                MkBrokerConfig
                  { demultiplexer = (`Initialize` Just ()),
                    resourceCreator = \_k mw ->
                      case mw of
                        Just () ->
                          return ()
                        Nothing ->
                          error "now initial message",
                    messageDispatcher = \_k _w _a -> do
                      putMVar done ()
                      return KeepResource,
                    resourceCleaner = \_k _a -> return ()
                  }
          (b, bA) <-
            either (error . show) id
              <$> spawnBroker BlockingUnlimited bCfg
          deliver_ b (1 :: Int)
          deliver_ b (1 :: Int)
          deliver_ b (2 :: Int)
          takeMVar done
          cancel bA
          waitCatch bA
            >>= liftIO
              . assertEqual "exit expected" "Left AsyncCancelled"
              . show
          -- prevent GC of msg box input:
          deliver_ b 666
    ]

newtype Interceptor = MkInterceptor
  { runInterceptor ::
      forall m.
      MonadUnliftIO m =>
      m (Maybe Bool)
  }

data InterceptingBoxArg b
  = MkInterceptingBoxArg
      Interceptor -- receive interceptor
      Interceptor -- deliver interceptor
      b

data InterceptingBox b a
  = MkInterceptingBox
      Interceptor -- receive interceptor
      Interceptor -- deliver interceptor
      (b a)

data InterceptingBoxIn i a
  = MkInterceptingBoxIn
      Interceptor
      (i a)

instance IsMessageBoxArg b => IsMessageBoxArg (InterceptingBoxArg b) where
  type MessageBox (InterceptingBoxArg b) = InterceptingBox (MessageBox b)
  newMessageBox (MkInterceptingBoxArg r i a) =
    MkInterceptingBox r i <$> newMessageBox a
  getConfiguredMessageLimit (MkInterceptingBoxArg _ _ a) =
    getConfiguredMessageLimit a

instance IsMessageBox b => IsMessageBox (InterceptingBox b) where
  type Input (InterceptingBox b) = InterceptingBoxIn (Input b)
  newInput (MkInterceptingBox _ i b) =
    MkInterceptingBoxIn i <$> newInput b
  receive (MkInterceptingBox r _ b) = do
    mi <- runInterceptor r
    case mi of
      Nothing ->
        receive b
      Just True ->
        receive b
      Just False ->
        return Nothing
  tryReceive (MkInterceptingBox r _ b) = do
    -- oh my ... dunno what to do in this case...
    _mi <- runInterceptor r
    tryReceive b

instance IsInput i => IsInput (InterceptingBoxIn i) where
  deliver (MkInterceptingBoxIn interceptIn i) x = do
    mi <- runInterceptor interceptIn
    case mi of
      Nothing ->
        deliver i x
      Just o ->
        return o