Generalized Constraints

[mfine]

The concrete IO types in hedis make it very hard to use with other kinds of Monads, especially in applications using Monad transformers. How do you feel about making more generalized types available? For example:

--- a/src/Database/Redis/Core.hs
+++ b/src/Database/Redis/Core.hs
@@ -1,3 +1,4 @@
+{-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE OverloadedStrings, GeneralizedNewtypeDeriving, RecordWildCards,
     MultiParamTypeClasses, FunctionalDependencies, FlexibleInstances, CPP #-}
 
@@ -15,6 +16,7 @@ import Prelude
 import Control.Applicative
 #endif
 import Control.Monad.Reader
+import Control.Monad.Trans.Control
 import qualified Data.ByteString as B
 import Data.IORef
 import Data.Pool
@@ -62,7 +64,7 @@ instance MonadRedis Redis where
 --  Each call of 'runRedis' takes a network connection from the 'Connection'
 --  pool and runs the given 'Redis' action. Calls to 'runRedis' may thus block
 --  while all connections from the pool are in use.
-runRedis :: Connection -> Redis a -> IO a
+runRedis :: (MonadIO m, MonadBaseControl IO m) => Connection -> Redis a -> m a
 runRedis (Conn pool) redis =
   withResource pool $ \conn -> runRedisInternal conn redis
 
@@ -82,8 +84,8 @@ reRedis r = Redis r
 
 -- |Internal version of 'runRedis' that does not depend on the 'Connection'
 --  abstraction. Used to run the AUTH command when connecting.
-runRedisInternal :: PP.Connection -> Redis a -> IO a
-runRedisInternal conn (Redis redis) = do
+runRedisInternal :: MonadIO m => PP.Connection -> Redis a -> m a
+runRedisInternal conn (Redis redis) = liftIO $ do
   -- Dummy reply in case no request is sent.
   ref <- newIORef (SingleLine "nobody will ever see this")
   r <- runReaderT redis (Env conn ref)

In my particular application, I'm extremely interested in building Conduit Producers/Sources around pubSub and Consumers/Sinks around publish. With the current IO types, I've had to introduce a and thread queue for sourcing from / sinking to redis with hedis instead of a having more general Producers/Sources and Consumers/Sinks:

sinkBus :: MonadIO m => Connection -> TBQueue (ByteString, ByteString) -> m ()
sinkBus conn msgQueue =
  liftIO $ runRedis conn $
    forever $ do
      (channel, message) <- liftIO $ atomically $ readTBQueue msgQueue
      publish channel message

sourceBus :: MonadIO m => Connection -> TBQueue (ByteString, ByteString) -> [ByteString] -> m ()
sourceBus conn msgQueue channels =
  liftIO $ runRedis conn $
    pubSub (psubscribe channels) $ \msg -> do
      atomically $ writeTBQueue msgQueue (msgChannel msg, msgMessage msg)
      return mempty

In particular, having to introduce a thread and queue for reading from / writing to redis is particularly onerous in my application with significant thread counts already :(

/cc @snoyberg in case I'm misunderstanding the limitations here around conduits. I'm happy to help with generalizing the types if that's desirable. Thanks for the great library!

[snoyberg]

runRedis :: (MonadIO m, MonadBaseControl IO m) => Connection -> Redis a -> m a

If the monad only appears in positive position (the result), then MonadBaseControl is unnecessary, and you can just wrap the whole thing in liftIO to get a MonadIO constraint. That's far better than a MonadBaseControl constraint.

I'm not sure about your specific case. Typically a queue like this would be orthogonal to conduit usage, and due to the half-duplex nature of conduit.

[mfine]

Oh nice - that worked out:

 --  Each call of 'runRedis' takes a network connection from the 'Connection'
 --  pool and runs the given 'Redis' action. Calls to 'runRedis' may thus block
 --  while all connections from the pool are in use.
-runRedis :: Connection -> Redis a -> IO a
+runRedis :: MonadIO m => Connection -> Redis a -> m a
 runRedis (Conn pool) redis =
-  withResource pool $ \conn -> runRedisInternal conn redis
+  liftIO $ withResource pool $ \conn -> runRedisInternal conn redis

With the current types I don't have the ability to compose conduit behavior directly from / to the redis instance. Instead of the above sinkBus and sourceBus which take queues, I would love to be able to have:

sourceBus :: MonadIO m => Connection -> [ByteString] -> Source m (ByteString, ByteString)
sourceBus = ???

sinkBus :: MonadIO m => Connection -> Sink (ByteString, ByteString) m ()
sinkBus = ???

And then be able to compose conduit behavior from these. Instead, I've had to use the above described functions where I pass in a queue, and then compose conduits in another thread that drains from the queue.

[snoyberg]

I doubt this change will give you what you want, given that it could have already been done in your code by adding in a liftIO call.

…

On Fri, Feb 17, 2017, 7:39 AM Mark Fine ***@***.***> wrote: Oh nice - that worked out: -- Each call of 'runRedis' takes a network connection from the 'Connection' -- pool and runs the given 'Redis' action. Calls to 'runRedis' may thus block -- while all connections from the pool are in use.-runRedis :: Connection -> Redis a -> IO a +runRedis :: MonadIO m => Connection -> Redis a -> m a runRedis (Conn pool) redis = - withResource pool $ \conn -> runRedisInternal conn redis+ liftIO $ withResource pool $ \conn -> runRedisInternal conn redis With the current types I don't have the ability to compose conduit behavior directly from / to the redis instance. Instead of the above sinkBus and sourceBus which take queues, I would love to be able to have: sourceBus :: MonadIO m => Connection -> [ByteString] -> Source m (ByteString, ByteString) sourceBus = ??? sinkBus :: MonadIO m => Connection -> Sink (ByteString, ByteString) m () sinkBus = ??? And then be able to compose conduit behavior from these. Instead, I've had to use the above described functions where I pass in a queue, and then compose conduits in another thread that drains from the queue. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#87 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AADBB4baTvtINMB5v1NAQ0gZHTrTqqsAks5rdTJ3gaJpZM4MDs3g> .

[mfine]

Right.

I can get the sink to work if I move the runRedis inside of the conduit:

sinkBus :: MonadIO m => Connection -> Sink (ByteString, ByteString) m ()
sinkBus conn =
  CL.mapM_ $ uncurry $ \channel message ->
    liftIO $ runRedis conn $
      void $ publish channel message

Which is ok, though I would have like to have maintained the connection. But for the source, it seems like I'm totally out of luck:

sourceBus :: MonadIO m => Connection -> [ByteString] -> Sink (ByteString, ByteString) m ()
sourceBus conn topics =
  liftIO $ runRedis conn $
    pubSub (psubscribe topics) $ \msg -> do
      yield (msgChannel msg, msgMessage msg)
      return mempty

Which yields me:

Bus.hs:161:7: Couldn't match expected type ‘IO a0’ …
                with actual type ‘ConduitM i0 (ByteString, ByteString) m0 ()’
    In a stmt of a 'do' block: yield (msgChannel msg, msgMessage msg)
    In the expression:
      do { yield (msgChannel msg, msgMessage msg);
           return mempty }
Compilation failed.