postgrest-ws 0.1.0.2 → 0.1.0.3

raw patch · 3 files changed

+44/−22 lines, 3 filesdep ~postgrestPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: postgrest

API changes (from Hackage documentation)

Files

• app/Main.hs +2/−2
• postgrest-ws.cabal +2/−2
• src/PostgRESTWS/Broadcast.hs +40/−18

@@ -8,7 +8,7 @@                                                        minimumPgVersion,                                                        prettyVersion,                                                        readOptions)-import           PostgREST.Error                      (prettyUsageError)+import           PostgREST.Error                      (encodeError) import           PostgREST.OpenAPI                    (isMalformedProxyUri) import           PostgREST.DbStructure import           PostgREST.App                        (postgrest)@@ -80,7 +80,7 @@     getDbStructure (toS $ configSchema conf)    forM_ (lefts [result]) $ \e -> do-    hPutStrLn stderr (prettyUsageError e)+    hPutStrLn stderr (toS $ encodeError e)     exitFailure    refDbStructure <- newIORef $ either (panic . show) id result

@@ -1,5 +1,5 @@ name:                postgrest-ws-version:             0.1.0.2+version:             0.1.0.3 synopsis:            PostgREST extension to map LISTEN/NOTIFY messages to Websockets description:         Please see README.md homepage:            https://github.com/diogob/postgrest-ws#readme@@ -57,7 +57,7 @@                      , warp >= 3.2 && < 4                      , unix >= 2.7 && < 3                      , jwt >= 0.7 && < 1-                     , postgrest+                     , postgrest >= 0.4.1.0                      , postgrest-ws                      , postgresql-libpq >= 0.9 && < 1.0                      , protolude >= 0.1.6 && < 0.2

@@ -1,3 +1,13 @@+{-|+Module      : PostgRESTWS.Auth+Description : PostgRESTWS functions to broadcast messages to several listening clients++This module provides a type called Multiplexer.+The multiplexer contains a map of channels and a producer thread.++This module avoids any database implementation details, it is used by HasqlBroadcast where+the database logic is combined.+-} module PostgRESTWS.Broadcast ( Multiplexer (src)                              , Message (..)                              , SourceCommands (..)@@ -7,7 +17,7 @@                              , relayMessagesForever                              , openChannelProducer                              , closeChannelProducer-                             -- reexports+                             -- * Re-exports                              , readTQueue                              , writeTQueue                              , readTChan@@ -34,15 +44,19 @@                        , close :: STM ()                        } +-- | Open a multiplexer's channel openChannelProducer :: Multiplexer -> ByteString -> STM () openChannelProducer multi ch = writeTQueue (commands multi) (Open ch) +-- | Close a multiplexer's channel closeChannelProducer ::  Multiplexer -> ByteString -> STM () closeChannelProducer multi chan = writeTQueue (commands multi) (Close chan) +-- | Opens a thread that relays messages from the producer thread to the channels forever relayMessagesForever :: Multiplexer -> IO ThreadId relayMessagesForever =  forkIO . forever . relayMessages +-- | Reads the messages from the producer and relays them to the active listeners in their respective channels. relayMessages :: Multiplexer -> IO () relayMessages multi =   atomically $ do@@ -72,23 +86,31 @@     openChannelProducer multi chan     return newChannel +{- |  Adds a listener to a certain multiplexer's channel.+      The listener must be a function that takes a 'TChan Message' and perform any IO action.+      All listeners run in their own thread.+      The first listener will open the channel, when a listener dies it will check if there acquire+      any others and close the channel when that's the case.+-} onMessage :: Multiplexer -> ByteString -> (TChan Message -> IO()) -> IO () onMessage multi chan action = do-  listener <- atomically $ do-    mC <- M.lookup chan (channels multi)-    c <- case mC of-              Nothing -> openChannel multi chan-              Just ch -> return ch-    M.delete chan (channels multi)-    let newChannel = Channel{ broadcast = broadcast c, listeners = listeners c + 1, close = close c}-    M.insert newChannel chan (channels multi)-    dupTChan $ broadcast newChannel-  void $ forkFinally (action listener) (\_ -> atomically $ do-    mC <- M.lookup chan (channels multi)-    let c = fromMaybe (panic $ "trying to remove listener from non existing channel: " <> toS chan) mC-    M.delete chan (channels multi)-    when (listeners c - 1 == 0) $ closeChannelProducer multi chan-    when (listeners c - 1 > 0) $ do-      let newChannel = Channel{ broadcast = broadcast c, listeners = listeners c - 1, close = close c}+  listener <- atomically $ openChannelWhenNotFound >>= addListener+  void $ forkFinally (action listener) disposeListener+  where+    disposeListener _ = atomically $ do+      mC <- M.lookup chan (channels multi)+      let c = fromMaybe (panic $ "trying to remove listener from non existing channel: " <> toS chan) mC+      M.delete chan (channels multi)+      if listeners c - 1 > 0+        then M.insert Channel{ broadcast = broadcast c, listeners = listeners c - 1, close = close c} chan (channels multi)+        else closeChannelProducer multi chan+    openChannelWhenNotFound =+      M.lookup chan (channels multi) >>= \mC ->+      case mC of+            Nothing -> openChannel multi chan+            Just ch -> return ch+    addListener ch = do+      M.delete chan (channels multi)+      let newChannel = Channel{ broadcast = broadcast ch, listeners = listeners ch + 1, close = close ch}       M.insert newChannel chan (channels multi)-    )+      dupTChan $ broadcast newChannel