Add concurrent test for worker notifications

Concurrency is a good way to trigger any interleaving bug.

test/Test/API/Worker.hs

@@ -19,7 +19,7 @@ module Test.API.Worker (
     tests
   ) where
 
-import Control.Concurrent.Async (withAsync)
+import Control.Concurrent.Async (withAsync, forConcurrently_)
 import Control.Concurrent.STM.TChan
 import Control.Lens
 import Control.Monad.STM (atomically)
@@ -40,6 +40,7 @@ import Test.Utils.Notifications
 import Gargantext.System.Logging
 import qualified Data.Text.Encoding as TE
 import qualified Data.ByteString as BL
+import Test.Tasty.HUnit (assertBool)
 
 
 
@@ -63,6 +64,30 @@ tests = sequential $ aroundAll withTestDBAndPort $ do
 
         mTimeout `shouldSatisfy` isJust
 
+    describe "concurrency" $ do
+
+      -- This test checks that two concurrent threads can both subscribe
+      -- to the same topic and get notified.
+      it "handles concurrent threads" $ \(SpecContext testEnv port _app _) -> do
+        let cfg = test_config testEnv
+        let log_cfg = (test_config testEnv) ^. gc_logging
+
+        let topic = DT.Ping
+        let competingThreads = 3
+
+        forConcurrently_ [ 1 .. competingThreads ] $ \(tid :: Int) -> do
+          tchan <- newTChanIO :: IO (TChan (Maybe DT.Notification))
+
+          withAsync (setupWsThread log_cfg topic tchan port) $ \_a -> do
+            _ <- sendJobWithCfg cfg Ping
+
+            mTimeout <- Timeout.timeout (5 * 1_000_000) $ do
+              md <- atomically $ readTChan tchan
+
+              md `shouldBe` Just DT.NPing
+
+            assertBool ("Competing Thread " <> show tid <> " didn't receive a value.") (isJust mTimeout)
+
 setupWsThread :: LogConfig -> DT.Topic -> TChan (Maybe DT.Notification) -> Int -> IO ()
 setupWsThread log_cfg topic tchan port = withLogger log_cfg $ \ioL -> do
   withWSConnection ("127.0.0.1", port) $ \conn -> do