postgrest-ws 0.3.1.0 → 0.3.2.0

raw patch · 6 files changed

+44/−14 lines, 6 filesdep +retryPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: retry

API changes (from Hackage documentation)

Files

• app/Main.hs +1/−3
• postgrest-ws.cabal +2/−1
• src/PostgRESTWS.hs +2/−1
• src/PostgRESTWS/Broadcast.hs +5/−0
• src/PostgRESTWS/HasqlBroadcast.hs +31/−6
• test/HasqlBroadcastSpec.hs +3/−3

@@ -8,7 +8,6 @@                                                        prettyVersion,                                                        readOptions) -import           Data.Function                        (id) import           Data.ByteString.Base64               (decode) import           Data.String                          (IsString (..)) import           Data.Text                            (stripPrefix, pack, replace)@@ -61,8 +60,7 @@   getTime <- mkAutoUpdate     defaultUpdateSettings { updateAction = getPOSIXTime } -  multiOrError <- newHasqlBroadcasterOrError pgSettings-  let multi = either (panic . show) id multiOrError+  multi <- newHasqlBroadcaster pgSettings    runSettings appSettings $     postgrestWsMiddleware (configJwtSecret conf) getTime pool multi $

@@ -1,5 +1,5 @@ name:                postgrest-ws-version:             0.3.1.0+version:             0.3.2.0 synopsis:            PostgREST extension to map LISTEN/NOTIFY messages to Websockets description:         Please see README.md homepage:            https://github.com/diogob/postgrest-ws#readme@@ -44,6 +44,7 @@                      , either                      , stm-containers                      , stm+                     , retry   default-language:    Haskell2010   default-extensions: OverloadedStrings, NoImplicitPrelude 

@@ -6,6 +6,7 @@ module PostgRESTWS   ( postgrestWsMiddleware   -- * Re-exports+  , newHasqlBroadcaster   , newHasqlBroadcasterOrError   ) where @@ -25,7 +26,7 @@ import PostgRESTWS.Claims import PostgRESTWS.Database import PostgRESTWS.Broadcast (Multiplexer, onMessage, readTChan)-import PostgRESTWS.HasqlBroadcast (newHasqlBroadcasterOrError)+import PostgRESTWS.HasqlBroadcast (newHasqlBroadcaster, newHasqlBroadcasterOrError) import qualified PostgRESTWS.Broadcast as B  data Message = Message

@@ -25,6 +25,8 @@ import Control.Concurrent.STM.TChan import Control.Concurrent.STM.TQueue +import GHC.Show+ data SourceCommands = Open ByteString | Close ByteString deriving (Show) data Message = Message { channel :: ByteString                , payload :: ByteString@@ -35,6 +37,9 @@                                , commands :: TQueue SourceCommands                                , messages :: TQueue Message                                }++instance Show Multiplexer where+  show Multiplexer{} = "Multiplexer"  data Channel = Channel { broadcast :: TChan Message                        , listeners :: Integer

@@ -15,20 +15,44 @@  import Hasql.Connection import Data.Either.Combinators (mapBoth)+import System.IO               (hPutStrLn)+import Data.Function           (id)+import Control.Retry           (RetryStatus, retrying, capDelay, exponentialBackoff)  import PostgRESTWS.Database import PostgRESTWS.Broadcast -{- | Returns a multiplexer from a connection URI or an error message on the left case+{- | Returns a multiplexer from a connection URI, keeps trying to connect in case there is any error.    This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners -}+newHasqlBroadcaster :: ByteString -> IO Multiplexer+newHasqlBroadcaster = newHasqlBroadcasterForConnection . tryUntilConnected +{- | Returns a multiplexer from a connection URI or an error message on the left case+   This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners+-} newHasqlBroadcasterOrError :: ByteString -> IO (Either ByteString Multiplexer) newHasqlBroadcasterOrError =-  acquire >=> (sequence . mapBoth show newHasqlBroadcaster)+  acquire >=> (sequence . mapBoth show (newHasqlBroadcasterForConnection . return)) -{- | Returns a multiplexer from a connection, listen for different database notification channels using that connection.+tryUntilConnected :: ByteString -> IO Connection+tryUntilConnected =+  fmap (either (panic "Failure on connection retry") id) . retryConnection+  where+    retryConnection conStr = retrying retryPolicy shouldRetry (const $ acquire conStr)+    maxDelayInMicroseconds = 32000000+    firstDelayInMicroseconds = 1000000+    retryPolicy = capDelay maxDelayInMicroseconds $ exponentialBackoff firstDelayInMicroseconds+    shouldRetry :: RetryStatus -> Either ConnectionError Connection -> IO Bool+    shouldRetry _ con =+      case con of+        Left err ->+          hPutStrLn stderr ("Error connecting notification listener to database: " <> show err)+          >> return True+        _ -> return False +{- | Returns a multiplexer from an IO Connection, listen for different database notification channels using the connection produced.+    This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners     To listen on channels *chat*@@ -49,10 +73,11 @@    @  -}-newHasqlBroadcaster :: Connection -> IO Multiplexer-newHasqlBroadcaster con = do+newHasqlBroadcasterForConnection :: IO Connection -> IO Multiplexer+newHasqlBroadcasterForConnection getCon = do   multi <-     newMultiplexer (\cmds msgs-> do+    con <- getCon     waitForNotifications       (\c m-> atomically $ writeTQueue msgs $ Message c m)       con@@ -61,6 +86,6 @@       case cmd of         Open ch -> listen con ch         Close ch -> unlisten con ch-    ) (\_ -> return ())+    ) (\_ -> hPutStrLn stderr "Broadcaster is dead")   void $ relayMessagesForever multi   return multi

@@ -22,10 +22,10 @@         newConnection connStr =             either (panic . show) id             <$> acquire connStr-            +     it "start listening on a database connection as we send an Open command" $ do       con <- newConnection "postgres://localhost/postgrest_test"-      multi <- liftIO $ newHasqlBroadcaster con+      multi <- liftIO $ newHasqlBroadcaster "postgres://localhost/postgrest_test"        atomically $ openChannelProducer multi "test" @@ -36,7 +36,7 @@      it "stops listening on a database connection as we send a Close command" $ do       con <- newConnection "postgres://localhost/postgrest_test"-      multi <- liftIO $ newHasqlBroadcaster con+      multi <- liftIO $ newHasqlBroadcaster "postgres://localhost/postgrest_test"        atomically $ closeChannelProducer multi "test"