postgres-websockets 0.5.0.2 → 0.6.0.0

raw patch · 6 files changed

+14/−141 lines, 6 filesdep +hasql-notificationsPVP ok

version bump matches the API change (PVP)

Dependencies added: hasql-notifications

API changes (from Hackage documentation)

Files

• postgres-websockets.cabal +4/−4
• src/PostgresWebsockets.hs +4/−4
• src/PostgresWebsockets/Database.hs +0/−98
• src/PostgresWebsockets/HasqlBroadcast.hs +4/−4
• test/DatabaseSpec.hs +0/−27
• test/HasqlBroadcastSpec.hs +2/−4

@@ -1,5 +1,5 @@ name:                postgres-websockets-version:             0.5.0.2+version:             0.6.0.0 synopsis:            Middleware to map LISTEN/NOTIFY messages to Websockets description:         Please see README.md homepage:            https://github.com/diogob/postgres-websockets#readme@@ -8,7 +8,7 @@ author:              Diogo Biazus maintainer:          diogo@biazus.ca copyright:           2016 Diogo Biazus-category:            Web+category:            Web, Database, PostgreSQL build-type:          Simple -- extra-source-files: cabal-version:       >=1.10@@ -18,7 +18,6 @@   ghc-options:         -Wall   exposed-modules:     PostgresWebsockets                      , PostgresWebsockets.Broadcast-                     , PostgresWebsockets.Database                      , PostgresWebsockets.HasqlBroadcast                      , PostgresWebsockets.Claims   build-depends:       base >= 4.7 && < 5@@ -37,6 +36,7 @@                      , aeson >= 0.11                      , protolude >= 0.2                      , hasql >= 0.19+                     , hasql-notifications >= 0.1.0.0 && < 0.2                      , either                      , stm-containers                      , stm@@ -77,7 +77,6 @@   main-is:             Spec.hs   other-modules:       BroadcastSpec                      , ClaimsSpec-                     , DatabaseSpec                      , HasqlBroadcastSpec   build-depends:       base                      , protolude >= 0.2@@ -89,6 +88,7 @@                      , aeson                      , hasql                      , hasql-pool+                     , hasql-notifications >= 0.1.0.0 && < 0.2                      , http-types                      , unordered-containers >= 0.2                      , wai-extra

@@ -11,6 +11,7 @@   ) where  import qualified Hasql.Pool                     as H+import qualified Hasql.Notifications            as H import qualified Network.Wai                    as Wai import qualified Network.Wai.Handler.WebSockets as WS import qualified Network.WebSockets             as WS@@ -25,7 +26,6 @@ import           PostgresWebsockets.Broadcast          (Multiplexer, onMessage) import qualified PostgresWebsockets.Broadcast          as B import           PostgresWebsockets.Claims-import           PostgresWebsockets.Database import           PostgresWebsockets.HasqlBroadcast     (newHasqlBroadcaster,                                                  newHasqlBroadcasterOrError) @@ -38,7 +38,7 @@ instance A.ToJSON Message  -- | Given a secret, a function to fetch the system time, a Hasql Pool and a Multiplexer this will give you a WAI middleware.-postgresWsMiddleware :: ByteString -> ByteString -> H.Pool -> Multiplexer -> Wai.Application -> Wai.Application+postgresWsMiddleware :: Text -> ByteString -> H.Pool -> Multiplexer -> Wai.Application -> Wai.Application postgresWsMiddleware =   WS.websocketsOr WS.defaultConnectionOptions `compose` wsApp   where@@ -48,7 +48,7 @@  -- when the websocket is closed a ConnectionClosed Exception is triggered -- this kills all children and frees resources for us-wsApp :: ByteString -> ByteString -> H.Pool -> Multiplexer -> WS.ServerApp+wsApp :: Text -> ByteString -> H.Pool -> Multiplexer -> WS.ServerApp wsApp dbChannel secret pool multi pendingConn =   validateClaims requestChannel secret (toS jwtToken) >>= either rejectRequest forkSessions   where@@ -74,7 +74,7 @@             onMessage multi ch $ WS.sendTextData conn . B.payload            when (hasWrite mode) $-            let sendNotifications = void . notifyPool pool dbChannel+            let sendNotifications = void . (H.notifyPool pool dbChannel) . toS             in notifySession validClaims (toS ch) conn sendNotifications            waitForever <- newEmptyMVar

