Add multithreaded Pub/Sub message processing

hedis.cabal

@@ -64,14 +64,17 @@ library
     ghc-prof-options: -auto-all
   exposed-modules:  Database.Redis
   build-depends:    scanner >= 0.2,
+                    async >= 2.1,
                     base >= 4.6 && < 5,
                     bytestring >= 0.9,
                     bytestring-lexing >= 0.5,
+                    unordered-containers,
                     text,
                     deepseq,
                     mtl >= 2,
                     network >= 2,
                     resource-pool >= 0.2,
+                    stm,
                     time,
                     vector >= 0.9
 
@@ -100,12 +103,17 @@ benchmark hedis-benchmark
 
 test-suite hedis-test
     type: exitcode-stdio-1.0
-    main-is: test/Test.hs
+    hs-source-dirs: test
+    main-is: Test.hs
+    other-modules: PubSubTest
     build-depends:
         base == 4.*,
         bytestring >= 0.10,
         hedis,
         HUnit,
+        async,
+        stm,
+        text,
         mtl == 2.*,
         slave-thread,
         test-framework,

src/Database/Redis/Core.hs

@@ -2,7 +2,7 @@
     MultiParamTypeClasses, FunctionalDependencies, FlexibleInstances, CPP #-}
 
 module Database.Redis.Core (
-    Connection, connect,
+    Connection(..), connect,
     ConnectInfo(..), defaultConnectInfo,
     Redis(), runRedis, unRedis, reRedis,
     RedisCtx(..), MonadRedis(..),
@@ -40,7 +40,6 @@ newtype Redis a = Redis (ReaderT RedisEnv IO a)
 
 data RedisEnv = Env { envConn :: PP.Connection, envLastReply :: IORef Reply }
 
-
 -- |This class captures the following behaviour: In a context @m@, a command
 --  will return it's result wrapped in a \"container\" of type @f@.
 --

src/Database/Redis/ProtocolPipelining.hs

@@ -15,7 +15,7 @@
 --
 module Database.Redis.ProtocolPipelining (
   Connection,
-  connect, disconnect, request, send, recv,
+  connect, disconnect, request, send, recv, flush,
   ConnectionLostException(..),
   HostName, PortID(..)
 ) where
@@ -92,6 +92,12 @@ recv Conn{..} = do
   writeIORef connReplies rs
   return r
 
+-- | Flush the socket.  Normally, the socket is flushed in 'recv' (actually 'conGetReplies'), but
+-- for the multithreaded pub/sub code, the sending thread needs to explicitly flush the subscription
+-- change requests.
+flush :: Connection -> IO ()
+flush Conn{..} = hFlush connHandle
+
 -- |Send a request and receive the corresponding reply
 request :: Connection -> S.ByteString -> IO Reply
 request conn req = send conn req >> recv conn