diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 $ diff --git a/postgrest-ws.cabal b/postgrest-ws.cabal --- a/postgrest-ws.cabal +++ b/postgrest-ws.cabal @@ -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 diff --git a/src/PostgRESTWS.hs b/src/PostgRESTWS.hs --- a/src/PostgRESTWS.hs +++ b/src/PostgRESTWS.hs @@ -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 diff --git a/src/PostgRESTWS/Broadcast.hs b/src/PostgRESTWS/Broadcast.hs --- a/src/PostgRESTWS/Broadcast.hs +++ b/src/PostgRESTWS/Broadcast.hs @@ -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 diff --git a/src/PostgRESTWS/HasqlBroadcast.hs b/src/PostgRESTWS/HasqlBroadcast.hs --- a/src/PostgRESTWS/HasqlBroadcast.hs +++ b/src/PostgRESTWS/HasqlBroadcast.hs @@ -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 diff --git a/test/HasqlBroadcastSpec.hs b/test/HasqlBroadcastSpec.hs --- a/test/HasqlBroadcastSpec.hs +++ b/test/HasqlBroadcastSpec.hs @@ -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"