@@ -1,98 +0,0 @@-{-| This module encapsulates knowledge about the SQL commands and the Hasql interface.--}-module PostgresWebsockets.Database-  ( notifyPool-  , notify-  , listen-  , unlisten-  , waitForNotifications-  , PgIdentifier-  , toPgIdentifier-  ) where--import Protolude hiding (replace)-import Hasql.Pool (Pool, UsageError, use)-import Hasql.Session (sql, run, statement)-import qualified Hasql.Session as S-import qualified Hasql.Statement as HST-import Hasql.Connection (Connection, withLibPQConnection)-import qualified Hasql.Decoders as HD-import qualified Hasql.Encoders as HE-import qualified Database.PostgreSQL.LibPQ      as PQ-import Data.Either.Combinators-import qualified Data.ByteString.Char8 as B-import Data.ByteString.Search (replace)-import Data.Functor.Contravariant (contramap)--newtype Error = NotifyError Text---- | A wrapped bytestring that represents a properly escaped and quoted PostgreSQL identifier-newtype PgIdentifier = PgIdentifier ByteString deriving (Show)---- | Given a PgIdentifier returns the wrapped bytestring-fromPgIdentifier :: PgIdentifier -> ByteString-fromPgIdentifier (PgIdentifier bs) = bs---- | Given a bytestring returns a properly quoted and escaped PgIdentifier-toPgIdentifier :: ByteString -> PgIdentifier-toPgIdentifier x = PgIdentifier $ "\"" <> strictlyReplaceQuotes (trimNullChars x) <> "\""-  where-    trimNullChars :: ByteString -> ByteString-    trimNullChars = B.takeWhile (/= '\x0')-    strictlyReplaceQuotes :: ByteString -> ByteString-    strictlyReplaceQuotes = toS . replace "\"" ("\"\"" :: ByteString)---- | Given a Hasql Pool, a channel and a message sends a notify command to the database-notifyPool :: Pool -> ByteString -> ByteString -> IO (Either Error ())-notifyPool pool channel mesg =-   mapError <$> use pool (statement (toS channel, toS mesg) callStatement)-   where-     mapError :: Either UsageError () -> Either Error ()-     mapError = mapLeft (NotifyError . show)-     callStatement = HST.Statement ("SELECT pg_notify" <> "($1, $2)") encoder HD.noResult False-     encoder = contramap fst (HE.param $ HE.nonNullable $ HE.text) <> contramap snd (HE.param $ HE.nonNullable $ HE.text)---- | Given a Hasql Connection, a channel and a message sends a notify command to the database-notify :: Connection -> PgIdentifier -> ByteString -> IO (Either Error ())-notify con channel mesg =-   mapError <$> run (sql ("NOTIFY " <> fromPgIdentifier channel <> ", '" <> mesg <> "'")) con-   where-     mapError :: Either S.QueryError () -> Either Error ()-     mapError = mapLeft (NotifyError . show)---- | Given a Hasql Connection and a channel sends a listen command to the database-listen :: Connection -> PgIdentifier -> IO ()-listen con channel =-  void $ withLibPQConnection con execListen-  where-    execListen pqCon = void $ PQ.exec pqCon $ "LISTEN " <> fromPgIdentifier channel---- | Given a Hasql Connection and a channel sends a unlisten command to the database-unlisten :: Connection -> PgIdentifier -> IO ()-unlisten con channel =-  void $ withLibPQConnection con execListen-  where-    execListen pqCon = void $ PQ.exec pqCon $ "UNLISTEN " <> fromPgIdentifier channel---{- | Given a function that handles notifications and a Hasql connection forks a thread that listens on the database connection and calls the handler everytime a message arrives.--   The message handler passed as first argument needs two parameters channel and payload.--}--waitForNotifications :: (ByteString -> ByteString -> IO()) -> Connection -> IO ()-waitForNotifications sendNotification con =-  withLibPQConnection con $ void . forever . pqFetch-  where-    pqFetch pqCon = do-      mNotification <- PQ.notifies pqCon-      case mNotification of-        Nothing -> do-          mfd <- PQ.socket pqCon-          case mfd of-            Nothing  -> panic "Error checking for PostgreSQL notifications"-            Just fd -> do-              void $ threadWaitRead fd-              void $ PQ.consumeInput pqCon-        Just notification ->-           sendNotification (PQ.notifyRelname notification) (PQ.notifyExtra notification)

@@ -14,19 +14,19 @@ import Protolude hiding (putErrLn)  import Hasql.Connection+import Hasql.Notifications import Data.Aeson              (decode, Value(..)) import Data.HashMap.Lazy       (lookupDefault) import Data.Either.Combinators (mapBoth) import Data.Function           (id) import Control.Retry           (RetryStatus, retrying, capDelay, exponentialBackoff) -import PostgresWebsockets.Database import PostgresWebsockets.Broadcast  {- | 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 -> ByteString -> IO Multiplexer+newHasqlBroadcaster :: Text -> ByteString -> IO Multiplexer newHasqlBroadcaster ch = newHasqlBroadcasterForConnection . tryUntilConnected   where     newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel ch@@ -34,7 +34,7 @@ {- | 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 -> ByteString -> IO (Either ByteString Multiplexer)+newHasqlBroadcasterOrError :: Text -> ByteString -> IO (Either ByteString Multiplexer) newHasqlBroadcasterOrError ch =   acquire >=> (sequence . mapBoth show (newHasqlBroadcasterForConnection . return))   where@@ -78,7 +78,7 @@    @  -}-newHasqlBroadcasterForChannel :: ByteString -> IO Connection -> IO Multiplexer+newHasqlBroadcasterForChannel :: Text -> IO Connection -> IO Multiplexer newHasqlBroadcasterForChannel ch getCon = do   multi <- newMultiplexer openProducer closeProducer   void $ relayMessagesForever multi

@@ -1,27 +0,0 @@-module DatabaseSpec (spec) where--import Protolude--import           Data.Function                        (id)-import qualified Hasql.Connection                     as H-import Test.Hspec--import PostgresWebsockets.Database--spec :: Spec-spec =-  describe "waitForNotifications" $-    it "does trigger action upon notification" $ do-      conOrError <- H.acquire "postgres://localhost/postgres_ws_test"-      let con = either (panic . show) id conOrError :: H.Connection-      notification <- liftIO newEmptyMVar--      race_-        (waitForNotifications (curry $ putMVar notification) con)-        (do listen con $ toPgIdentifier "test"--            conOrError2 <- H.acquire "postgres://localhost/postgres_ws_test"-            let con2 = either (panic . show) id conOrError2 :: H.Connection-            void $ notify con2 (toPgIdentifier "test") "hello there"--            readMVar notification `shouldReturn` ("test", "hello there"))

@@ -2,13 +2,11 @@  import Protolude -import           Data.Function                        (id)-+import Data.Function (id) import Test.Hspec- import PostgresWebsockets.Broadcast import PostgresWebsockets.HasqlBroadcast-import PostgresWebsockets.Database+import Hasql.Notifications  spec :: Spec spec = describe "newHasqlBroadcaster" $ do